feat(mcp): preserve MCP Apps result metadata

nanobot/agent/tools/base.py

@@ -145,10 +145,18 @@ class ToolResult(str):
     """String-compatible tool output with structured status."""
 
     is_error: bool
-
-    def __new__(cls, content: str, *, is_error: bool = False) -> ToolResult:
+    data: dict[str, Any] | None
+
+    def __new__(
+        cls,
+        content: str,
+        *,
+        is_error: bool = False,
+        data: dict[str, Any] | None = None,
+    ) -> ToolResult:
         obj = str.__new__(cls, content)
         obj.is_error = is_error
+        obj.data = data
         return obj
 
     @classmethod

nanobot/agent/tools/mcp.py

@@ -57,6 +57,7 @@
 _ReconnectCallback = Callable[[str, str, Tool], Awaitable[Tool | None]]
 MCPServerLoader = Callable[[], Mapping[str, "MCPServerConfig"]]
 MCPRuntimeStatus = Literal["connecting", "connected", "failed"]
+_MCP_APP_RESULT_KIND = "mcp_app_result"
 
 
 class MCPConnection(Protocol):
@@ -122,6 +123,89 @@ def _payload_value(payload: Any, key: str) -> Any:
     return getattr(payload, key, None)
 
 
+def _mcp_json_value(value: Any) -> Any:
+    """Convert SDK models to JSON-compatible values without trusting their shape."""
+    if value is None or isinstance(value, (str, int, float, bool)):
+        return value
+    model_dump = getattr(value, "model_dump", None)
+    if callable(model_dump):
+        return _mcp_json_value(model_dump(by_alias=True, exclude_none=True))
+    if isinstance(value, Mapping):
+        return {
+            str(key): _mcp_json_value(item)
+            for key, item in cast(Mapping[Any, Any], value).items()
+        }
+    if isinstance(value, (list, tuple)):
+        return [_mcp_json_value(item) for item in cast(Iterable[Any], value)]
+    try:
+        attributes = cast(dict[str, Any], vars(value))
+    except TypeError:
+        return str(value)
+    return {
+        str(key): _mcp_json_value(item)
+        for key, item in attributes.items()
+        if not str(key).startswith("_")
+    }
+
+
+def _mcp_tool_meta(tool_def: Any) -> dict[str, Any]:
+    raw = getattr(tool_def, "meta", None)
+    if raw is None:
+        raw = getattr(tool_def, "_meta", None)
+    value = _mcp_json_value(raw)
+    return cast(dict[str, Any], value) if isinstance(value, dict) else {}
+
+
+def _mcp_app_ui(tool_def: Any) -> dict[str, Any] | None:
+    """Normalize nested and slash-delimited MCP Apps tool metadata."""
+    meta = _mcp_tool_meta(tool_def)
+    nested = meta.get("ui")
+    ui = dict(cast(dict[str, Any], nested)) if isinstance(nested, dict) else {}
+    for field in ("resourceUri", "visibility"):
+        legacy = meta.get(f"ui/{field}")
+        if field not in ui and legacy is not None:
+            ui[field] = legacy
+    return ui or None
+
+
+def _mcp_tool_is_app_only(tool_def: Any) -> bool:
+    ui = _mcp_app_ui(tool_def)
+    if ui is None:
+        return False
+    raw_visibility = ui.get("visibility")
+    if not isinstance(raw_visibility, list):
+        return False
+    visibility = {
+        str(item).strip().lower()
+        for item in cast(list[Any], raw_visibility)
+        if str(item).strip()
+    }
+    return "app" in visibility and "model" not in visibility
+
+
+def _mcp_app_tool_data(server_name: str, tool_def: Any) -> dict[str, Any] | None:
+    ui = _mcp_app_ui(tool_def)
+    if ui is None:
+        return None
+    data: dict[str, Any] = {
+        "server": server_name,
+        "name": str(tool_def.name),
+        "ui": ui,
+    }
+    for output_key, attribute in (
+        ("outputSchema", "outputSchema"),
+        ("annotations", "annotations"),
+        ("_meta", "meta"),
+    ):
+        raw = getattr(tool_def, attribute, None)
+        if raw is None and attribute == "meta":
+            raw = getattr(tool_def, "_meta", None)
+        value = _mcp_json_value(raw)
+        if value is not None:
+            data[output_key] = value
+    return data
+
+
 def _progress_params_have_token(params: Any) -> bool:
     if isinstance(params, Mapping):
         return "progressToken" in params
