API Reference¶

class BaseAgentService(ABC):
    """Abstract base for all agent services.

    Subclass this and implement the abstract properties/methods.
    The transport layer reads these to wire up MCP + A2A automatically.
    """

    # Set by create_app() - available after app is built
    output_dir: Path | None = None
    base_url: str | None = None

    def file_url(self, filename: str) -> str:
        """Return the download URL for a file in the output directory."""
        from urllib.parse import quote

        return f"{self.base_url}/files/{quote(filename)}"

    def resolve_file(
        self,
        source: str,
        default_ext: str = ".bin",
        filename: str = "",
    ) -> Path:
        """Resolve a source string to a local file Path.

        Detects the format first, then resolves:
        1. data URI (data:mime;base64,...) -> decode and write to temp file
        2. raw base64 -> decode and write to temp file
        3. HTTP(S) URL -> fetch and write to temp file
        4. file path -> return as-is if it exists

        If filename is provided, the temp file preserves that name
        (important for downstream tools that infer type from name).
        """
        assert self.output_dir, "output_dir must be set before resolving files"

        # 1. data URI
        if source.startswith("data:"):
            _, _, encoded = source.partition(",")
            return self._write_decoded(encoded, default_ext, filename)

        # 2. HTTP(S) URL
        if source.startswith(("http://", "https://")):
            import httpx as _httpx

            resp = _httpx.get(source, timeout=30)
            resp.raise_for_status()
            ext = Path(source.rsplit("/", 1)[-1]).suffix or default_ext
            fn = filename or f"_fetch_{uuid.uuid4().hex[:8]}{ext}"
            tmp = self.output_dir / fn
            tmp.write_bytes(resp.content)
            return tmp

        # 3. Existing file path (only try for short strings that look like paths)
        if len(source) < 1024:
            p = Path(source)
            if p.exists():
                return p
            # Try relative to output_dir
            p2 = self.output_dir / source
            if p2.exists():
                return p2

        # 4. Raw base64 (no data: prefix)
        try:
            data = base64.b64decode(source, validate=True)
            if len(data) > 4:
                return self._write_bytes(data, default_ext, filename)
        except Exception:
            pass

        # Fallback: return as path (may not exist - tool will error)
        return Path(source)

    def _write_decoded(
        self,
        encoded: str,
        default_ext: str,
        filename: str,
    ) -> Path:
        """Decode base64 and write to a temp file."""
        data = base64.b64decode(encoded)
        return self._write_bytes(data, default_ext, filename)

    def _write_bytes(
        self,
        data: bytes,
        default_ext: str,
        filename: str,
    ) -> Path:
        """Write bytes to a temp file, preserving filename if given."""
        if filename:
            subdir = self.output_dir / f"_att_{uuid.uuid4().hex[:8]}"
            subdir.mkdir(exist_ok=True)
            tmp = subdir / filename
            # Create parent dirs for nested names (e.g. "images/cat.png")
            tmp.parent.mkdir(parents=True, exist_ok=True)
        else:
            tmp = self.output_dir / f"_upload_{uuid.uuid4().hex[:8]}{default_ext}"
        tmp.write_bytes(data)
        return tmp

    def resolve_file_attachment(
        self,
        source: FileAttachment | str,
        default_ext: str = ".bin",
    ) -> Path:
        """Resolve a FileAttachment dict or plain string to a local Path."""
        if isinstance(source, dict):
            name = source.get("name", "")
            content = source.get("content", name)
            ext = Path(name).suffix if name else default_ext
            return self.resolve_file(content, default_ext=ext, filename=name)
        return self.resolve_file(source, default_ext=default_ext)

    @property
    @abstractmethod
    def agent_name(self) -> str:
        """Human-readable agent name (e.g. 'Email Agent')."""

    @property
    @abstractmethod
    def agent_description(self) -> str:
        """Short description of what this agent does."""

    @property
    def agent_version(self) -> str:
        return "0.1.0"

    @property
    def agent_system_prompt(self) -> str:
        """Agent-specific system prompt, appended to LLMExecutor's base prompt.

        Default: current date/time. Override in subclasses:
        - Extend: return super().agent_system_prompt + "\\n\\nExtra instructions.".
        - Replace: return "Custom prompt." (omits date/time).
        """
        from datetime import datetime

        now = datetime.now().astimezone()
        tz = now.strftime("%Z") or str(now.tzinfo)
        return now.strftime(f"Current date/time: %Y-%m-%d %H:%M (%A) {tz}.")

    # Optional LLM router for A2A natural language dispatch.
    # Set ROUTER_LLM_MODEL, ROUTER_LLM_API_KEY, ROUTER_LLM_API_BASE
    # in .env to enable. Override in subclass for custom logic.
    @property
    def router_llm_model(self) -> str:
        return os.environ.get("ROUTER_LLM_MODEL", "")

    @property
    def router_llm_api_key(self) -> str:
        return os.environ.get("ROUTER_LLM_API_KEY", "")

    @property
    def router_llm_api_base(self) -> str:
        return os.environ.get("ROUTER_LLM_API_BASE", "")

    @abstractmethod
    def get_tools(self) -> list[AgentTool]:
        """Return all MCP tools this agent exposes."""

    @abstractmethod
    def get_skills(self) -> list[SkillDef]:
        """Return all A2A skills this agent exposes."""

    @abstractmethod
    async def execute_skill(
        self,
        skill_id: str,
        message: str,
        *,
        task_id: str | None = None,
    ) -> str:
        """Execute an A2A skill by ID with the given message."""

