diff --git a/README.md b/README.md index 96f28dfd..4f30a270 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ A template for building [Deep Agents](https://github.com/langchain-ai/deepagents - Orchestrator with analyst and publisher subagents - Skills: `client-intake`, `bmi-report`, `email-formatter` - MCP auth modes: SSO pass-through, OAuth, and DCR +- MCP Apps host APIs (`resources/read` + app `tools/call`) for interactive `ui://` UIs in Template UI **Infrastructure:** - Aegra dev server with Redis-backed SSE streaming @@ -85,6 +86,8 @@ curl -N -X POST "http://localhost:5002/threads/THREAD_ID/runs/stream" \ | `/mcp/{name}/connect` | POST | Start OAuth/DCR flow for an MCP server | | `/mcp/oauth/callback` | GET | OAuth redirect handler | | `/mcp/{name}/status` | GET | MCP connection status for current user | +| `/mcp/{name}/resources/read` | POST | MCP Apps: read a `ui://` resource | +| `/mcp/{name}/tools/call` | POST | MCP Apps: app-initiated tool call | Use [template-ui](https://github.com/redhat-data-and-ai/template-ui) for a full chat experience against this API. @@ -156,6 +159,23 @@ MCP servers are defined in [`config/agent/mcp.json`](./config/agent/mcp.json) an Set `"auth": false` for public/local MCP servers with no Authorization header. +### MCP Apps (interactive UI) + +The agent advertises the MCP Apps UI extension on `initialize` and exposes host proxy routes used by Template UI: + +| Endpoint | Method | Description | +|---|---|---| +| `/mcp/{name}/resources/read` | POST | Read a `ui://` Apps resource (HTML) | +| `/mcp/{name}/tools/call` | POST | App-initiated tool call (`visibility` must include `app`) | + +Add any SEP-1865-compliant App server to `mcp.json` the same way as a normal MCP — no agent code changes. + +Compliance smoke tests: + +```bash +.venv/bin/python -m pytest tests/unit/aegra/test_mcp_apps_smoke.py -q +``` + ### MCP URL by run mode | Mode | `url` in `mcp.json` | diff --git a/config/agent/runtime/ui.yaml b/config/agent/runtime/ui.yaml index 64b59374..640d6e1f 100644 --- a/config/agent/runtime/ui.yaml +++ b/config/agent/runtime/ui.yaml @@ -35,6 +35,7 @@ security: connect_src: ["'self'"] font_src: ["'self'", "data:"] object_src: ["'none'"] + frame_src: ["'self'"] # MCP Apps sandbox proxy iframe frame_ancestors: ["'none'"] cross_origin_embedder_policy: false # Must be false — breaks SSE streaming @@ -55,6 +56,11 @@ otel: enabled: true # Master switch — set true to activate OTEL tracing service_name: "template-ui" +# --- Features --- +features: + mcp_apps: + enabled: true # Serve /sandbox_proxy.html for MCP Apps host rendering + # --- Announcement Banner --- announcement: enabled: true diff --git a/deep_agent/aegra/mcp.py b/deep_agent/aegra/mcp.py index 21bcd0f5..f9ec4b9e 100644 --- a/deep_agent/aegra/mcp.py +++ b/deep_agent/aegra/mcp.py @@ -25,6 +25,11 @@ import httpx from langchain_mcp_adapters.client import MultiServerMCPClient +from deep_agent.aegra.mcp_apps import ( + McpAppCallToolResultInterceptor, + ensure_mcp_apps_capability_advertised, + prepare_tools_for_model, +) from deep_agent.src.agent.config import agent_config from deep_agent.src.error_handling import CircuitBreaker, create_circuit_breaker from deep_agent.src.settings import settings @@ -32,6 +37,9 @@ logger = get_python_logger(log_level=settings.PYTHON_LOG_LEVEL) +# Advertise MCP Apps UI capability for langchain-mcp-adapters sessions. +ensure_mcp_apps_capability_advertised() + _SSO_TOKEN_URL: str = "" _mcp_breaker: CircuitBreaker | None = None @@ -406,22 +414,31 @@ async def _connect_single_server( *, required: bool = False, server_key: str | None = None, + mcp_server: str | None = None, ) -> list[Any]: - """Connect to one MCP server and return its tools. + """Connect to one MCP server and return its model-visible tools. Failures are logged and return empty list for fault isolation. Updates the module-level circuit breaker on success/failure. + Tools are annotated with ``mcp_server`` (mcp.json key) and app-only + tools (``visibility: ["app"]``) are filtered out before return so the + LLM never sees them. No shared Apps registry is written. + Args: - name: Human-readable server identifier used in log messages. + name: Human-readable server identifier used in log messages / + MultiServerMCPClient connection key (may be ``tool_prefix``). config: MCP client connection config (url, transport, headers, etc.). server_cfg: Raw MCP server definition from ``mcp.json`` (auth, ssl_verify). timeout: Seconds before the connection attempt is cancelled. required: If True the server is explicitly enabled in config, so connection failures are logged at error level. - server_key: Optional key override for auth token lookup. + server_key: Optional mcp.json key for auth token lookup and Apps + metadata. Defaults to ``mcp_server`` or ``name`` when omitted. + mcp_server: Alias for ``server_key`` (v2 Apps naming). Ignored when + ``server_key`` is provided. """ - auth_key = server_key or name + auth_key = server_key or mcp_server or name breaker = _get_mcp_breaker() if breaker.is_open: logger.warning(f"[{name}] circuit breaker open — skipping connection") @@ -437,13 +454,25 @@ async def _connect_single_server( _TokenInjectorInterceptor( name, server_cfg, server_key=auth_key ), + # Capture raw MCP CallToolResult before LC conversion so + # UI-bound tools embed spec-faithful mcp_app.result. + McpAppCallToolResultInterceptor(), ], tool_name_prefix=bool(server_cfg.get("tool_prefix", "")), ) tools: list[Any] = await client.get_tools() - logger.info(f"[{name}] loaded {len(tools)} tool(s)") + model_tools = prepare_tools_for_model(tools, auth_key) + skipped = len(tools) - len(model_tools) + if skipped: + logger.info( + "[%s] hid %d app-only tool(s) from the model (%d model-visible)", + name, + skipped, + len(model_tools), + ) + logger.info(f"[{name}] loaded {len(model_tools)} tool(s)") breaker.record_success() - return tools + return model_tools except TimeoutError: if attempt < max_attempts: logger.warning( @@ -458,7 +487,10 @@ async def _connect_single_server( "[%s] MCP OAuth required — returning auth placeholder tool", name, ) - return [_create_auth_placeholder_tool(auth_key, server_cfg)] + return prepare_tools_for_model( + [_create_auth_placeholder_tool(auth_key, server_cfg)], + auth_key, + ) elif _is_auth_error(exc): auth_mode = server_cfg.get("auth_mode", "sso") if auth_mode in ("oauth", "dcr"): @@ -466,7 +498,10 @@ async def _connect_single_server( "[%s] MCP tool discovery auth failed — returning auth placeholder tool", name, ) - return [_create_auth_placeholder_tool(auth_key, server_cfg)] + return prepare_tools_for_model( + [_create_auth_placeholder_tool(auth_key, server_cfg)], + auth_key, + ) else: logger.warning( f"[{name}] MCP auth failed — {type(exc).__name__}: {exc}" @@ -639,7 +674,12 @@ async def get_mcp_tools( mcp_prefix_name, auth_mode, ) - placeholder_tools.append([_create_auth_placeholder_tool(name, entry)]) + placeholder_tools.append( + prepare_tools_for_model( + [_create_auth_placeholder_tool(name, entry)], + name, + ) + ) continue connect_jobs.append( _connect_single_server( diff --git a/deep_agent/aegra/mcp_apps.py b/deep_agent/aegra/mcp_apps.py new file mode 100644 index 00000000..37242ac1 --- /dev/null +++ b/deep_agent/aegra/mcp_apps.py @@ -0,0 +1,403 @@ +"""MCP Apps (SEP-1865) client-side capability helpers. + +Advertises ``io.modelcontextprotocol/ui`` during MCP ``initialize`` so +spec-compliant App servers can enable UI-bound tools. The Python MCP SDK +and langchain-mcp-adapters do not yet expose a first-class hook for +``capabilities.extensions``, so we wrap ``ClientSession.initialize`` once +at process start. + +Also provides pure helpers to read tool ``_meta.ui``, stamp ``mcp_server``, +and filter app-only tools out of the model-facing tool list (no shared +in-memory Apps registry — safe for multi-pod). +""" + +from __future__ import annotations + +import contextvars +import inspect +from typing import Any + +from mcp import types +from mcp.client.session import ClientSession + +# SEP-1865 / ext-apps extension identifier and MVP MIME type. +MCP_APPS_EXTENSION_ID = "io.modelcontextprotocol/ui" +MCP_APPS_MIME_TYPE = "text/html;profile=mcp-app" + +_DEFAULT_VISIBILITY: tuple[str, ...] = ("model", "app") + +_original_initialize: Any | None = None +_patch_installed = False + +# Request-scoped (same async task) capture of the raw MCP CallToolResult so we can +# embed MCP-shaped content on ToolMessage.artifact.mcp_app. Not a cross-pod cache — +# the snapshot is copied onto the message before the tool call returns. +_captured_call_tool_result: contextvars.ContextVar[dict[str, Any] | None] = ( + contextvars.ContextVar("_captured_call_tool_result", default=None) +) + + +def mcp_apps_extension_settings() -> dict[str, Any]: + """Return the settings map advertised under capabilities.extensions.""" + return {"mimeTypes": [MCP_APPS_MIME_TYPE]} + + +def get_tool_ui_meta(tool: Any) -> dict[str, Any]: + """Extract MCP Apps UI metadata from a LangChain tool. + + Reads ``metadata["_meta"]["ui"]`` (current) and falls back to the + deprecated flat ``metadata["_meta"]["ui/resourceUri"]`` form. + """ + metadata = getattr(tool, "metadata", None) or {} + if not isinstance(metadata, dict): + return {} + + meta = metadata.get("_meta") + if not isinstance(meta, dict): + return {} + + ui = meta.get("ui") + if isinstance(ui, dict): + return dict(ui) + + # Deprecated: _meta["ui/resourceUri"] = "ui://..." + deprecated_uri = meta.get("ui/resourceUri") + if isinstance(deprecated_uri, str) and deprecated_uri: + return {"resourceUri": deprecated_uri} + + return {} + + +def get_tool_visibility(tool: Any) -> list[str]: + """Return tool visibility list; defaults to ``["model", "app"]`` when omitted. + + An explicit empty list ``[]`` means visible to neither model nor app. + """ + ui = get_tool_ui_meta(tool) + visibility = ui.get("visibility") + if isinstance(visibility, list): + return [str(v) for v in visibility] + return list(_DEFAULT_VISIBILITY) + + +def is_model_visible(tool: Any) -> bool: + """True if the tool may be listed for / called by the model.""" + return "model" in get_tool_visibility(tool) + + +def is_app_callable(tool: Any) -> bool: + """True if an MCP App may call this tool via the host proxy.""" + return "app" in get_tool_visibility(tool) + + +def annotate_mcp_tool(tool: Any, mcp_server: str) -> Any: + """Stamp ``mcp_server`` onto tool.metadata (in place). Preserves ``_meta``.""" + metadata = getattr(tool, "metadata", None) + if not isinstance(metadata, dict): + metadata = {} + else: + metadata = dict(metadata) + metadata["mcp_server"] = mcp_server + tool.metadata = metadata + return tool + + +def build_mcp_app_descriptor(tool: Any) -> dict[str, Any] | None: + """Return a host-facing Apps descriptor if the tool declares a ui:// resource.""" + ui = get_tool_ui_meta(tool) + resource_uri = ui.get("resourceUri") + if not isinstance(resource_uri, str) or not resource_uri.startswith("ui://"): + return None + + metadata = getattr(tool, "metadata", None) or {} + server = metadata.get("mcp_server") if isinstance(metadata, dict) else None + if not isinstance(server, str) or not server: + return None + + return { + "server": server, + "resourceUri": resource_uri, + "toolName": getattr(tool, "name", None), + "visibility": get_tool_visibility(tool), + } + + +def _as_content_blocks(content: Any) -> list[Any]: + """Normalize LangChain tool content into MCP CallToolResult content blocks.""" + if content is None: + return [] + if isinstance(content, str): + return [{"type": "text", "text": content}] + if isinstance(content, list): + return list(content) + return [{"type": "text", "text": str(content)}] + + +def serialize_call_tool_result_for_app(result: Any) -> dict[str, Any]: + """Convert a raw MCP ``CallToolResult`` into a JSON-friendly Apps host payload.""" + content_out: list[Any] = [] + for block in getattr(result, "content", None) or []: + if hasattr(block, "model_dump"): + content_out.append( + block.model_dump(by_alias=True, exclude_none=True, mode="json") + ) + elif isinstance(block, dict): + content_out.append(block) + else: + content_out.append({"type": "text", "text": str(block)}) + + meta = getattr(result, "meta", None) + if meta is not None and hasattr(meta, "model_dump"): + meta = meta.model_dump(by_alias=True, exclude_none=True, mode="json") + + payload: dict[str, Any] = { + "content": content_out, + "isError": bool(getattr(result, "isError", False)), + } + # Omit nulls — hosts validate with CallToolResultSchema which rejects null. + structured = getattr(result, "structuredContent", None) + if structured is not None: + payload["structuredContent"] = structured + if meta is not None: + payload["_meta"] = meta + return payload + + +def capture_call_tool_result_for_app(result: Any) -> None: + """Store a raw MCP result for the current tool invocation (same async task).""" + _captured_call_tool_result.set(serialize_call_tool_result_for_app(result)) + + +def take_captured_call_tool_result() -> dict[str, Any] | None: + """Return and clear the captured MCP result for this tool invocation.""" + value = _captured_call_tool_result.get() + _captured_call_tool_result.set(None) + return value + + +class McpAppCallToolResultInterceptor: + """Capture raw MCP ``CallToolResult`` before LangChain content conversion. + + Registered on ``MultiServerMCPClient`` so model-invoked UI tools can embed a + spec-faithful ``mcp_app.result`` (including ``isError`` and ``_meta``) on the + ToolMessage artifact. Capture is contextvar-scoped to the call — not a pod cache. + """ + + async def __call__(self, request: Any, handler: Any) -> Any: + """Run the next handler and capture ``CallToolResult`` when present.""" + _captured_call_tool_result.set(None) + result = await handler(request) + if isinstance(result, types.CallToolResult): + capture_call_tool_result_for_app(result) + return result + + +def attach_mcp_app_to_tool_result( + result: Any, + descriptor: dict[str, Any], + *, + mcp_result: dict[str, Any] | None = None, +) -> Any: + """Embed ``mcp_app`` on a content_and_artifact tool result for streaming hosts. + + Prefer ``mcp_result`` (raw MCP ``CallToolResult`` shape) when provided so the + View receives MCP content blocks, ``structuredContent``, ``isError``, and ``_meta``. + """ + if isinstance(result, tuple) and len(result) == 2: + content, artifact = result + else: + content, artifact = result, None + + art: dict[str, Any] = dict(artifact) if isinstance(artifact, dict) else {} + + if mcp_result is not None: + app_result: dict[str, Any] = { + "content": list(mcp_result.get("content") or []), + "isError": bool(mcp_result.get("isError")), + } + structured = mcp_result.get("structuredContent") + if structured is not None: + app_result["structuredContent"] = structured + meta = mcp_result.get("_meta") + if meta is not None: + app_result["_meta"] = meta + else: + structured = art.get("structured_content") + if structured is None: + structured = art.get("structuredContent") + app_result = { + "content": _as_content_blocks(content), + "isError": False, + } + if structured is not None: + app_result["structuredContent"] = structured + + art["mcp_app"] = {**descriptor, "result": app_result} + return content, art + + +def extract_mcp_app_from_message(msg: Any) -> dict[str, Any] | None: + """Pull ``mcp_app`` from a ToolMessage (artifact / kwargs / response_metadata).""" + artifact = getattr(msg, "artifact", None) + if isinstance(artifact, dict): + payload = artifact.get("mcp_app") or artifact.get("mcpApp") + if isinstance(payload, dict): + return payload + + for container_name in ("additional_kwargs", "response_metadata"): + container = getattr(msg, container_name, None) + if isinstance(container, dict): + payload = container.get("mcpApp") or container.get("mcp_app") + if isinstance(payload, dict): + return payload + return None + + +def wrap_tool_to_attach_mcp_app(tool: Any) -> Any: + """Wrap a tool coroutine so UI-bound results carry ``artifact.mcp_app``.""" + descriptor = build_mcp_app_descriptor(tool) + if not descriptor: + return tool + + coroutine = getattr(tool, "coroutine", None) + if not inspect.iscoroutinefunction(coroutine): + return tool + + async def wrapped_coroutine(**kwargs: Any) -> Any: + from langchain_core.tools import ToolException + + try: + result = await coroutine(**kwargs) + except Exception as exc: + # ToolException: adapters raise when CallToolResult.isError is true. + # NotImplementedError (etc.): adapters fail converting AudioContent / + # other blocks to LangChain — interceptor already captured the raw MCP + # result; keep the App mountable with a text stub for the model. + captured = take_captured_call_tool_result() + if captured is None: + if not isinstance(exc, ToolException): + raise + captured = { + "content": [{"type": "text", "text": str(exc)}], + "isError": True, + } + text_parts: list[str] = [] + for block in captured.get("content") or []: + if isinstance(block, dict) and block.get("type") == "text": + text = block.get("text") + if isinstance(text, str) and text: + text_parts.append(text) + if text_parts: + fallback = "\n".join(text_parts) + elif isinstance(exc, ToolException): + fallback = str(exc) + else: + fallback = ( + "Tool returned content the model path cannot convert " + f"({type(exc).__name__}); see the interactive UI." + ) + return attach_mcp_app_to_tool_result( + ([{"type": "text", "text": fallback}], None), + {**descriptor, "arguments": kwargs}, + mcp_result=captured, + ) + + captured = take_captured_call_tool_result() + return attach_mcp_app_to_tool_result( + result, + {**descriptor, "arguments": kwargs}, + mcp_result=captured, + ) + + try: + return tool.model_copy(update={"coroutine": wrapped_coroutine}) + except Exception: + tool.coroutine = wrapped_coroutine + return tool + + +def prepare_tools_for_model(tools: list[Any], mcp_server: str) -> list[Any]: + """Annotate tools with ``mcp_server`` and drop app-only tools for the LLM. + + UI-bound tools are wrapped so their results embed an ``mcp_app`` descriptor + for the chat host. Does not store tools in a shared registry. + """ + prepared: list[Any] = [] + for tool in tools: + annotate_mcp_tool(tool, mcp_server) + if is_model_visible(tool): + prepared.append(wrap_tool_to_attach_mcp_app(tool)) + return prepared + + +def inject_ui_extension_into_request( + request: types.ClientRequest, +) -> types.ClientRequest: + """Return a copy of *request* with MCP Apps UI capability on initialize. + + Non-initialize requests are returned unchanged. Idempotent if extensions + are already present. + """ + root = request.root + if not isinstance(root, types.InitializeRequest): + return request + + caps = root.params.capabilities + existing = getattr(caps, "extensions", None) or {} + if ( + isinstance(existing, dict) + and MCP_APPS_EXTENSION_ID in existing + and isinstance(existing[MCP_APPS_EXTENSION_ID], dict) + and MCP_APPS_MIME_TYPE + in (existing[MCP_APPS_EXTENSION_ID].get("mimeTypes") or []) + ): + return request + + extensions = { + **(existing if isinstance(existing, dict) else {}), + MCP_APPS_EXTENSION_ID: mcp_apps_extension_settings(), + } + new_caps = caps.model_copy(update={"extensions": extensions}) + new_params = root.params.model_copy(update={"capabilities": new_caps}) + new_root = root.model_copy(update={"params": new_params}) + return types.ClientRequest(new_root) + + +async def _initialize_with_mcp_apps(self: ClientSession) -> types.InitializeResult: + """Wrap stock initialize so the handshake advertises MCP Apps support.""" + if _original_initialize is None: + raise RuntimeError("MCP Apps initialize patch was not installed") + + original_send_request = self.send_request + + async def send_request_with_extensions( + request: Any, + result_type: Any, + **kwargs: Any, + ) -> Any: + if isinstance(request, types.ClientRequest): + request = inject_ui_extension_into_request(request) + return await original_send_request(request, result_type, **kwargs) + + self.send_request = send_request_with_extensions + try: + return await _original_initialize(self) + finally: + self.send_request = original_send_request + + +def ensure_mcp_apps_capability_advertised() -> bool: + """Patch ``ClientSession.initialize`` once (idempotent). + + Returns: + True if the patch was newly installed, False if already installed. + """ + global _original_initialize, _patch_installed # noqa: PLW0603 + + if _patch_installed: + return False + + _original_initialize = ClientSession.initialize + ClientSession.initialize = _initialize_with_mcp_apps + _patch_installed = True + return True diff --git a/deep_agent/aegra/mcp_host.py b/deep_agent/aegra/mcp_host.py new file mode 100644 index 00000000..ea9f6f87 --- /dev/null +++ b/deep_agent/aegra/mcp_host.py @@ -0,0 +1,300 @@ +"""Request-scoped MCP Apps host proxy (resources + app tools/call). + +These helpers open a short-lived MCP session per HTTP request. Auth tokens +come from Redis (oauth/dcr) or the caller's SSO bearer — nothing is stored +in process memory for multi-pod safety. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from types import SimpleNamespace +from typing import Any, cast + +from fastapi import HTTPException +from langchain_mcp_adapters.client import MultiServerMCPClient + +from deep_agent.aegra.mcp import ( + _build_server_config, + _get_server_configs, + _resolve_connection_token, +) +from deep_agent.aegra.mcp_apps import ( + ensure_mcp_apps_capability_advertised, + get_tool_visibility, + is_app_callable, +) +from deep_agent.utils.pylogger import get_python_logger + +logger = get_python_logger() + +# Cap tools/list pagination when resolving a tool by name (host tools/call). +_MAX_TOOLS_LIST_PAGES = 20 + + +def _tool_from_mcp_meta(name: str, meta: dict[str, Any] | None) -> SimpleNamespace: + """Adapt an MCP tool meta dict into the shape used by mcp_apps helpers.""" + return SimpleNamespace(name=name, metadata={"_meta": meta or {}}) + + +def _authorization_required(mcp_name: str) -> HTTPException: + from deep_agent.aegra.mcp_auth import get_mcp_credential_resolver + + return HTTPException( + status_code=401, + detail={ + "error": "authorization_required", + "mcp_name": mcp_name, + "connect_url": get_mcp_credential_resolver().connect_url(mcp_name), + }, + ) + + +def _get_enabled_server(mcp_name: str) -> dict[str, Any]: + entry = _get_server_configs().get(mcp_name) + if not entry or not entry.get("enabled", False): + raise HTTPException( + status_code=404, + detail=f"Unknown or disabled MCP server: {mcp_name}", + ) + return entry + + +async def _resolve_bearer( + mcp_name: str, + entry: dict[str, Any], + *, + user_id: str, + sso_token: str | None, +) -> str | None: + """Resolve a bearer token for this request or raise HTTP 401 when required.""" + from deep_agent.aegra.mcp_auth import NeedsAuthorization + + auth_mode = entry.get("auth_mode", "sso") + auth_required = entry.get("auth", True) + + try: + bearer = await _resolve_connection_token(mcp_name, entry, sso_token, user_id) + except NeedsAuthorization: + raise _authorization_required(mcp_name) from None + + if not auth_required: + return bearer + + if auth_mode in ("oauth", "dcr") and not bearer: + raise _authorization_required(mcp_name) + + if auth_mode == "sso" and not bearer: + raise HTTPException( + status_code=401, + detail="Missing Authorization bearer token for SSO MCP access", + ) + + if auth_mode == "api_key" and not bearer: + raise HTTPException( + status_code=500, + detail=f"MCP '{mcp_name}' api_key is not configured on the agent", + ) + + return bearer + + +@asynccontextmanager +async def mcp_session( + mcp_name: str, + *, + user_id: str, + sso_token: str | None, +) -> AsyncIterator[Any]: + """Open a short-lived MCP client session for host proxy operations.""" + ensure_mcp_apps_capability_advertised() + entry = _get_enabled_server(mcp_name) + bearer = await _resolve_bearer( + mcp_name, entry, user_id=user_id, sso_token=sso_token + ) + config = _build_server_config(entry, bearer) + client = MultiServerMCPClient({mcp_name: config}) + timeout = float(entry.get("timeout", 30)) + async with asyncio.timeout(timeout): + async with client.session(mcp_name) as session: + yield session + + +async def _find_mcp_tool(session: Any, tool_name: str) -> Any | None: + """Find a tool by name using paginated tools/list (live server, not cache).""" + cursor: str | None = None + for _ in range(_MAX_TOOLS_LIST_PAGES): + page = await session.list_tools(cursor=cursor) + for tool in page.tools or []: + if tool.name == tool_name: + return tool + cursor = page.nextCursor + if not cursor: + return None + return None + + +async def list_tools( + mcp_name: str, + *, + cursor: str | None = None, + user_id: str, + sso_token: str | None, +) -> dict[str, Any]: + """Proxy ``tools/list`` on *mcp_name* (host metadata; live server).""" + async with mcp_session(mcp_name, user_id=user_id, sso_token=sso_token) as session: + result = await session.list_tools(cursor=cursor) + + payload = cast( + dict[str, Any], + result.model_dump(by_alias=True, mode="json", exclude_none=True), + ) + logger.info( + "MCP Apps tools/list ok server=%s count=%d", + mcp_name, + len(payload.get("tools") or []), + ) + return payload + + +async def list_resources( + mcp_name: str, + *, + cursor: str | None = None, + user_id: str, + sso_token: str | None, +) -> dict[str, Any]: + """Proxy ``resources/list`` on *mcp_name* (View → Host → Server).""" + async with mcp_session(mcp_name, user_id=user_id, sso_token=sso_token) as session: + result = await session.list_resources(cursor=cursor) + + payload = cast( + dict[str, Any], + result.model_dump(by_alias=True, mode="json", exclude_none=True), + ) + logger.info( + "MCP Apps resources/list ok server=%s count=%d", + mcp_name, + len(payload.get("resources") or []), + ) + return payload + + +async def list_resource_templates( + mcp_name: str, + *, + cursor: str | None = None, + user_id: str, + sso_token: str | None, +) -> dict[str, Any]: + """Proxy ``resources/templates/list`` on *mcp_name* (View → Host → Server).""" + async with mcp_session(mcp_name, user_id=user_id, sso_token=sso_token) as session: + result = await session.list_resource_templates(cursor=cursor) + + payload = cast( + dict[str, Any], + result.model_dump(by_alias=True, mode="json", exclude_none=True), + ) + logger.info( + "MCP Apps resources/templates/list ok server=%s count=%d", + mcp_name, + len( + payload.get("resourceTemplates") or payload.get("resource_templates") or [] + ), + ) + return payload + + +async def read_resource( + mcp_name: str, + uri: str, + *, + user_id: str, + sso_token: str | None, +) -> dict[str, Any]: + """Proxy ``resources/read`` for any resource URI on *mcp_name*. + + Used both for host HTML fetch (``ui://``) and View ``readServerResource`` + (any server-registered URI, e.g. ``showcase://sample.json``). + """ + if not isinstance(uri, str) or not uri.strip(): + raise HTTPException(status_code=400, detail="uri is required") + + async with mcp_session(mcp_name, user_id=user_id, sso_token=sso_token) as session: + result = await session.read_resource(uri) + + payload = cast( + dict[str, Any], + result.model_dump(by_alias=True, mode="json", exclude_none=True), + ) + logger.info( + "MCP Apps resources/read ok server=%s uri=%s contents=%d", + mcp_name, + uri, + len(payload.get("contents") or []), + ) + return payload + + +# Back-compat alias for callers/tests that used the old name. +read_ui_resource = read_resource + + +async def call_app_tool( + mcp_name: str, + tool_name: str, + arguments: dict[str, Any] | None, + *, + user_id: str, + sso_token: str | None, +) -> dict[str, Any]: + r"""Proxy ``tools/call`` for an app-visible tool on *mcp_name*. + + Enforces ``_meta.ui.visibility`` includes ``\"app\"`` using a live + ``tools/list`` (not the per-pod LLM tool cache). + """ + if not tool_name or not isinstance(tool_name, str): + raise HTTPException(status_code=400, detail="tool name is required") + + # Defer HTTPException until after mcp_session exits. Raising inside the + # streamable-HTTP TaskGroup wraps FastAPI errors as ExceptionGroup → 500. + not_found_detail: str | None = None + deny_detail: dict[str, Any] | None = None + result: Any | None = None + + async with mcp_session(mcp_name, user_id=user_id, sso_token=sso_token) as session: + tool = await _find_mcp_tool(session, tool_name) + if tool is None: + not_found_detail = f"Tool not found on MCP server '{mcp_name}': {tool_name}" + else: + adapted = _tool_from_mcp_meta(tool.name, getattr(tool, "meta", None)) + if not is_app_callable(adapted): + deny_detail = { + "error": "tool_not_app_callable", + "mcp_name": mcp_name, + "tool": tool_name, + "visibility": get_tool_visibility(adapted), + } + else: + result = await session.call_tool(tool_name, arguments or {}) + + if not_found_detail is not None: + raise HTTPException(status_code=404, detail=not_found_detail) + if deny_detail is not None: + raise HTTPException(status_code=403, detail=deny_detail) + if result is None: + raise HTTPException(status_code=500, detail="tools/call produced no result") + + payload = cast( + dict[str, Any], + result.model_dump(by_alias=True, mode="json", exclude_none=True), + ) + logger.info( + "MCP Apps tools/call ok server=%s tool=%s isError=%s", + mcp_name, + tool_name, + payload.get("isError", False), + ) + return payload diff --git a/deep_agent/aegra/mcp_routes.py b/deep_agent/aegra/mcp_routes.py index d1995c29..38919230 100644 --- a/deep_agent/aegra/mcp_routes.py +++ b/deep_agent/aegra/mcp_routes.py @@ -1,4 +1,4 @@ -"""HTTP routes for per-MCP OAuth/DCR connect, callback, and status.""" +"""HTTP routes for per-MCP OAuth/DCR connect, callback, status, and Apps host proxy.""" from __future__ import annotations @@ -9,7 +9,7 @@ from deep_agent.src.agent.config import agent_config -router = APIRouter(tags=["mcp-oauth"]) +router = APIRouter(tags=["mcp"]) async def _authenticated_user_id(request: Request) -> str: @@ -44,6 +44,26 @@ async def _authenticated_user_id(request: Request) -> str: return str(payload["sub"]) +def _sso_bearer_from_request(request: Request) -> str | None: + """Return the raw Bearer token (SSO access token) when present.""" + auth_header = request.headers.get("authorization", "") + if auth_header.startswith("Bearer "): + return auth_header[7:].strip() or None + return None + + +def _http_exception_response(exc: HTTPException) -> JSONResponse | None: + """Return JSONResponse for dict ``detail`` so aegra does not stringify-fail. + + aegra's ``AgentProtocolError.message`` is a string; dict details (e.g. + ``authorization_required``, ``tool_not_app_callable``) must bypass that + handler or the client sees 500 instead of 401/403. + """ + if isinstance(exc.detail, dict): + return JSONResponse(status_code=exc.status_code, content=exc.detail) + return None + + @router.post("/mcp/{mcp_name}/connect") async def mcp_connect(mcp_name: str, request: Request) -> JSONResponse: """Start OAuth/DCR authorization for an MCP server.""" @@ -102,3 +122,178 @@ async def get_agent_info() -> dict[str, Any]: and (cfg.get("oauth") or {}).get("grant_type") != "client_credentials" ) return {"name": agent_config.get_name(), "oauth_mcps": oauth_mcps} + + +@router.post("/mcp/{mcp_name}/tools/list") +async def mcp_tools_list(mcp_name: str, request: Request) -> JSONResponse: + """Proxy MCP ``tools/list`` for Apps host metadata (stateless).""" + from deep_agent.aegra.mcp_host import list_tools + + user_id = await _authenticated_user_id(request) + cursor: str | None = None + try: + body = await request.json() + except Exception: + body = {} + if body is not None and not isinstance(body, dict): + raise HTTPException(status_code=400, detail="JSON body must be an object") + if isinstance(body, dict): + raw_cursor = body.get("cursor") + if raw_cursor is not None and not isinstance(raw_cursor, str): + raise HTTPException(status_code=400, detail="cursor must be a string") + cursor = raw_cursor + + try: + result = await list_tools( + mcp_name, + cursor=cursor, + user_id=user_id, + sso_token=_sso_bearer_from_request(request), + ) + except HTTPException as exc: + as_json = _http_exception_response(exc) + if as_json is not None: + return as_json + raise + return JSONResponse(content=result) + + +@router.post("/mcp/{mcp_name}/resources/list") +async def mcp_resources_list(mcp_name: str, request: Request) -> JSONResponse: + """Proxy MCP ``resources/list`` for Apps Views (stateless).""" + from deep_agent.aegra.mcp_host import list_resources + + user_id = await _authenticated_user_id(request) + cursor: str | None = None + try: + body = await request.json() + except Exception: + body = {} + if body is not None and not isinstance(body, dict): + raise HTTPException(status_code=400, detail="JSON body must be an object") + if isinstance(body, dict): + raw_cursor = body.get("cursor") + if raw_cursor is not None and not isinstance(raw_cursor, str): + raise HTTPException(status_code=400, detail="cursor must be a string") + cursor = raw_cursor + + try: + result = await list_resources( + mcp_name, + cursor=cursor, + user_id=user_id, + sso_token=_sso_bearer_from_request(request), + ) + except HTTPException as exc: + as_json = _http_exception_response(exc) + if as_json is not None: + return as_json + raise + return JSONResponse(content=result) + + +@router.post("/mcp/{mcp_name}/resources/templates/list") +async def mcp_resource_templates_list(mcp_name: str, request: Request) -> JSONResponse: + """Proxy MCP ``resources/templates/list`` for Apps Views (stateless).""" + from deep_agent.aegra.mcp_host import list_resource_templates + + user_id = await _authenticated_user_id(request) + cursor: str | None = None + try: + body = await request.json() + except Exception: + body = {} + if body is not None and not isinstance(body, dict): + raise HTTPException(status_code=400, detail="JSON body must be an object") + if isinstance(body, dict): + raw_cursor = body.get("cursor") + if raw_cursor is not None and not isinstance(raw_cursor, str): + raise HTTPException(status_code=400, detail="cursor must be a string") + cursor = raw_cursor + + try: + result = await list_resource_templates( + mcp_name, + cursor=cursor, + user_id=user_id, + sso_token=_sso_bearer_from_request(request), + ) + except HTTPException as exc: + as_json = _http_exception_response(exc) + if as_json is not None: + return as_json + raise + return JSONResponse(content=result) + + +@router.post("/mcp/{mcp_name}/resources/read") +async def mcp_resources_read(mcp_name: str, request: Request) -> JSONResponse: + """Proxy MCP ``resources/read`` for Apps (any resource URI; stateless).""" + from deep_agent.aegra.mcp_host import read_resource + + user_id = await _authenticated_user_id(request) + try: + body = await request.json() + except Exception as exc: + raise HTTPException(status_code=400, detail="Invalid JSON body") from exc + + if not isinstance(body, dict): + raise HTTPException(status_code=400, detail="JSON body must be an object") + + uri = body.get("uri") + try: + result = await read_resource( + mcp_name, + uri if isinstance(uri, str) else "", + user_id=user_id, + sso_token=_sso_bearer_from_request(request), + ) + except HTTPException as exc: + as_json = _http_exception_response(exc) + if as_json is not None: + return as_json + raise + return JSONResponse(content=result) + + +@router.post("/mcp/{mcp_name}/tools/call") +async def mcp_tools_call(mcp_name: str, request: Request) -> JSONResponse: + """Proxy MCP ``tools/call`` for app-initiated tools (stateless). + + Enforces tool ``visibility`` includes ``app``. Does not use the per-pod + LLM tool cache — each call lists/calls against the live MCP server. + """ + from deep_agent.aegra.mcp_host import call_app_tool + + user_id = await _authenticated_user_id(request) + try: + body = await request.json() + except Exception as exc: + raise HTTPException(status_code=400, detail="Invalid JSON body") from exc + + if not isinstance(body, dict): + raise HTTPException(status_code=400, detail="JSON body must be an object") + + tool_name = body.get("name") + arguments = body.get("arguments") + if arguments is None: + arguments = {} + if not isinstance(arguments, dict): + raise HTTPException( + status_code=400, detail="arguments must be a JSON object when provided" + ) + + try: + result = await call_app_tool( + mcp_name, + tool_name if isinstance(tool_name, str) else "", + arguments, + user_id=user_id, + sso_token=_sso_bearer_from_request(request), + ) + except HTTPException as exc: + as_json = _http_exception_response(exc) + if as_json is not None: + return as_json + raise + return JSONResponse(content=result) diff --git a/deep_agent/aegra/serialization.py b/deep_agent/aegra/serialization.py index 992a669a..9c9fd1c7 100644 --- a/deep_agent/aegra/serialization.py +++ b/deep_agent/aegra/serialization.py @@ -39,10 +39,22 @@ def serialize_message(msg: BaseMessage) -> dict[str, Any]: if isinstance(msg, ToolMessage): data["tool_call_id"] = msg.tool_call_id data["name"] = getattr(msg, "name", None) + artifact = getattr(msg, "artifact", None) + if artifact is not None: + data["artifact"] = _safe_serialize(artifact) + from deep_agent.aegra.mcp_apps import extract_mcp_app_from_message + + mcp_app = extract_mcp_app_from_message(msg) + if mcp_app is not None: + data["mcpApp"] = _safe_serialize(mcp_app) if msg.response_metadata: data["response_metadata"] = _safe_serialize(msg.response_metadata) + additional_kwargs = getattr(msg, "additional_kwargs", None) + if additional_kwargs: + data["additional_kwargs"] = _safe_serialize(additional_kwargs) + return data @@ -51,25 +63,41 @@ def deserialize_message(data: dict[str, Any]) -> BaseMessage: msg_type = data.get("type", "human") content = data.get("content", "") msg_id = data.get("id") + additional_kwargs = data.get("additional_kwargs") + kwargs_extra: dict[str, Any] = {} + if isinstance(additional_kwargs, dict): + kwargs_extra["additional_kwargs"] = dict(additional_kwargs) if msg_type == "human": - return HumanMessage(content=content, id=msg_id) + return HumanMessage(content=content, id=msg_id, **kwargs_extra) elif msg_type == "ai": - kwargs: dict[str, Any] = {"content": content, "id": msg_id} + kwargs: dict[str, Any] = {"content": content, "id": msg_id, **kwargs_extra} if "tool_calls" in data: kwargs["tool_calls"] = data["tool_calls"] return AIMessage(**kwargs) elif msg_type == "system": - return SystemMessage(content=content, id=msg_id) + return SystemMessage(content=content, id=msg_id, **kwargs_extra) elif msg_type == "tool": - return ToolMessage( - content=content, - tool_call_id=data.get("tool_call_id", ""), - name=data.get("name"), - id=msg_id, - ) + tool_kwargs: dict[str, Any] = { + "content": content, + "tool_call_id": data.get("tool_call_id", ""), + "name": data.get("name"), + "id": msg_id, + } + if "artifact" in data: + tool_kwargs["artifact"] = data["artifact"] + merged_kwargs = dict(kwargs_extra.get("additional_kwargs") or {}) + if ( + "mcpApp" in data + and "mcpApp" not in merged_kwargs + and "mcp_app" not in merged_kwargs + ): + merged_kwargs["mcpApp"] = data["mcpApp"] + if merged_kwargs: + tool_kwargs["additional_kwargs"] = merged_kwargs + return ToolMessage(**tool_kwargs) else: - return HumanMessage(content=content, id=msg_id) + return HumanMessage(content=content, id=msg_id, **kwargs_extra) def serialize_state(state: dict[str, Any]) -> dict[str, Any]: diff --git a/deep_agent/aegra/startup.py b/deep_agent/aegra/startup.py index b64586ee..e61bdab9 100644 --- a/deep_agent/aegra/startup.py +++ b/deep_agent/aegra/startup.py @@ -55,6 +55,7 @@ async def run_startup() -> dict[str, str]: results["database"] = await _ensure_database() _check_mcp_encryption_key() results["resume"] = await _resume_interrupted_runs() + results["mcp_apps"] = _setup_mcp_apps_capability() results["cache"] = await _warm_caches() results["scheduler"] = await _start_scheduler() results["otel"] = _setup_otel() @@ -279,9 +280,10 @@ def _check_mcp_encryption_key() -> None: """Warn if any MCP server uses oauth/dcr but MCP_TOKEN_ENCRYPTION_KEY is not set.""" try: from deep_agent.src.agent.config import agent_config + from deep_agent.src.settings import settings servers = agent_config.get_mcp_servers() - dcr_enabled = os.environ.get("MCP_DCR_ENABLED", "true").lower() == "true" + dcr_enabled = settings.MCP_DCR_ENABLED check_modes = {"oauth", "dcr"} if dcr_enabled else {"oauth"} needs_key = any( s.get("auth_mode") in check_modes @@ -297,6 +299,18 @@ def _check_mcp_encryption_key() -> None: logger.debug("MCP encryption key check skipped", exc_info=True) +def _setup_mcp_apps_capability() -> str: + """Ensure MCP initialize advertises the Apps UI extension (SEP-1865).""" + try: + from deep_agent.aegra.mcp_apps import ensure_mcp_apps_capability_advertised + + newly_installed = ensure_mcp_apps_capability_advertised() + return "ok" if newly_installed else "already_installed" + except Exception as exc: + logger.error("MCP Apps capability setup failed: %s", exc) + return f"error: {exc}" + + async def _ensure_database() -> str: """Create personalization, feedback, and token budget tables if they don't exist.""" try: diff --git a/tests/unit/aegra/test_mcp_apps.py b/tests/unit/aegra/test_mcp_apps.py new file mode 100644 index 00000000..46a69461 --- /dev/null +++ b/tests/unit/aegra/test_mcp_apps.py @@ -0,0 +1,515 @@ +"""Unit tests for MCP Apps client capability advertising.""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from mcp import types +from mcp.client.session import ClientSession + +from deep_agent.aegra import mcp_apps +from deep_agent.aegra.mcp_apps import ( + MCP_APPS_EXTENSION_ID, + MCP_APPS_MIME_TYPE, + McpAppCallToolResultInterceptor, + annotate_mcp_tool, + attach_mcp_app_to_tool_result, + build_mcp_app_descriptor, + capture_call_tool_result_for_app, + ensure_mcp_apps_capability_advertised, + extract_mcp_app_from_message, + get_tool_ui_meta, + get_tool_visibility, + inject_ui_extension_into_request, + is_app_callable, + is_model_visible, + mcp_apps_extension_settings, + prepare_tools_for_model, + serialize_call_tool_result_for_app, + take_captured_call_tool_result, +) + + +def _make_initialize_request( + *, + extensions: dict | None = None, +) -> types.ClientRequest: + caps_kwargs: dict = {"experimental": None} + if extensions is not None: + caps_kwargs["extensions"] = extensions + params = types.InitializeRequestParams( + protocolVersion=types.LATEST_PROTOCOL_VERSION, + capabilities=types.ClientCapabilities(**caps_kwargs), + clientInfo=types.Implementation(name="test-host", version="0.0.1"), + ) + return types.ClientRequest(types.InitializeRequest(params=params)) + + +class TestMcpAppsExtensionSettings: + def test_mime_type_matches_spec(self): + settings = mcp_apps_extension_settings() + assert settings == {"mimeTypes": [MCP_APPS_MIME_TYPE]} + assert MCP_APPS_MIME_TYPE == "text/html;profile=mcp-app" + assert MCP_APPS_EXTENSION_ID == "io.modelcontextprotocol/ui" + + +class TestInjectUiExtensionIntoRequest: + def test_injects_extensions_on_initialize(self): + request = _make_initialize_request() + updated = inject_ui_extension_into_request(request) + + caps = updated.root.params.capabilities + extensions = getattr(caps, "extensions", None) + assert extensions is not None + assert extensions[MCP_APPS_EXTENSION_ID]["mimeTypes"] == [MCP_APPS_MIME_TYPE] + + dumped = updated.model_dump(exclude_none=True) + assert dumped["params"]["capabilities"]["extensions"][MCP_APPS_EXTENSION_ID][ + "mimeTypes" + ] == [MCP_APPS_MIME_TYPE] + + def test_idempotent_when_already_present(self): + request = _make_initialize_request( + extensions={ + MCP_APPS_EXTENSION_ID: mcp_apps_extension_settings(), + } + ) + updated = inject_ui_extension_into_request(request) + assert updated is request + + def test_preserves_other_extensions(self): + request = _make_initialize_request( + extensions={"com.example/other": {"enabled": True}} + ) + updated = inject_ui_extension_into_request(request) + extensions = updated.root.params.capabilities.extensions + assert extensions["com.example/other"] == {"enabled": True} + assert extensions[MCP_APPS_EXTENSION_ID]["mimeTypes"] == [MCP_APPS_MIME_TYPE] + + def test_ignores_non_initialize_requests(self): + request = types.ClientRequest(types.PingRequest()) + assert inject_ui_extension_into_request(request) is request + + +def _tool_with_meta(name: str, meta: dict | None = None) -> SimpleNamespace: + metadata: dict = {} + if meta is not None: + metadata["_meta"] = meta + return SimpleNamespace(name=name, metadata=metadata) + + +class TestToolUiMetaAndVisibility: + def test_no_meta_defaults_visibility(self): + tool = _tool_with_meta("plain") + assert get_tool_ui_meta(tool) == {} + assert get_tool_visibility(tool) == ["model", "app"] + assert is_model_visible(tool) is True + assert is_app_callable(tool) is True + + def test_nested_ui_meta(self): + tool = _tool_with_meta( + "chart", + { + "ui": { + "resourceUri": "ui://charts/app.html", + "visibility": ["model", "app"], + } + }, + ) + assert get_tool_ui_meta(tool)["resourceUri"] == "ui://charts/app.html" + assert is_model_visible(tool) is True + + def test_app_only_hidden_from_model(self): + tool = _tool_with_meta( + "refresh", + {"ui": {"resourceUri": "ui://x", "visibility": ["app"]}}, + ) + assert is_model_visible(tool) is False + assert is_app_callable(tool) is True + + def test_explicit_empty_visibility_is_not_default(self): + tool = _tool_with_meta( + "hidden", + {"ui": {"resourceUri": "ui://x", "visibility": []}}, + ) + assert get_tool_visibility(tool) == [] + assert is_model_visible(tool) is False + assert is_app_callable(tool) is False + + def test_deprecated_flat_resource_uri(self): + tool = _tool_with_meta( + "legacy", + {"ui/resourceUri": "ui://legacy/app.html"}, + ) + assert get_tool_ui_meta(tool) == {"resourceUri": "ui://legacy/app.html"} + assert is_model_visible(tool) is True + + def test_annotate_and_prepare_for_model(self): + model_tool = _tool_with_meta( + "show", + {"ui": {"resourceUri": "ui://show", "visibility": ["model", "app"]}}, + ) + app_only = _tool_with_meta( + "refresh", + {"ui": {"visibility": ["app"]}}, + ) + plain = _tool_with_meta("search") + + prepared = prepare_tools_for_model( + [model_tool, app_only, plain], "chart-mcp-server" + ) + names = [t.name for t in prepared] + assert names == ["show", "search"] + assert app_only.name not in names + assert model_tool.metadata["mcp_server"] == "chart-mcp-server" + assert plain.metadata["mcp_server"] == "chart-mcp-server" + # App-only tool is still annotated even though filtered out of model list + assert app_only.metadata["mcp_server"] == "chart-mcp-server" + + def test_annotate_preserves_existing_meta(self): + tool = _tool_with_meta( + "show", + {"ui": {"resourceUri": "ui://x"}}, + ) + annotate_mcp_tool(tool, "my-server") + assert tool.metadata["_meta"]["ui"]["resourceUri"] == "ui://x" + assert tool.metadata["mcp_server"] == "my-server" + + def test_build_descriptor_and_attach_to_result(self): + tool = _tool_with_meta( + "show_chart", + {"ui": {"resourceUri": "ui://charts/app.html"}}, + ) + annotate_mcp_tool(tool, "chart-mcp-server") + descriptor = build_mcp_app_descriptor(tool) + assert descriptor is not None + assert descriptor["server"] == "chart-mcp-server" + assert descriptor["resourceUri"] == "ui://charts/app.html" + + content, artifact = attach_mcp_app_to_tool_result( + ( + [{"type": "text", "text": "ok"}], + {"structured_content": {"rows": [1]}}, + ), + descriptor, + ) + assert content[0]["text"] == "ok" + assert artifact["mcp_app"]["server"] == "chart-mcp-server" + assert artifact["mcp_app"]["result"]["structuredContent"] == {"rows": [1]} + assert artifact["mcp_app"]["result"]["isError"] is False + + def test_attach_prefers_raw_mcp_result_shape(self): + tool = _tool_with_meta( + "show_chart", + {"ui": {"resourceUri": "ui://charts/app.html"}}, + ) + annotate_mcp_tool(tool, "chart-mcp-server") + descriptor = build_mcp_app_descriptor(tool) + assert descriptor is not None + + # LangChain-shaped content would use base64/mime_type; MCP uses data/mimeType. + lc_content = [{"type": "image", "base64": "YWJj", "mime_type": "image/png"}] + mcp_result = { + "content": [ + {"type": "image", "data": "YWJj", "mimeType": "image/png"}, + ], + "structuredContent": {"rows": [1]}, + "isError": False, + "_meta": {"source": "charts"}, + } + _content, artifact = attach_mcp_app_to_tool_result( + (lc_content, None), + descriptor, + mcp_result=mcp_result, + ) + app_result = artifact["mcp_app"]["result"] + assert app_result["content"] == mcp_result["content"] + assert app_result["structuredContent"] == {"rows": [1]} + assert app_result["_meta"] == {"source": "charts"} + assert app_result["isError"] is False + + @pytest.mark.asyncio + async def test_interceptor_captures_call_tool_result(self): + mcp_result = types.CallToolResult( + content=[ + types.TextContent(type="text", text="ok"), + types.ImageContent(type="image", data="YWJj", mimeType="image/png"), + ], + structuredContent={"k": 1}, + isError=False, + ) + + async def handler(_request): + return mcp_result + + interceptor = McpAppCallToolResultInterceptor() + out = await interceptor(SimpleNamespace(name="show"), handler) + assert out is mcp_result + captured = take_captured_call_tool_result() + assert captured is not None + assert captured["content"][0] == {"type": "text", "text": "ok"} + assert captured["content"][1]["mimeType"] == "image/png" + assert captured["content"][1]["data"] == "YWJj" + assert captured["structuredContent"] == {"k": 1} + assert take_captured_call_tool_result() is None + + @pytest.mark.asyncio + async def test_prepare_wraps_ui_tool_coroutine(self): + async def _coro(**kwargs): + return ("plain", {"structured_content": {"v": 1}}) + + tool = _tool_with_meta( + "show", + {"ui": {"resourceUri": "ui://x"}}, + ) + tool.coroutine = _coro + prepared = prepare_tools_for_model([tool], "srv") + assert len(prepared) == 1 + + capture_call_tool_result_for_app( + types.CallToolResult( + content=[types.TextContent(type="text", text="mcp-ok")], + structuredContent={"v": 1}, + isError=False, + ) + ) + _content, artifact = await prepared[0].coroutine(topic="sales") + assert artifact["mcp_app"]["server"] == "srv" + assert artifact["mcp_app"]["arguments"] == {"topic": "sales"} + assert artifact["mcp_app"]["result"]["content"] == [ + {"type": "text", "text": "mcp-ok"} + ] + assert artifact["mcp_app"]["result"]["structuredContent"] == {"v": 1} + + @pytest.mark.asyncio + async def test_wrap_attaches_mcp_app_on_tool_exception(self): + from langchain_core.tools import ToolException + + async def _coro(**kwargs): + capture_call_tool_result_for_app( + types.CallToolResult( + content=[types.TextContent(type="text", text="boom")], + structuredContent={"err": True}, + isError=True, + ) + ) + raise ToolException("boom") + + tool = _tool_with_meta( + "show", + {"ui": {"resourceUri": "ui://x"}}, + ) + tool.coroutine = _coro + prepared = prepare_tools_for_model([tool], "srv") + _content, artifact = await prepared[0].coroutine() + assert artifact["mcp_app"]["result"]["isError"] is True + assert artifact["mcp_app"]["result"]["content"] == [ + {"type": "text", "text": "boom"} + ] + assert artifact["mcp_app"]["result"]["structuredContent"] == {"err": True} + assert _content[0]["text"] == "boom" + + @pytest.mark.asyncio + async def test_wrap_attaches_mcp_app_on_audio_conversion_failure(self): + """langchain-mcp-adapters cannot convert AudioContent; App must still mount.""" + + async def _coro(**kwargs): + capture_call_tool_result_for_app( + types.CallToolResult( + content=[ + types.TextContent(type="text", text="lab ready"), + types.AudioContent( + type="audio", + data="UklGRg==", + mimeType="audio/wav", + ), + ], + structuredContent={"contentType": "audio"}, + isError=False, + ) + ) + raise NotImplementedError( + "AudioContent conversion to LangChain content blocks " + "is not yet supported. Received audio with mime type: audio/wav" + ) + + tool = _tool_with_meta( + "open_conformance_lab", + {"ui": {"resourceUri": "ui://lab/app.html"}}, + ) + tool.coroutine = _coro + prepared = prepare_tools_for_model([tool], "srv") + content, artifact = await prepared[0].coroutine(contentType="audio") + assert content[0]["text"] == "lab ready" + app_result = artifact["mcp_app"]["result"] + assert app_result["isError"] is False + assert app_result["structuredContent"] == {"contentType": "audio"} + assert app_result["content"][0] == {"type": "text", "text": "lab ready"} + assert app_result["content"][1]["type"] == "audio" + assert app_result["content"][1]["mimeType"] == "audio/wav" + + @pytest.mark.asyncio + async def test_interceptor_ignores_non_call_tool_result(self): + async def handler(_request): + return {"not": "a CallToolResult"} + + interceptor = McpAppCallToolResultInterceptor() + out = await interceptor(SimpleNamespace(name="x"), handler) + assert out == {"not": "a CallToolResult"} + assert take_captured_call_tool_result() is None + + def test_serialize_handles_dict_and_plain_blocks(self): + class _Meta: + def model_dump(self, **_kwargs): + return {"k": 1} + + result = SimpleNamespace( + content=[ + {"type": "text", "text": "dict-block"}, + 42, + ], + isError=False, + structuredContent={"ok": True}, + meta=_Meta(), + ) + serialized = serialize_call_tool_result_for_app(result) + assert serialized["content"][0] == {"type": "text", "text": "dict-block"} + assert serialized["content"][1] == {"type": "text", "text": "42"} + assert serialized["structuredContent"] == {"ok": True} + assert serialized["_meta"] == {"k": 1} + + def test_attach_non_tuple_uses_content_blocks(self): + descriptor = { + "server": "srv", + "resourceUri": "ui://x", + "toolName": "show", + "visibility": ["model", "app"], + } + _content, artifact = attach_mcp_app_to_tool_result( + "hello", + descriptor, + ) + assert _content == "hello" + assert artifact["mcp_app"]["result"]["content"] == [ + {"type": "text", "text": "hello"} + ] + + def test_attach_list_content_without_mcp_result(self): + descriptor = { + "server": "srv", + "resourceUri": "ui://x", + "toolName": "show", + "visibility": ["app"], + } + _content, artifact = attach_mcp_app_to_tool_result( + ([{"type": "text", "text": "a"}], {"structuredContent": {"n": 2}}), + descriptor, + ) + assert artifact["mcp_app"]["result"]["structuredContent"] == {"n": 2} + + def test_extract_mcp_app_from_additional_kwargs(self): + from langchain_core.messages import ToolMessage + + msg = ToolMessage( + content="ok", + tool_call_id="tc1", + name="show", + additional_kwargs={ + "mcpApp": { + "server": "srv", + "resourceUri": "ui://from-kwargs", + "result": {"content": [], "isError": False}, + } + }, + ) + assert extract_mcp_app_from_message(msg)["resourceUri"] == "ui://from-kwargs" + + def test_extract_mcp_app_returns_none_without_payload(self): + from langchain_core.messages import ToolMessage + + msg = ToolMessage(content="ok", tool_call_id="tc1", name="show") + assert extract_mcp_app_from_message(msg) is None + + def test_serialize_call_tool_result_for_app(self): + serialized = serialize_call_tool_result_for_app( + types.CallToolResult( + content=[types.TextContent(type="text", text="x")], + isError=True, + ) + ) + assert serialized["isError"] is True + assert serialized["content"] == [{"type": "text", "text": "x"}] + + def test_extract_mcp_app_from_tool_message(self): + from langchain_core.messages import ToolMessage + + msg = ToolMessage( + content="ok", + tool_call_id="tc1", + name="show", + artifact={ + "mcp_app": { + "server": "srv", + "resourceUri": "ui://x", + "result": {"content": [], "isError": False}, + } + }, + ) + assert extract_mcp_app_from_message(msg)["resourceUri"] == "ui://x" + + +class TestEnsureMcpAppsCapabilityAdvertised: + def test_install_is_idempotent(self): + # Module import of deep_agent.aegra.mcp may already have installed the patch. + ensure_mcp_apps_capability_advertised() + second = ensure_mcp_apps_capability_advertised() + assert second is False + assert mcp_apps._patch_installed is True + assert ClientSession.initialize is mcp_apps._initialize_with_mcp_apps + + @pytest.mark.asyncio + async def test_patched_initialize_sends_extensions(self): + ensure_mcp_apps_capability_advertised() + + captured: dict[str, types.ClientRequest] = {} + + async def fake_original_initialize( + self: ClientSession, + ) -> types.InitializeResult: + # Simulate stock initialize issuing an InitializeRequest via send_request. + req = _make_initialize_request() + await self.send_request(req, types.InitializeResult) + return types.InitializeResult( + protocolVersion=types.LATEST_PROTOCOL_VERSION, + capabilities=types.ServerCapabilities(), + serverInfo=types.Implementation(name="mock", version="0"), + ) + + async def fake_send_request(request, result_type, **kwargs): + captured["request"] = request + return types.InitializeResult( + protocolVersion=types.LATEST_PROTOCOL_VERSION, + capabilities=types.ServerCapabilities(), + serverInfo=types.Implementation(name="mock", version="0"), + ) + + session = object.__new__(ClientSession) + original_send = AsyncMock(side_effect=fake_send_request) + session.send_request = original_send + + previous = mcp_apps._original_initialize + mcp_apps._original_initialize = fake_original_initialize + try: + result = await ClientSession.initialize(session) + assert isinstance(result, types.InitializeResult) + assert "request" in captured + caps = captured["request"].root.params.capabilities + assert caps.extensions[MCP_APPS_EXTENSION_ID]["mimeTypes"] == [ + MCP_APPS_MIME_TYPE + ] + # send_request must be restored after initialize + assert session.send_request is original_send + finally: + mcp_apps._original_initialize = previous diff --git a/tests/unit/aegra/test_mcp_apps_smoke.py b/tests/unit/aegra/test_mcp_apps_smoke.py new file mode 100644 index 00000000..eaf90648 --- /dev/null +++ b/tests/unit/aegra/test_mcp_apps_smoke.py @@ -0,0 +1,212 @@ +"""Compliance smoke tests for the MCP Apps host (agent side). + +Locks the generic host contract: capability advertising, visibility filtering, +and request-scoped resources/list + resources/templates/list + resources/read + tools/call rules. +""" + +from __future__ import annotations + +from contextlib import asynccontextmanager +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import HTTPException +from langchain_core.tools import StructuredTool, ToolException +from mcp import types + +from deep_agent.aegra.mcp_apps import ( + MCP_APPS_EXTENSION_ID, + MCP_APPS_MIME_TYPE, + annotate_mcp_tool, + capture_call_tool_result_for_app, + ensure_mcp_apps_capability_advertised, + is_app_callable, + is_model_visible, + mcp_apps_extension_settings, + prepare_tools_for_model, +) +from deep_agent.aegra.mcp_host import ( + call_app_tool, + list_resource_templates, + read_resource, +) + + +def _tool(name: str, meta: dict | None = None) -> StructuredTool: + return StructuredTool.from_function( + func=lambda: "ok", + name=name, + description=name, + metadata={"_meta": meta} if meta is not None else {}, + ) + + +def _server_cfg(**overrides): + cfg = { + "url": "http://localhost:5003/mcp", + "transport": "streamable_http", + "enabled": True, + "auth": False, + "auth_mode": "sso", + "ssl_verify": False, + "timeout": 30, + } + cfg.update(overrides) + return cfg + + +@asynccontextmanager +async def _fake_session(session): + yield session + + +class TestMcpAppsSmokeCapability: + def test_extension_settings_match_sep_1865(self): + settings = mcp_apps_extension_settings() + assert settings["mimeTypes"] == [MCP_APPS_MIME_TYPE] + assert MCP_APPS_EXTENSION_ID == "io.modelcontextprotocol/ui" + assert MCP_APPS_MIME_TYPE == "text/html;profile=mcp-app" + + def test_capability_patch_is_installed(self): + ensure_mcp_apps_capability_advertised() + from mcp.client.session import ClientSession + from deep_agent.aegra import mcp_apps + + assert mcp_apps._patch_installed is True + assert ClientSession.initialize is mcp_apps._initialize_with_mcp_apps + + +class TestMcpAppsSmokeVisibility: + def test_app_only_tools_hidden_from_model(self): + model_tool = _tool( + "show", + {"ui": {"resourceUri": "ui://x", "visibility": ["model", "app"]}}, + ) + app_only = _tool("refresh", {"ui": {"visibility": ["app"]}}) + annotate_mcp_tool(model_tool, "charts") + annotate_mcp_tool(app_only, "charts") + + assert is_model_visible(model_tool) is True + assert is_model_visible(app_only) is False + assert is_app_callable(app_only) is True + + prepared = prepare_tools_for_model([model_tool, app_only], "charts") + assert [t.name for t in prepared] == ["show"] + assert prepared[0].metadata.get("mcp_server") == "charts" + + +class TestMcpAppsSmokeHostProxy: + @pytest.mark.asyncio + async def test_resources_read_rejects_empty_uri(self): + with pytest.raises(HTTPException) as exc: + await read_resource( + "charts", + "", + user_id="u1", + sso_token=None, + ) + assert exc.value.status_code == 400 + + @pytest.mark.asyncio + async def test_resource_templates_list_is_request_scoped(self): + session = MagicMock() + session.list_resource_templates = AsyncMock( + return_value=types.ListResourceTemplatesResult(resourceTemplates=[]) + ) + with ( + patch( + "deep_agent.aegra.mcp_host._get_server_configs", + return_value={"charts": _server_cfg()}, + ), + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "deep_agent.aegra.mcp_host.MultiServerMCPClient", + ) as mock_client_cls, + ): + client = MagicMock() + client.session = lambda _name: _fake_session(session) + mock_client_cls.return_value = client + result = await list_resource_templates( + "charts", + user_id="u1", + sso_token=None, + ) + session.list_resource_templates.assert_awaited_once_with(cursor=None) + assert ( + result.get("resourceTemplates") == [] + or result.get("resource_templates") == [] + ) + + @pytest.mark.asyncio + async def test_error_tool_result_still_embeds_mcp_app(self): + async def _coro(**_kwargs): + capture_call_tool_result_for_app( + types.CallToolResult( + content=[types.TextContent(type="text", text="failed")], + isError=True, + ) + ) + raise ToolException("failed") + + tool = _tool( + "show", + { + "ui": { + "resourceUri": "ui://charts/app.html", + "visibility": ["model", "app"], + } + }, + ) + tool.coroutine = _coro + prepared = prepare_tools_for_model([tool], "charts") + _content, artifact = await prepared[0].coroutine() + assert artifact["mcp_app"]["result"]["isError"] is True + assert artifact["mcp_app"]["result"]["content"][0]["text"] == "failed" + assert artifact["mcp_app"]["resourceUri"] == "ui://charts/app.html" + + @pytest.mark.asyncio + async def test_tools_call_rejects_model_only(self): + tool = types.Tool.model_validate( + { + "name": "model_only", + "inputSchema": {"type": "object"}, + "_meta": {"ui": {"visibility": ["model"]}}, + } + ) + session = MagicMock() + session.list_tools = AsyncMock(return_value=types.ListToolsResult(tools=[tool])) + session.call_tool = AsyncMock() + + with ( + patch( + "deep_agent.aegra.mcp_host._get_server_configs", + return_value={"charts": _server_cfg()}, + ), + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "deep_agent.aegra.mcp_host.MultiServerMCPClient", + ) as mock_client_cls, + pytest.raises(HTTPException) as exc, + ): + client = MagicMock() + client.session = lambda _name: _fake_session(session) + mock_client_cls.return_value = client + + await call_app_tool( + "charts", + "model_only", + {}, + user_id="u1", + sso_token=None, + ) + + assert exc.value.status_code == 403 + session.call_tool.assert_not_called() diff --git a/tests/unit/aegra/test_mcp_host.py b/tests/unit/aegra/test_mcp_host.py new file mode 100644 index 00000000..ca8294f8 --- /dev/null +++ b/tests/unit/aegra/test_mcp_host.py @@ -0,0 +1,1320 @@ +"""Unit tests for MCP Apps host proxy (resources/read + tools/call).""" + +from __future__ import annotations + +from contextlib import asynccontextmanager +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import HTTPException +from mcp import types + +from deep_agent.aegra.mcp_host import ( + _find_mcp_tool, + _resolve_bearer, + call_app_tool, + list_resource_templates, + list_resources, + list_tools, + read_resource, +) + + +def _server_cfg(**overrides): + cfg = { + "url": "http://localhost:5003/mcp", + "transport": "streamable_http", + "enabled": True, + "auth": False, + "auth_mode": "sso", + "ssl_verify": False, + "timeout": 30, + } + cfg.update(overrides) + return cfg + + +@asynccontextmanager +async def _fake_session(session): + yield session + + +class TestReadResource: + @pytest.mark.asyncio + async def test_rejects_empty_uri(self): + with pytest.raises(HTTPException) as exc: + await read_resource( + "charts", + "", + user_id="u1", + sso_token=None, + ) + assert exc.value.status_code == 400 + + @pytest.mark.asyncio + async def test_unknown_server_404(self): + with ( + patch( + "deep_agent.aegra.mcp_host._get_server_configs", + return_value={}, + ), + pytest.raises(HTTPException) as exc, + ): + await read_resource( + "missing", + "ui://charts/app.html", + user_id="u1", + sso_token=None, + ) + assert exc.value.status_code == 404 + + @pytest.mark.asyncio + async def test_reads_any_resource_uri(self): + content = types.TextResourceContents.model_validate( + { + "uri": "showcase://sample.json", + "mimeType": "application/json", + "text": '{"ok": true}', + } + ) + session = MagicMock() + session.read_resource = AsyncMock( + return_value=types.ReadResourceResult(contents=[content]) + ) + + with ( + patch( + "deep_agent.aegra.mcp_host._get_server_configs", + return_value={"charts": _server_cfg()}, + ), + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "deep_agent.aegra.mcp_host.MultiServerMCPClient", + ) as mock_client_cls, + ): + client = MagicMock() + client.session = lambda _name: _fake_session(session) + mock_client_cls.return_value = client + + result = await read_resource( + "charts", + "showcase://sample.json", + user_id="u1", + sso_token=None, + ) + + session.read_resource.assert_awaited_once() + assert result["contents"][0]["text"] == '{"ok": true}' + + +class TestListResources: + @pytest.mark.asyncio + async def test_lists_resources(self): + session = MagicMock() + session.list_resources = AsyncMock( + return_value=types.ListResourcesResult( + resources=[ + types.Resource.model_validate( + { + "uri": "showcase://sample.json", + "name": "sample", + "mimeType": "application/json", + } + ) + ] + ) + ) + + with ( + patch( + "deep_agent.aegra.mcp_host._get_server_configs", + return_value={"charts": _server_cfg()}, + ), + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "deep_agent.aegra.mcp_host.MultiServerMCPClient", + ) as mock_client_cls, + ): + client = MagicMock() + client.session = lambda _name: _fake_session(session) + mock_client_cls.return_value = client + + result = await list_resources( + "charts", + user_id="u1", + sso_token=None, + ) + + session.list_resources.assert_awaited_once_with(cursor=None) + assert result["resources"][0]["uri"] == "showcase://sample.json" + + +class TestListResourceTemplates: + @pytest.mark.asyncio + async def test_lists_resource_templates(self): + session = MagicMock() + session.list_resource_templates = AsyncMock( + return_value=types.ListResourceTemplatesResult( + resourceTemplates=[ + types.ResourceTemplate.model_validate( + { + "uriTemplate": "showcase://{id}", + "name": "sample_template", + } + ) + ] + ) + ) + + with ( + patch( + "deep_agent.aegra.mcp_host._get_server_configs", + return_value={"charts": _server_cfg()}, + ), + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "deep_agent.aegra.mcp_host.MultiServerMCPClient", + ) as mock_client_cls, + ): + client = MagicMock() + client.session = lambda _name: _fake_session(session) + mock_client_cls.return_value = client + + result = await list_resource_templates( + "charts", + user_id="u1", + sso_token=None, + ) + + session.list_resource_templates.assert_awaited_once_with(cursor=None) + templates = result.get("resourceTemplates") or result.get("resource_templates") + assert templates[0]["name"] == "sample_template" + + +class TestCallAppTool: + @pytest.mark.asyncio + async def test_rejects_model_only_tool(self): + tool = types.Tool.model_validate( + { + "name": "secret_admin", + "inputSchema": {"type": "object"}, + "_meta": {"ui": {"visibility": ["model"]}}, + } + ) + session = MagicMock() + session.list_tools = AsyncMock(return_value=types.ListToolsResult(tools=[tool])) + session.call_tool = AsyncMock() + + with ( + patch( + "deep_agent.aegra.mcp_host._get_server_configs", + return_value={"charts": _server_cfg()}, + ), + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "deep_agent.aegra.mcp_host.MultiServerMCPClient", + ) as mock_client_cls, + pytest.raises(HTTPException) as exc, + ): + client = MagicMock() + client.session = lambda _name: _fake_session(session) + mock_client_cls.return_value = client + + await call_app_tool( + "charts", + "secret_admin", + {}, + user_id="u1", + sso_token=None, + ) + + assert exc.value.status_code == 403 + assert exc.value.detail["error"] == "tool_not_app_callable" + assert exc.value.detail["tool"] == "secret_admin" + assert exc.value.detail["visibility"] == ["model"] + session.call_tool.assert_not_called() + + @pytest.mark.asyncio + async def test_calls_app_visible_tool(self): + tool = types.Tool.model_validate( + { + "name": "refresh_showcase", + "inputSchema": {"type": "object"}, + "_meta": { + "ui": {"visibility": ["app"], "resourceUri": "ui://x"}, + }, + } + ) + session = MagicMock() + session.list_tools = AsyncMock(return_value=types.ListToolsResult(tools=[tool])) + session.call_tool = AsyncMock( + return_value=types.CallToolResult( + content=[types.TextContent(type="text", text="ok")], + structuredContent={"status": "refreshed"}, + ) + ) + + with ( + patch( + "deep_agent.aegra.mcp_host._get_server_configs", + return_value={"charts": _server_cfg()}, + ), + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "deep_agent.aegra.mcp_host.MultiServerMCPClient", + ) as mock_client_cls, + ): + client = MagicMock() + client.session = lambda _name: _fake_session(session) + mock_client_cls.return_value = client + + result = await call_app_tool( + "charts", + "refresh_showcase", + {"topic": "demo"}, + user_id="u1", + sso_token=None, + ) + + session.call_tool.assert_awaited_once_with( + "refresh_showcase", {"topic": "demo"} + ) + assert result["structuredContent"]["status"] == "refreshed" + + @pytest.mark.asyncio + async def test_default_visibility_allows_app_call(self): + """Tools without visibility default to model+app and remain callable.""" + tool = types.Tool.model_validate( + { + "name": "show_chart", + "inputSchema": {"type": "object"}, + "_meta": {"ui": {"resourceUri": "ui://charts/app.html"}}, + } + ) + session = MagicMock() + session.list_tools = AsyncMock(return_value=types.ListToolsResult(tools=[tool])) + session.call_tool = AsyncMock( + return_value=types.CallToolResult( + content=[types.TextContent(type="text", text="ok")], + ) + ) + + with ( + patch( + "deep_agent.aegra.mcp_host._get_server_configs", + return_value={"charts": _server_cfg()}, + ), + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "deep_agent.aegra.mcp_host.MultiServerMCPClient", + ) as mock_client_cls, + ): + client = MagicMock() + client.session = lambda _name: _fake_session(session) + mock_client_cls.return_value = client + + result = await call_app_tool( + "charts", + "show_chart", + None, + user_id="u1", + sso_token=None, + ) + + session.call_tool.assert_awaited_once_with("show_chart", {}) + assert result["content"][0]["text"] == "ok" + + @pytest.mark.asyncio + async def test_oauth_missing_token_returns_401(self): + with ( + patch( + "deep_agent.aegra.mcp_host._get_server_configs", + return_value={ + "preset": _server_cfg(auth=True, auth_mode="oauth"), + }, + ), + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "deep_agent.aegra.mcp_auth.get_mcp_credential_resolver", + ) as mock_resolver, + pytest.raises(HTTPException) as exc, + ): + mock_resolver.return_value.connect_url.return_value = "/mcp/preset/connect" + await call_app_tool( + "preset", + "refresh", + {}, + user_id="u1", + sso_token=None, + ) + + assert exc.value.status_code == 401 + assert exc.value.detail["error"] == "authorization_required" + assert exc.value.detail["connect_url"] == "/mcp/preset/connect" + + +class TestRouteWiring: + """Smoke-test FastAPI route handlers parse bodies and delegate.""" + + @pytest.mark.asyncio + async def test_resources_read_route(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(return_value={"uri": "ui://x"}) + request.headers = {"authorization": "Bearer tok"} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_host.read_resource", + new_callable=AsyncMock, + return_value={"contents": []}, + ) as mock_read, + ): + response = await mcp_routes.mcp_resources_read("charts", request) + + assert response.status_code == 200 + mock_read.assert_awaited_once_with( + "charts", + "ui://x", + user_id="user-1", + sso_token="tok", + ) + + @pytest.mark.asyncio + async def test_resources_list_route(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(return_value={}) + request.headers = {"authorization": "Bearer tok"} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_host.list_resources", + new_callable=AsyncMock, + return_value={"resources": []}, + ) as mock_list, + ): + response = await mcp_routes.mcp_resources_list("charts", request) + + assert response.status_code == 200 + mock_list.assert_awaited_once_with( + "charts", + cursor=None, + user_id="user-1", + sso_token="tok", + ) + + @pytest.mark.asyncio + async def test_tools_list_route(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(return_value={}) + request.headers = {"authorization": "Bearer tok"} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_host.list_tools", + new_callable=AsyncMock, + return_value={ + "tools": [ + { + "name": "show_chart", + "inputSchema": {"type": "object"}, + } + ] + }, + ) as mock_list, + ): + response = await mcp_routes.mcp_tools_list("charts", request) + + assert response.status_code == 200 + mock_list.assert_awaited_once_with( + "charts", + cursor=None, + user_id="user-1", + sso_token="tok", + ) + + @pytest.mark.asyncio + async def test_tools_call_route_returns_dict_http_errors_as_json(self): + """Dict HTTPException.detail must not hit AgentProtocolError (string message).""" + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock( + return_value={"name": "hostile_model_only", "arguments": {}} + ) + request.headers = {"authorization": "Bearer tok"} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_host.call_app_tool", + new_callable=AsyncMock, + side_effect=HTTPException( + status_code=403, + detail={ + "error": "tool_not_app_callable", + "tool": "hostile_model_only", + "visibility": ["model"], + }, + ), + ), + ): + response = await mcp_routes.mcp_tools_call("mcp-app-test", request) + + assert response.status_code == 403 + assert response.body is not None + import json + + body = json.loads(response.body) + assert body["error"] == "tool_not_app_callable" + assert body["tool"] == "hostile_model_only" + + @pytest.mark.asyncio + async def test_resources_read_route_returns_dict_http_errors_as_json(self): + """authorization_required dict detail must return 401 JSON, not 500.""" + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(return_value={"uri": "ui://charts/app.html"}) + request.headers = {"authorization": "Bearer tok"} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_host.read_resource", + new_callable=AsyncMock, + side_effect=HTTPException( + status_code=401, + detail={ + "error": "authorization_required", + "mcp_name": "charts", + "connect_url": "/mcp/charts/connect", + }, + ), + ), + ): + response = await mcp_routes.mcp_resources_read("charts", request) + + assert response.status_code == 401 + import json + + body = json.loads(response.body) + assert body["error"] == "authorization_required" + assert body["mcp_name"] == "charts" + + @pytest.mark.asyncio + async def test_resources_list_route_returns_dict_http_errors_as_json(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(return_value={}) + request.headers = {"authorization": "Bearer tok"} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_host.list_resources", + new_callable=AsyncMock, + side_effect=HTTPException( + status_code=401, + detail={ + "error": "authorization_required", + "mcp_name": "charts", + "connect_url": "/mcp/charts/connect", + }, + ), + ), + ): + response = await mcp_routes.mcp_resources_list("charts", request) + + assert response.status_code == 401 + import json + + assert json.loads(response.body)["error"] == "authorization_required" + + @pytest.mark.asyncio + async def test_tools_call_route(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock( + return_value={"name": "refresh_showcase", "arguments": {"a": 1}} + ) + request.headers = {"authorization": "Bearer tok"} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_host.call_app_tool", + new_callable=AsyncMock, + return_value={"content": []}, + ) as mock_call, + ): + response = await mcp_routes.mcp_tools_call("charts", request) + + assert response.status_code == 200 + mock_call.assert_awaited_once_with( + "charts", + "refresh_showcase", + {"a": 1}, + user_id="user-1", + sso_token="tok", + ) + + @pytest.mark.asyncio + async def test_tools_list_empty_body_when_json_fails(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(side_effect=ValueError("empty")) + request.headers = {"authorization": "Bearer tok"} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_host.list_tools", + new_callable=AsyncMock, + return_value={"tools": []}, + ) as mock_list, + ): + response = await mcp_routes.mcp_tools_list("charts", request) + + assert response.status_code == 200 + mock_list.assert_awaited_once_with( + "charts", + cursor=None, + user_id="user-1", + sso_token="tok", + ) + + @pytest.mark.asyncio + async def test_tools_list_returns_dict_http_errors_as_json(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(return_value={}) + request.headers = {"authorization": "Bearer tok"} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_host.list_tools", + new_callable=AsyncMock, + side_effect=HTTPException( + status_code=401, + detail={"error": "authorization_required", "mcp_name": "charts"}, + ), + ), + ): + response = await mcp_routes.mcp_tools_list("charts", request) + + assert response.status_code == 401 + import json + + assert json.loads(response.body)["error"] == "authorization_required" + + @pytest.mark.asyncio + async def test_resources_list_rejects_bad_cursor_and_body(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.headers = {} + + with patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ): + request.json = AsyncMock(return_value=["x"]) + with pytest.raises(HTTPException) as exc: + await mcp_routes.mcp_resources_list("charts", request) + assert exc.value.status_code == 400 + + request.json = AsyncMock(return_value={"cursor": 1}) + with pytest.raises(HTTPException) as exc2: + await mcp_routes.mcp_resources_list("charts", request) + assert exc2.value.status_code == 400 + + @pytest.mark.asyncio + async def test_templates_list_validation_and_dict_error(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.headers = {"authorization": "Bearer tok"} + + with patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ): + request.json = AsyncMock(side_effect=ValueError("x")) + with patch( + "deep_agent.aegra.mcp_host.list_resource_templates", + new_callable=AsyncMock, + return_value={"resourceTemplates": []}, + ) as mock_list: + response = await mcp_routes.mcp_resource_templates_list( + "charts", request + ) + assert response.status_code == 200 + mock_list.assert_awaited_once() + + request.json = AsyncMock(return_value=["bad"]) + with pytest.raises(HTTPException): + await mcp_routes.mcp_resource_templates_list("charts", request) + + request.json = AsyncMock(return_value={"cursor": False}) + with pytest.raises(HTTPException): + await mcp_routes.mcp_resource_templates_list("charts", request) + + request.json = AsyncMock(return_value={}) + with patch( + "deep_agent.aegra.mcp_host.list_resource_templates", + new_callable=AsyncMock, + side_effect=HTTPException( + status_code=401, + detail={"error": "authorization_required"}, + ), + ): + err = await mcp_routes.mcp_resource_templates_list("charts", request) + assert err.status_code == 401 + + @pytest.mark.asyncio + async def test_resources_read_invalid_json_and_string_detail(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.headers = {} + + with patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ): + request.json = AsyncMock(side_effect=ValueError("bad")) + with pytest.raises(HTTPException) as exc: + await mcp_routes.mcp_resources_read("charts", request) + assert exc.value.status_code == 400 + + request.json = AsyncMock(return_value={"uri": "ui://x"}) + with ( + patch( + "deep_agent.aegra.mcp_host.read_resource", + new_callable=AsyncMock, + side_effect=HTTPException(status_code=404, detail="missing"), + ), + pytest.raises(HTTPException) as exc2, + ): + await mcp_routes.mcp_resources_read("charts", request) + assert exc2.value.detail == "missing" + + @pytest.mark.asyncio + async def test_tools_call_defaults_arguments_and_rejects_non_object(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.headers = {"authorization": "Bearer tok"} + + with patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ): + request.json = AsyncMock(return_value={"name": "t"}) + with patch( + "deep_agent.aegra.mcp_host.call_app_tool", + new_callable=AsyncMock, + return_value={"content": []}, + ) as mock_call: + response = await mcp_routes.mcp_tools_call("charts", request) + assert response.status_code == 200 + mock_call.assert_awaited_once_with( + "charts", + "t", + {}, + user_id="user-1", + sso_token="tok", + ) + + request.json = AsyncMock(return_value=["nope"]) + with pytest.raises(HTTPException) as exc: + await mcp_routes.mcp_tools_call("charts", request) + assert exc.value.status_code == 400 + + @pytest.mark.asyncio + async def test_resource_templates_list_route(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(return_value={"cursor": "c1"}) + request.headers = {"authorization": "Bearer tok"} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_host.list_resource_templates", + new_callable=AsyncMock, + return_value={"resourceTemplates": []}, + ) as mock_list, + ): + response = await mcp_routes.mcp_resource_templates_list("charts", request) + + assert response.status_code == 200 + mock_list.assert_awaited_once_with( + "charts", + cursor="c1", + user_id="user-1", + sso_token="tok", + ) + + @pytest.mark.asyncio + async def test_tools_list_rejects_non_object_body(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(return_value=["not", "an", "object"]) + request.headers = {} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + pytest.raises(HTTPException) as exc, + ): + await mcp_routes.mcp_tools_list("charts", request) + + assert exc.value.status_code == 400 + assert "object" in str(exc.value.detail) + + @pytest.mark.asyncio + async def test_tools_list_rejects_non_string_cursor(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(return_value={"cursor": 123}) + request.headers = {} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + pytest.raises(HTTPException) as exc, + ): + await mcp_routes.mcp_tools_list("charts", request) + + assert exc.value.status_code == 400 + + @pytest.mark.asyncio + async def test_tools_call_rejects_invalid_json(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(side_effect=ValueError("bad json")) + request.headers = {} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + pytest.raises(HTTPException) as exc, + ): + await mcp_routes.mcp_tools_call("charts", request) + + assert exc.value.status_code == 400 + + @pytest.mark.asyncio + async def test_tools_call_rejects_non_object_arguments(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(return_value={"name": "t", "arguments": "nope"}) + request.headers = {} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + pytest.raises(HTTPException) as exc, + ): + await mcp_routes.mcp_tools_call("charts", request) + + assert exc.value.status_code == 400 + + @pytest.mark.asyncio + async def test_tools_call_reraises_string_http_detail(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(return_value={"name": "t", "arguments": {}}) + request.headers = {"authorization": "Bearer tok"} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_host.call_app_tool", + new_callable=AsyncMock, + side_effect=HTTPException(status_code=404, detail="Tool not found"), + ), + pytest.raises(HTTPException) as exc, + ): + await mcp_routes.mcp_tools_call("charts", request) + + assert exc.value.status_code == 404 + assert exc.value.detail == "Tool not found" + + @pytest.mark.asyncio + async def test_resources_read_rejects_non_object_body(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.json = AsyncMock(return_value="uri://x") + request.headers = {} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + pytest.raises(HTTPException) as exc, + ): + await mcp_routes.mcp_resources_read("charts", request) + + assert exc.value.status_code == 400 + + @pytest.mark.asyncio + async def test_mcp_connect_route(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.headers = {} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_oauth_handlers.handle_mcp_connect", + new_callable=AsyncMock, + return_value={"authorization_url": "https://idp/auth"}, + ) as mock_connect, + ): + response = await mcp_routes.mcp_connect("charts", request) + + assert response.status_code == 200 + mock_connect.assert_awaited_once_with("user-1", "charts", caller_origin=None) + + @pytest.mark.asyncio + async def test_mcp_status_route(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.headers = {} + + with ( + patch( + "deep_agent.aegra.mcp_routes._authenticated_user_id", + new_callable=AsyncMock, + return_value="user-1", + ), + patch( + "deep_agent.aegra.mcp_oauth_handlers.handle_mcp_status", + new_callable=AsyncMock, + return_value={"connected": True}, + ) as mock_status, + ): + response = await mcp_routes.mcp_status("charts", request) + + assert response.status_code == 200 + mock_status.assert_awaited_once_with("user-1", "charts") + + @pytest.mark.asyncio + async def test_mcp_oauth_callback_route(self): + from deep_agent.aegra import mcp_routes + from fastapi.responses import HTMLResponse + + request = MagicMock() + html = HTMLResponse("ok") + + with patch( + "deep_agent.aegra.mcp_oauth_handlers.handle_mcp_oauth_callback", + new_callable=AsyncMock, + return_value=html, + ) as mock_cb: + response = await mcp_routes.mcp_oauth_callback(request, code="c", state="s") + + assert response is html + mock_cb.assert_awaited_once_with("c", "s", request) + + @pytest.mark.asyncio + async def test_get_agent_info(self): + from deep_agent.aegra import mcp_routes + + with ( + patch( + "deep_agent.aegra.mcp_routes.agent_config.get_mcp_servers", + return_value={ + "charts": {"enabled": True, "auth_mode": "oauth"}, + "off": {"enabled": False, "auth_mode": "oauth"}, + "sso": {"enabled": True, "auth_mode": "sso"}, + "dcr": {"enabled": True, "auth_mode": "dcr"}, + }, + ), + patch( + "deep_agent.aegra.mcp_routes.agent_config.get_name", + return_value="demo-agent", + ), + ): + info = await mcp_routes.get_agent_info() + + assert info["name"] == "demo-agent" + assert info["oauth_mcps"] == ["charts", "dcr"] + + +class TestRouteAuthHelpers: + @pytest.mark.asyncio + async def test_authenticated_user_id_bypasses_when_auth_disabled(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.headers = {} + + with ( + patch("deep_agent.aegra.auth.ENABLE_AUTH", False), + patch("deep_agent.aegra.auth.ENVIRONMENT", "development"), + patch("deep_agent.aegra.auth.DEV_USER_ID", "dev-user"), + ): + user_id = await mcp_routes._authenticated_user_id(request) + + assert user_id == "dev-user" + + @pytest.mark.asyncio + async def test_authenticated_user_id_blocks_prod_bypass(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.headers = {} + + with ( + patch("deep_agent.aegra.auth.ENABLE_AUTH", False), + patch("deep_agent.aegra.auth.ENVIRONMENT", "production"), + pytest.raises(HTTPException) as exc, + ): + await mcp_routes._authenticated_user_id(request) + + assert exc.value.status_code == 500 + + @pytest.mark.asyncio + async def test_authenticated_user_id_requires_bearer_when_auth_on(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.headers = {} + + with ( + patch("deep_agent.aegra.auth.ENABLE_AUTH", True), + patch("deep_agent.aegra.auth.ENVIRONMENT", "development"), + pytest.raises(HTTPException) as exc, + ): + await mcp_routes._authenticated_user_id(request) + + assert exc.value.status_code == 401 + + @pytest.mark.asyncio + async def test_authenticated_user_id_decodes_bearer(self): + from deep_agent.aegra import mcp_routes + + request = MagicMock() + request.headers = {"authorization": "Bearer abc.def"} + + with ( + patch("deep_agent.aegra.auth.ENABLE_AUTH", True), + patch("deep_agent.aegra.auth.ENVIRONMENT", "development"), + patch( + "deep_agent.aegra.auth._decode_token", + return_value={"sub": "user-42"}, + ), + ): + user_id = await mcp_routes._authenticated_user_id(request) + + assert user_id == "user-42" + + def test_sso_bearer_from_request(self): + from deep_agent.aegra import mcp_routes + + with_tok = MagicMock() + with_tok.headers = {"authorization": "Bearer secret"} + empty = MagicMock() + empty.headers = {"authorization": "Bearer "} + missing = MagicMock() + missing.headers = {} + + assert mcp_routes._sso_bearer_from_request(with_tok) == "secret" + assert mcp_routes._sso_bearer_from_request(empty) is None + assert mcp_routes._sso_bearer_from_request(missing) is None + + def test_http_exception_response_only_for_dict_detail(self): + from deep_agent.aegra import mcp_routes + + as_json = mcp_routes._http_exception_response( + HTTPException(status_code=401, detail={"error": "authorization_required"}) + ) + assert as_json is not None + assert as_json.status_code == 401 + + assert ( + mcp_routes._http_exception_response( + HTTPException(status_code=400, detail="plain") + ) + is None + ) + + +class TestResolveBearer: + @pytest.mark.asyncio + async def test_needs_authorization_maps_to_401(self): + from deep_agent.aegra.mcp_auth import NeedsAuthorization + + with ( + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + side_effect=NeedsAuthorization("charts", "/mcp/charts/connect"), + ), + patch( + "deep_agent.aegra.mcp_auth.get_mcp_credential_resolver", + ) as mock_resolver, + pytest.raises(HTTPException) as exc, + ): + mock_resolver.return_value.connect_url.return_value = "/mcp/charts/connect" + await _resolve_bearer( + "charts", + _server_cfg(auth=True, auth_mode="oauth"), + user_id="u1", + sso_token=None, + ) + + assert exc.value.status_code == 401 + assert exc.value.detail["error"] == "authorization_required" + + @pytest.mark.asyncio + async def test_auth_not_required_returns_bearer(self): + with patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ): + bearer = await _resolve_bearer( + "charts", + _server_cfg(auth=False), + user_id="u1", + sso_token=None, + ) + assert bearer is None + + @pytest.mark.asyncio + async def test_sso_missing_bearer_raises(self): + with ( + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ), + pytest.raises(HTTPException) as exc, + ): + await _resolve_bearer( + "charts", + _server_cfg(auth=True, auth_mode="sso"), + user_id="u1", + sso_token=None, + ) + assert exc.value.status_code == 401 + assert "SSO" in str(exc.value.detail) + + @pytest.mark.asyncio + async def test_api_key_missing_raises_500(self): + with ( + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ), + pytest.raises(HTTPException) as exc, + ): + await _resolve_bearer( + "charts", + _server_cfg(auth=True, auth_mode="api_key"), + user_id="u1", + sso_token=None, + ) + assert exc.value.status_code == 500 + + +class TestListToolsAndFindTool: + @pytest.mark.asyncio + async def test_list_tools(self): + tool = types.Tool( + name="show_chart", + inputSchema={"type": "object"}, + ) + session = MagicMock() + session.list_tools = AsyncMock(return_value=types.ListToolsResult(tools=[tool])) + + with ( + patch( + "deep_agent.aegra.mcp_host._get_server_configs", + return_value={"charts": _server_cfg()}, + ), + patch( + "deep_agent.aegra.mcp_host._resolve_connection_token", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "deep_agent.aegra.mcp_host.MultiServerMCPClient", + ) as mock_client_cls, + ): + client = MagicMock() + client.session = lambda _name: _fake_session(session) + mock_client_cls.return_value = client + + result = await list_tools( + "charts", + cursor=None, + user_id="u1", + sso_token=None, + ) + + assert result["tools"][0]["name"] == "show_chart" + session.list_tools.assert_awaited_once_with(cursor=None) + + @pytest.mark.asyncio + async def test_find_mcp_tool_paginates(self): + page1 = types.ListToolsResult( + tools=[types.Tool(name="a", inputSchema={"type": "object"})], + nextCursor="page2", + ) + page2 = types.ListToolsResult( + tools=[types.Tool(name="target", inputSchema={"type": "object"})], + ) + session = MagicMock() + session.list_tools = AsyncMock(side_effect=[page1, page2]) + + found = await _find_mcp_tool(session, "target") + assert found is not None + assert found.name == "target" + assert session.list_tools.await_count == 2 + + @pytest.mark.asyncio + async def test_find_mcp_tool_returns_none(self): + session = MagicMock() + session.list_tools = AsyncMock( + return_value=types.ListToolsResult( + tools=[types.Tool(name="other", inputSchema={"type": "object"})] + ) + ) + assert await _find_mcp_tool(session, "missing") is None + + @pytest.mark.asyncio + async def test_find_mcp_tool_stops_after_max_pages(self): + from deep_agent.aegra import mcp_host + + endless = types.ListToolsResult( + tools=[types.Tool(name="other", inputSchema={"type": "object"})], + nextCursor="again", + ) + session = MagicMock() + session.list_tools = AsyncMock(return_value=endless) + + assert await _find_mcp_tool(session, "missing") is None + assert session.list_tools.await_count == mcp_host._MAX_TOOLS_LIST_PAGES diff --git a/tests/unit/aegra/test_serialization.py b/tests/unit/aegra/test_serialization.py index ddab13b4..23c924f3 100644 --- a/tests/unit/aegra/test_serialization.py +++ b/tests/unit/aegra/test_serialization.py @@ -52,6 +52,33 @@ def test_tool_message(self): assert result["tool_call_id"] == "tc1" assert result["name"] == "search" + def test_tool_message_preserves_artifact_and_mcp_app(self): + msg = ToolMessage( + content="ok", + tool_call_id="tc1", + name="show_chart", + id="t1", + artifact={ + "structured_content": {"n": 1}, + "mcp_app": { + "server": "chart-mcp-server", + "resourceUri": "ui://charts/app.html", + "result": { + "content": [{"type": "text", "text": "ok"}], + "structuredContent": {"n": 1}, + "isError": False, + }, + }, + }, + ) + result = serialize_message(msg) + assert result["artifact"]["structured_content"]["n"] == 1 + assert result["mcpApp"]["resourceUri"] == "ui://charts/app.html" + assert result["mcpApp"]["server"] == "chart-mcp-server" + assert result["mcpApp"]["result"]["isError"] is False + assert result["mcpApp"]["result"]["structuredContent"] == {"n": 1} + assert result["mcpApp"]["result"]["content"] == [{"type": "text", "text": "ok"}] + def test_system_message(self): msg = SystemMessage(content="you are helpful") result = serialize_message(msg) @@ -104,6 +131,62 @@ def test_tool_message(self): assert isinstance(msg, ToolMessage) assert msg.tool_call_id == "tc1" + def test_tool_message_with_artifact(self): + data = { + "type": "tool", + "content": "ok", + "tool_call_id": "tc1", + "name": "show", + "artifact": {"structured_content": {"n": 1}}, + } + msg = deserialize_message(data) + assert isinstance(msg, ToolMessage) + assert msg.artifact["structured_content"]["n"] == 1 + + def test_tool_message_mcp_app_without_artifact(self): + data = { + "type": "tool", + "content": "ok", + "tool_call_id": "tc1", + "name": "show", + "mcpApp": { + "server": "srv", + "resourceUri": "ui://x", + "result": {"content": [], "isError": False}, + }, + } + msg = deserialize_message(data) + assert isinstance(msg, ToolMessage) + assert msg.additional_kwargs["mcpApp"]["resourceUri"] == "ui://x" + assert msg.additional_kwargs["mcpApp"]["result"]["isError"] is False + + def test_tool_message_merges_mcp_app_into_existing_additional_kwargs(self): + data = { + "type": "tool", + "content": "ok", + "tool_call_id": "tc1", + "name": "show", + "additional_kwargs": {"trace": "t1"}, + "mcpApp": { + "server": "srv", + "resourceUri": "ui://merged", + "result": {"content": [], "isError": False}, + }, + } + msg = deserialize_message(data) + assert msg.additional_kwargs["trace"] == "t1" + assert msg.additional_kwargs["mcpApp"]["resourceUri"] == "ui://merged" + + def test_human_message_roundtrips_additional_kwargs(self): + original = HumanMessage( + content="retry", + id="h1", + additional_kwargs={"opa_retry": True}, + ) + restored = deserialize_message(serialize_message(original)) + assert isinstance(restored, HumanMessage) + assert restored.additional_kwargs["opa_retry"] is True + def test_unknown_type_defaults_to_human(self): data = {"type": "unknown_type", "content": "fallback"} msg = deserialize_message(data) diff --git a/tests/unit/aegra/test_startup.py b/tests/unit/aegra/test_startup.py index 4a6abd33..d0dceb75 100644 --- a/tests/unit/aegra/test_startup.py +++ b/tests/unit/aegra/test_startup.py @@ -166,6 +166,31 @@ def test_failure(self): assert "warning" in result +class TestSetupMcpAppsCapability: + def test_ok_when_newly_installed(self): + with patch( + "deep_agent.aegra.mcp_apps.ensure_mcp_apps_capability_advertised", + return_value=True, + ): + assert startup._setup_mcp_apps_capability() == "ok" + + def test_already_installed(self): + with patch( + "deep_agent.aegra.mcp_apps.ensure_mcp_apps_capability_advertised", + return_value=False, + ): + assert startup._setup_mcp_apps_capability() == "already_installed" + + def test_error(self): + with patch( + "deep_agent.aegra.mcp_apps.ensure_mcp_apps_capability_advertised", + side_effect=RuntimeError("boom"), + ): + result = startup._setup_mcp_apps_capability() + assert result.startswith("error:") + assert "boom" in result + + class TestIsReady: def test_not_ready_initially(self): startup._startup_complete = False diff --git a/tests/unit/infrastructure/test_mcp.py b/tests/unit/infrastructure/test_mcp.py index 7132734b..7a99d416 100644 --- a/tests/unit/infrastructure/test_mcp.py +++ b/tests/unit/infrastructure/test_mcp.py @@ -144,6 +144,7 @@ async def test_successful_connection(self): """Test successful connection to MCP server.""" mock_tool = MagicMock() mock_tool.name = "test_tool" + mock_tool.metadata = None mock_client = MagicMock() mock_client.get_tools = AsyncMock(return_value=[mock_tool]) @@ -154,10 +155,59 @@ async def test_successful_connection(self): "deep_agent.aegra.mcp.MultiServerMCPClient", return_value=mock_client, ): - tools = await _connect_single_server("test_server", config, {}, timeout=5) + tools = await _connect_single_server( + "test_server", + config, + {}, + timeout=5, + mcp_server="test_server", + ) assert len(tools) == 1 assert tools[0].name == "test_tool" + assert tools[0].metadata["mcp_server"] == "test_server" + + @pytest.mark.asyncio + async def test_filters_app_only_tools_from_model_list(self): + """App-only tools are annotated but not returned for the LLM.""" + from types import SimpleNamespace + + model_tool = SimpleNamespace( + name="show_chart", + metadata={ + "_meta": { + "ui": { + "resourceUri": "ui://charts/app.html", + "visibility": ["model", "app"], + } + } + }, + ) + app_only = SimpleNamespace( + name="refresh_chart", + metadata={"_meta": {"ui": {"visibility": ["app"]}}}, + ) + + mock_client = MagicMock() + mock_client.get_tools = AsyncMock(return_value=[model_tool, app_only]) + + config = {"url": "http://localhost:8000/mcp/", "transport": "http"} + + with patch( + "deep_agent.aegra.mcp.MultiServerMCPClient", + return_value=mock_client, + ): + tools = await _connect_single_server( + "charts", + config, + {}, + timeout=5, + mcp_server="chart-mcp-server", + ) + + assert [t.name for t in tools] == ["show_chart"] + assert tools[0].metadata["mcp_server"] == "chart-mcp-server" + assert app_only.metadata["mcp_server"] == "chart-mcp-server" @pytest.mark.asyncio async def test_connection_timeout_returns_empty_list(self):