@@ -611,6 +695,7 @@ def __init__(
         raw_schema = tool_def.inputSchema or {"type": "object", "properties": {}}
         self._parameters = _normalize_schema_for_openai(raw_schema)
         self._tool_timeout = tool_timeout
+        self._mcp_app_tool = _mcp_app_tool_data(server_name, tool_def)
 
     @property
     def name(self) -> str:
@@ -687,7 +772,11 @@ async def execute(self, **kwargs: Any) -> str:
                 # Success — extract text and persist any image content as artifacts.
                 try:
                     rendered = self._render_call_result(result.content, kwargs)
-                    if getattr(result, "isError", False):
+                    is_error = bool(getattr(result, "isError", False))
+                    data = self._mcp_app_result_data(result)
+                    if data is not None:
+                        return ToolResult(rendered, is_error=is_error, data=data)
+                    if is_error:
                         return ToolResult.error(rendered)
                     return rendered
                 except Exception as exc:
@@ -701,6 +790,26 @@ async def execute(self, **kwargs: Any) -> str:
                         f"(MCP tool returned malformed content: {type(exc).__name__})"
                     )
 
+    def _mcp_app_result_data(self, result: Any) -> dict[str, Any] | None:
+        """Preserve MCP Apps fields outside the model-facing result string."""
+        if self._mcp_app_tool is None:
+            return None
+        result_meta = getattr(result, "meta", None)
+        if result_meta is None:
+            result_meta = getattr(result, "_meta", None)
+        return {
+            "kind": _MCP_APP_RESULT_KIND,
+            "tool": self._mcp_app_tool,
+            "result": {
+                "content": _mcp_json_value(getattr(result, "content", [])),
+                "structuredContent": _mcp_json_value(
+                    getattr(result, "structuredContent", None)
+                ),
+                "_meta": _mcp_json_value(result_meta),
+                "isError": bool(getattr(result, "isError", False)),
+            },
+        }
+
     def _render_call_result(self, content: Any, arguments: Mapping[str, Any]) -> str:
         """Turn MCP content blocks into a tool result string.
 
@@ -1142,9 +1251,23 @@ def httpx_client_factory(
             allow_all_tools = "*" in enabled_tools
             registered_count = 0
             matched_enabled_tools: set[str] = set()
-            available_raw_names = [tool_def.name for tool_def in tools.tools]
-            available_wrapped_names = [_sanitize_mcp_tool_name(f"mcp_{name}_{tool_def.name}") for tool_def in tools.tools]
-            for tool_def in tools.tools:
+            model_tools: list["MCPToolDefinition"] = []
+            tool_definitions: list["MCPToolDefinition"] = tools.tools
+            for tool_def in tool_definitions:
+                if _mcp_tool_is_app_only(tool_def):
+                    logger.debug(
+                        "MCP: skipping app-only tool '{}' from server '{}'",
+                        tool_def.name,
+                        name,
+                    )
+                    continue
+                model_tools.append(tool_def)
+            available_raw_names = [tool_def.name for tool_def in model_tools]
+            available_wrapped_names = [
+                _sanitize_mcp_tool_name(f"mcp_{name}_{tool_def.name}")
+                for tool_def in model_tools
+            ]
+            for tool_def in model_tools:
                 wrapped_name = _sanitize_mcp_tool_name(f"mcp_{name}_{tool_def.name}")
                 if (
                     not allow_all_tools

nanobot/utils/progress_events.py

@@ -7,6 +7,7 @@
 from typing import Any, cast
 
 from nanobot.agent.hook import AgentHookContext
+from nanobot.agent.tools.base import ToolResult
 
 
 def on_progress_accepts_tool_events(cb: Callable[..., Any]) -> bool:
@@ -79,6 +80,13 @@ def tool_event_result_extras(result: Any) -> tuple[list[Any], list[Any]]:
     return files, embeds
 
 
+def tool_event_result_data(result: Any) -> dict[str, Any] | None:
+    """Return structured UI data without changing the model-facing tool result."""
+    if isinstance(result, ToolResult):
+        return result.data
+    return None
+
+
 def build_tool_event_finish_payloads(context: AgentHookContext) -> list[dict[str, Any]]:
     payloads: list[dict[str, Any]] = []
     count = min(len(context.tool_calls), len(context.tool_results), len(context.tool_events))
@@ -89,6 +97,7 @@ def build_tool_event_finish_payloads(context: AgentHookContext) -> list[dict[str
         status = event.get("status")
         phase = "end" if status == "ok" else "error"
         files, embeds = tool_event_result_extras(result)
+        data = tool_event_result_data(result)
         payload = {
             "version": 1,
             "phase": phase,
@@ -100,6 +109,8 @@ def build_tool_event_finish_payloads(context: AgentHookContext) -> list[dict[str
             "files": files,
             "embeds": embeds,
         }
+        if data is not None:
+            payload["data"] = data
         if phase == "error":
             if isinstance(result, str) and result.strip():
                 payload["error"] = result.strip()

tests/tools/test_mcp_tool.py

@@ -11,6 +11,7 @@
 import pytest
 
 import nanobot.agent.tools.mcp as mcp_mod
+from nanobot.agent.tools.base import ToolResult
 from nanobot.agent.tools.mcp import (
     MCPPromptWrapper,
     MCPProvider,
@@ -144,11 +145,19 @@ def __init__(self, code: int = -1, message: str = "error"):
     monkeypatch.setitem(sys.modules, "mcp.shared.exceptions", exc_mod)
 
 
-def _make_wrapper(session: object, *, timeout: float = 0.1) -> MCPToolWrapper:
+def _make_wrapper(
+    session: object,
+    *,
+    timeout: float = 0.1,
+    meta: dict | None = None,
+) -> MCPToolWrapper:
     tool_def = SimpleNamespace(
         name="demo",
         description="demo tool",
         inputSchema={"type": "object", "properties": {}},
+        outputSchema={"type": "object"},
+        annotations={"readOnlyHint": True},
+        meta=meta,
     )
     return MCPToolWrapper(session, "test", tool_def, tool_timeout=timeout)
 
@@ -564,6 +573,58 @@ async def call_tool(_name: str, arguments: dict) -> object:
     assert not is_tool_error_result(result)
 
 
+@pytest.mark.asyncio
+async def test_execute_preserves_mcp_app_result_outside_model_text() -> None:
+    async def call_tool(_name: str, arguments: dict) -> object:
+        return SimpleNamespace(
+            content=[_FakeTextContent("model summary")],
+            structuredContent={"rows": [{"id": 1}]},
+            meta={"requestId": "req-1"},
+            isError=False,
+        )
+
+    wrapper = _make_wrapper(
+        SimpleNamespace(call_tool=call_tool),
+        meta={
+            "ui": {
+                "resourceUri": "ui://inventory/dashboard",
+                "visibility": ["model", "app"],
+            }
+        },
+    )
+
+    result = await wrapper.execute()
+
+    assert isinstance(result, ToolResult)
+    assert str(result) == "model summary"
+    assert "rows" not in str(result)
+    assert result.data == {
+        "kind": "mcp_app_result",
+        "tool": {
+            "server": "test",
+            "name": "demo",
+            "ui": {
+                "resourceUri": "ui://inventory/dashboard",
+                "visibility": ["model", "app"],
+            },
+            "outputSchema": {"type": "object"},
+            "annotations": {"readOnlyHint": True},
+            "_meta": {
+                "ui": {
+                    "resourceUri": "ui://inventory/dashboard",
+                    "visibility": ["model", "app"],
+                }
+            },
+        },
+        "result": {
+            "content": [{"text": "model summary"}],
+            "structuredContent": {"rows": [{"id": 1}]},
+            "_meta": {"requestId": "req-1"},
+            "isError": False,
+        },
+    }
+
+
 # Smallest valid 1x1 PNG, base64 without the data: prefix.
 _PNG_B64 = (
     "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8"
@@ -677,11 +738,14 @@ async def call_tool(_name: str, arguments: dict) -> object:
     assert is_tool_error_result(result)
 
 
-def _make_tool_def(name: str) -> SimpleNamespace:
+def _make_tool_def(name: str, *, meta: dict | None = None) -> SimpleNamespace:
     return SimpleNamespace(
         name=name,
         description=f"{name} tool",
         inputSchema={"type": "object", "properties": {}},
+        outputSchema=None,
+        annotations=None,
+        meta=meta,
     )
 
 
@@ -727,6 +791,40 @@ async def test_connect_mcp_servers_enabled_tools_defaults_to_all(
     assert registry.tool_names == ["mcp_test_demo", "mcp_test_other"]
 
 
+@pytest.mark.asyncio
+async def test_connect_mcp_servers_hides_app_only_tools_from_model(
+    fake_mcp_runtime: dict[str, object | None],
+) -> None:
+    session = _make_fake_session(["model-visible"])
+
+    async def list_tools() -> SimpleNamespace:
+        return SimpleNamespace(
+            tools=[
+                _make_tool_def(
+                    "app-only",
+                    meta={"ui/visibility": ["app"], "ui/resourceUri": "ui://app"},
+                ),
+                _make_tool_def(
+                    "model-visible",
+                    meta={"ui/visibility": ["model", "app"]},
+                ),
+            ]
+        )
+
+    session.list_tools = list_tools
+    fake_mcp_runtime["session"] = session
+    registry = ToolRegistry()
+
+    stacks = await connect_mcp_servers(
+        {"test": MCPServerConfig(command="fake")},
+        registry,
+    )
+    for stack in stacks.values():
+        await stack.aclose()
+
+    assert registry.tool_names == ["mcp_test_model-visible"]
+
+
 @pytest.mark.asyncio
 async def test_connect_mcp_servers_enabled_tools_supports_wrapped_names(
     fake_mcp_runtime: dict[str, object | None],