class LLMExecutor:
    """Multi-step LLM tool-calling loop.

    Supports Anthropic (native + Azure AI Foundry) and OpenAI-compatible APIs.
    Injects 5 synthetic tools for task lifecycle control.
    """

    def __init__(
        self,
        tools: list[AgentTool],
        model: str,
        api_key: str,
        api_base: str,
        system_prompt: str = "",
        max_steps: int = 10,
        on_progress: Any | None = None,
        max_tokens: int = 32000,
    ):
        self.tools = tools
        self.model = model
        self.api_key = api_key
        self.api_base = api_base
        base = self._synthetic_tools_prompt()
        self.system_prompt = f"{base}\n\n{system_prompt}" if system_prompt else base
        self.max_steps = max_steps
        # Per-turn output budget. Higher = the model can emit a full document
        # in one write_file call instead of fragmenting into many small files.
        self.max_tokens = max_tokens
        self.on_progress = on_progress  # async callback(message: str)
        self._provider = _detect_provider(api_base)
        self._tool_map = {t.name: t for t in tools}
        self._produced_files: list[dict] = []

    @staticmethod
    def _synthetic_tools_prompt() -> str:
        """Instructions for synthetic task-lifecycle tools. Always included."""
        return (
            "You MUST use the provided tools to "
            "complete tasks. Do NOT describe what you would do - execute "
            "it by calling the appropriate tool. After getting a tool "
            "result, either call another tool or provide your final "
            "answer as plain text. Use _request_input if you need "
            "clarification. Use _reject_task if the task is outside "
            "your capabilities. Use _return_result to provide your "
            "final answer with specific files."
        )

    async def run(
        self,
        message: str,
        files: list[dict] | None = None,
        history: list[dict] | None = None,
    ) -> ExecutorResult:
        """Run the agentic loop.

        Args:
            message: User/requester message text.
            files: FileAttachment dicts [{name, content}] for inline files.
            history: Prior conversation messages for continuation.

        Returns:
            ExecutorResult with text, files, status, and history.
        """
        self._produced_files = []
        self.last_messages: list[dict] = []  # exposed for cancel recovery
        progress_updates: list[str] = []

        # Build conversation
        messages = list(history or [])
        user_content = message
        if files:
            file_names = [f.get("name", "file") for f in files]
            user_content += f"\n\n[Attached files: {', '.join(file_names)}]"
        messages.append({"role": "user", "content": user_content})
        self.last_messages = messages  # live ref for cancel recovery

        for step in range(self.max_steps):
            try:
                force_tool = step == 0 and not history
                response = await self._call_llm(
                    messages,
                    force_tool=force_tool,
                )
            except Exception as e:
                logger.exception("LLM call failed at step %d", step)
                return ExecutorResult(
                    text=f"LLM error: {e}",
                    status="failed",
                    history=messages,
                )

            # Parse response
            text_parts, tool_calls = self._parse_response(response)

            if not tool_calls:
                # LLM produced text only - done
                final_text = "\n".join(text_parts)
                messages.append({"role": "assistant", "content": final_text})
                return ExecutorResult(
                    text=final_text,
                    files=self._produced_files,
                    status="completed",
                    progress_updates=progress_updates,
                    history=messages,
                )

            # Append assistant message with tool calls
            messages.append(self._assistant_message(text_parts, tool_calls))

            # Split into synthetic (sequential) and real (parallel) tools
            synthetic = [tc for tc in tool_calls if tc.name in _SYNTHETIC_TOOLS]
            real = [tc for tc in tool_calls if tc.name not in _SYNTHETIC_TOOLS]

            # Handle synthetic tools first (may return early)
            for tc in synthetic:
                result = self._handle_synthetic(
                    tc,
                    messages,
                    progress_updates,
                )
                if result is not None:
                    result.progress_updates = progress_updates
                    return result
                messages.append(self._tool_result_message(tc.id, "Progress noted."))

            # Execute real tools in parallel
            if real:
                results = await asyncio.gather(
                    *(self._execute_tool(tc.name, tc.arguments, files) for tc in real),
                    return_exceptions=True,
                )
                for tc, tool_result in zip(real, results, strict=True):
                    if isinstance(tool_result, BaseException):
                        tool_result = f"Error executing {tc.name}: {tool_result}"
                    elif not isinstance(tool_result, str):
                        tool_result = str(tool_result)
                    messages.append(self._tool_result_message(tc.id, tool_result))

        # Max steps reached
        final_text = "Task partially completed (max steps reached)."
        return ExecutorResult(
            text=final_text,
            files=self._produced_files,
            status="completed",
            history=messages,
            progress_updates=progress_updates,
        )

    def _handle_synthetic(
        self,
        tc: _ToolCall,
        messages: list[dict],
        progress_updates: list[str],
    ) -> ExecutorResult | None:
        """Handle a synthetic tool call. Returns result or None to continue."""
        args = tc.arguments

        if tc.name == TOOL_RETURN_RESULT:
            text = args.get("message", "")
            requested_files = args.get("files")
            if requested_files:
                files = [f for f in self._produced_files if f.get("filename") in requested_files]
            else:
                files = self._produced_files
            # Append tool_result so history is API-valid for continuation
            messages.append(self._tool_result_message(tc.id, "Result delivered."))
            return ExecutorResult(
                text=text,
                files=files,
                status="completed",
                history=messages,
            )

        if tc.name == TOOL_REQUEST_INPUT:
            question = args.get("question", "")
            # Append tool_result so history is valid for continuation
            messages.append(self._tool_result_message(tc.id, "Waiting for input."))
            return ExecutorResult(
                text=question,
                files=self._produced_files,
                status="input_required",
                question=question,
                history=messages,
            )

        if tc.name == TOOL_REJECT_TASK:
            reason = args.get("reason", "Task rejected")
            messages.append(self._tool_result_message(tc.id, "Task rejected."))
            return ExecutorResult(
                text=reason,
                status="rejected",
                history=messages,
            )

        if tc.name == TOOL_REQUEST_AUTH:
            messages.append(self._tool_result_message(tc.id, "Waiting for auth."))
            return ExecutorResult(
                text=args.get("message", "Authentication required"),
                status="auth_required",
                auth_scheme=args.get("scheme", ""),
                history=messages,
            )

        if tc.name == TOOL_REPORT_PROGRESS:
            msg = args.get("message", "")
            progress_updates.append(msg)
            if self.on_progress:
                import asyncio

                if asyncio.iscoroutinefunction(self.on_progress):
                    asyncio.ensure_future(self.on_progress(msg))
                else:
                    self.on_progress(msg)
            return None  # continue loop

        return None

    async def _execute_tool(
        self,
        name: str,
        arguments: dict,
        files: list[dict] | None,
    ) -> str:
        """Execute a real tool and return result as string."""
        td = self._tool_map.get(name)
        if not td:
            return f"Error: unknown tool '{name}'"

        file_params = getattr(td, "resolved_file_params", [])

        # Inject file content into file params.
        if file_params:
            from .file_middleware import strip_vfs_prefix

            by_name: dict[str, str] = {}
            for f in files or []:
                name = f.get("name", "")
                content = f.get("content", "")
                if name and content:
                    by_name[strip_vfs_prefix(name)] = content
            for pf in self._produced_files:
                fn = pf.get("filename", "")
                url = pf.get("download_url", "")
                if fn and url:
                    by_name[strip_vfs_prefix(fn)] = url
            for param in file_params:
                val = arguments.get(param)
                if isinstance(val, str):
                    key = strip_vfs_prefix(val)
                    # Also try basename for path-like values
                    bare = key.rsplit("/", 1)[-1]
                    match_key = key if key in by_name else bare if bare in by_name else None
                    if match_key:
                        fa = {"name": match_key, "content": by_name[match_key]}
                        raw = _raw_tool_schema(td)
                        if _param_is_array(raw, param):
                            arguments[param] = [fa]
                        else:
                            arguments[param] = fa
                    else:
                        logger.warning(
                            "File param %s=%r not in files: %s",
                            param,
                            val,
                            list(by_name.keys()),
                        )

        # Validate input against the raw JSON schema. This preserves
        # oneOf/anyOf/const/enum (a generated Pydantic model flattens
        # combinators to Any and validates nothing useful).
        raw_schema = _raw_tool_schema(td)
        if raw_schema.get("properties"):
            try:
                jsonschema.validate(arguments, raw_schema)
            except jsonschema.ValidationError as e:
                return f"Validation error for {name}: {e.message}"

        try:
            result = await td._arun(**arguments)
        except Exception as e:
            logger.warning("Tool %s failed: %s", name, e)
            return f"Error executing {name}: {e}"

        return self._process_tool_output(result)

    def _process_tool_output(self, result: Any) -> str:
        """Convert tool output to string and track produced files."""
        from pathlib import Path as PathType

        from .base import NamedFile, ToolResult

        if result is None:
            return ""

        # ToolResult: text + data + files
        if isinstance(result, ToolResult):
            parts = []
            if result.text:
                parts.append(result.text)
            if result.data:
                parts.append(json.dumps(result.data, ensure_ascii=False))
            for f in result.files:
                meta = self._track_file(f)
                if meta:
                    parts.append(json.dumps(meta, ensure_ascii=False))
            return "\n".join(parts) if parts else ""

        # NamedFile
        if isinstance(result, NamedFile):
            meta = self._track_file(result)
            return json.dumps(meta, ensure_ascii=False) if meta else result.name

        # Path
        if isinstance(result, PathType) and result.exists():
            meta = self._track_file(result)
            return json.dumps(meta, ensure_ascii=False) if meta else result.name

        # String (may be JSON with legacy download_url)
        text = str(result)
        try:
            data = json.loads(text)
            if isinstance(data, dict) and "download_url" in data:
                self._produced_files.append(data)
        except (json.JSONDecodeError, TypeError):
            pass
        return text

    def _track_file(self, f: Any) -> dict | None:
        """Track a produced file and return metadata dict."""
        import mimetypes
        from pathlib import Path as PathType

        from .base import NamedFile

        if isinstance(f, NamedFile):
            path, name = f.path, f.name
        elif isinstance(f, PathType):
            path, name = f, f.name
        else:
            return None
        mime, _ = mimetypes.guess_type(str(path))
        meta = {
            "filename": name,
            "download_url": f"files/{path.name}",
            "mime_type": mime or "application/octet-stream",
            "size_bytes": path.stat().st_size if path.exists() else 0,
        }
        self._produced_files.append(meta)
        return meta

    # LLM API calls

    async def _call_llm(
        self,
        messages: list[dict],
        *,
        force_tool: bool = False,
    ) -> dict:
        if self._provider in ("azure_anthropic", "anthropic"):
            return await self._call_anthropic(
                messages,
                force_tool=force_tool,
            )
        return await self._call_openai(
            messages,
            force_tool=force_tool,
        )

    async def _call_anthropic(
        self,
        messages: list[dict],
        *,
        force_tool: bool = False,
    ) -> dict:
        url = f"{self.api_base.rstrip('/')}/v1/messages"
        headers: dict[str, str] = {
            "Content-Type": "application/json",
            "anthropic-version": "2023-06-01",
        }
        if self._provider == "azure_anthropic":
            headers["Authorization"] = f"Bearer {self.api_key}"
        else:
            headers["x-api-key"] = self.api_key

        # Build tool list: real + synthetic
        tools = [_tool_schema(t) for t in self.tools]
        tools.extend(_synthetic_tool_schemas_anthropic())

        # Separate system from messages
        api_messages = [m for m in messages if m.get("role") != "system"]

        payload: dict[str, Any] = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "system": self.system_prompt,
            "tools": tools,
            "messages": api_messages,
        }
        if force_tool:
            payload["tool_choice"] = {"type": "any"}

        async with httpx.AsyncClient() as client:
            resp = await client.post(
                url,
                headers=headers,
                json=payload,
                timeout=300.0,
            )
            if resp.status_code >= 400:
                raise RuntimeError(f"Anthropic API error {resp.status_code}: {resp.text[:300]}")
            return resp.json()

    async def _call_openai(
        self,
        messages: list[dict],
        *,
        force_tool: bool = False,
    ) -> dict:
        url = f"{self.api_base.rstrip('/')}/chat/completions"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}",
        }

        tools = [_tool_schema_openai(t) for t in self.tools]
        tools.extend(_synthetic_tool_schemas_openai())

        api_messages = [
            {"role": "system", "content": self.system_prompt},
        ]
        api_messages.extend(messages)

        payload: dict[str, Any] = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "tools": tools,
            "messages": api_messages,
        }
        if force_tool:
            payload["tool_choice"] = "required"

        async with httpx.AsyncClient() as client:
            resp = await client.post(
                url,
                headers=headers,
                json=payload,
                timeout=300.0,
            )
            if resp.status_code >= 400:
                raise RuntimeError(f"OpenAI API error {resp.status_code}: {resp.text[:300]}")
            return resp.json()

    # Response parsing

    def _parse_response(
        self,
        response: dict,
    ) -> tuple[list[str], list[_ToolCall]]:
        """Parse LLM response into text parts and tool calls."""
        if self._provider in ("azure_anthropic", "anthropic"):
            return self._parse_anthropic(response)
        return self._parse_openai(response)

    def _parse_anthropic(
        self,
        response: dict,
    ) -> tuple[list[str], list[_ToolCall]]:
        texts = []
        calls = []
        for block in response.get("content", []):
            if block.get("type") == "text":
                texts.append(block["text"])
            elif block.get("type") == "tool_use":
                calls.append(
                    _ToolCall(
                        id=block["id"],
                        name=block["name"],
                        arguments=block.get("input", {}),
                    )
                )
        return texts, calls

    def _parse_openai(
        self,
        response: dict,
    ) -> tuple[list[str], list[_ToolCall]]:
        texts = []
        calls = []
        choices = response.get("choices", [])
        if not choices:
            return texts, calls
        msg = choices[0].get("message", {})
        if msg.get("content"):
            texts.append(msg["content"])
        for tc in msg.get("tool_calls", []):
            fn = tc.get("function", {})
            try:
                args = json.loads(fn.get("arguments", "{}"))
            except json.JSONDecodeError:
                args = {}
            calls.append(
                _ToolCall(
                    id=tc.get("id", ""),
                    name=fn.get("name", ""),
                    arguments=args,
                )
            )
        return texts, calls

    # Message construction

    def _assistant_message(
        self,
        text_parts: list[str],
        tool_calls: list[_ToolCall],
    ) -> dict:
        """Build assistant message with tool calls."""
        if self._provider in ("azure_anthropic", "anthropic"):
            content = []
            for t in text_parts:
                content.append({"type": "text", "text": t})
            for tc in tool_calls:
                content.append(
                    {
                        "type": "tool_use",
                        "id": tc.id,
                        "name": tc.name,
                        "input": tc.arguments,
                    }
                )
            return {"role": "assistant", "content": content}
        # OpenAI format
        msg: dict[str, Any] = {"role": "assistant"}
        if text_parts:
            msg["content"] = "\n".join(text_parts)
        if tool_calls:
            msg["tool_calls"] = [
                {
                    "id": tc.id,
                    "type": "function",
                    "function": {
                        "name": tc.name,
                        "arguments": json.dumps(tc.arguments),
                    },
                }
                for tc in tool_calls
            ]
        return msg

    def _tool_result_message(self, tool_call_id: str, result: str) -> dict:
        """Build tool result message."""
        if self._provider in ("azure_anthropic", "anthropic"):
            return {
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": tool_call_id,
                        "content": result,
                    }
                ],
            }
        return {
            "role": "tool",
            "tool_call_id": tool_call_id,
            "content": result,
        }