Typed tool returns: enforced ToolResult|ToolError contract, served as MCP structured content

README.md

@@ -108,17 +108,28 @@ tools and merge.
 2. Write your tools in `toolsets/my-toolset/src/my_toolset/tools.py`:
 
    ```python
+   from typing import Any, NotRequired
+
    from langchain_core.tools import tool
 
+   from mcp_runtime.tool_result import ToolError, ToolResult
+
+   class DoSomethingResult(ToolResult):
+       """Matches for the query, each with an 'id' and a 'score'."""
+
+       matches: NotRequired[list[dict[str, Any]]]
+
    @tool
-   def do_something(query: str, limit: int = 10) -> list[dict]:
+   def do_something(query: str, limit: int = 10) -> DoSomethingResult | ToolError:
        """One-line description — docstrings and type hints ARE the MCP schema."""
        ...
+       return DoSomethingResult(message=f"Found {len(matches)} match(es).", matches=matches)
 
    TOOLS = [do_something]
    ```
 
-   `TOOLS` is the only required export. Non-empty docstrings are enforced by a
+   `TOOLS` is the only required export. Non-empty docstrings and the
+   [ToolResult return contract](#typed-tool-returns) are enforced by a
    contract test. If a tool does I/O (HTTP, database), write it as
    `async def` — `@tool` supports coroutines natively; sync tools are fine
    for pure computation (the runtime runs them in a thread pool). If a tool
@@ -139,6 +150,56 @@ tools and merge.
 Conventions: directory `toolsets/<name>` (kebab-case) → module
 `<name_snake_case>.tools` → service `mcp-<name>`.
 
+## Typed tool returns
+
+Every tool returns one dict per call, in one of two shapes from
+`mcp_runtime.tool_result`:
+
+- **`ToolResult`** — success: a required str `message` (the human-readable
+  answer a model or UI reads first) plus any data keys your tool declares.
+- **`ToolError`** — a structured error: a short machine-readable `error`
+  kind and a `detail` saying what happened or what to do next.
+
+The runtime derives each tool's MCP `outputSchema` from its return
+annotation, advertises it in `tools/list`, validates every result against it
+before sending, and delivers results as typed `structuredContent` (alongside
+the usual text block). A tool whose annotation doesn't follow the contract
+**fails at startup** (`build_server` aborts, naming the tool) and fails the
+contract test in CI — never silently at chat time.
+
+How to annotate:
+
+- Minimum: `-> ToolResult | ToolError` for tools whose message is the whole
+  answer (drop the `ToolError` arm if the tool raises instead of returning
+  errors — exceptions become MCP `isError` results, which skip schema
+  validation).
+- Recommended: one `ToolResult` subclass per tool, adding each data key as
+  `NotRequired[...]`, annotated `-> MyResult | ToolError`. Give the subclass
+  a one-line docstring — it becomes the schema's `description`. Nested
+  payloads can be TypedDicts or pydantic models all the way down.
+- Construct returns with TypedDict call syntax —
+  `ToolResult(message=...)`, `ToolError(error="not_found", detail=...)` —
+  mypy-checked, still a plain dict at runtime. `is_error()` (a `TypeIs`
+  guard) narrows helper results typed `dict[str, Any] | ToolError` in both
+  branches.
+
+Rules and gotchas:
+
+- Keys not declared in the annotation are silently dropped from
+  `structuredContent` — the annotation is the complete list of keys a client
+  can see, and mypy flags undeclared keys in return literals.
+- Union arms must all be TypedDicts/pydantic models; bare `str`/`list`
+  returns and `dict[str, Any]` are rejected at startup (FastMCP would wrap
+  the former in `{"result": ...}`, changing your payload shape; the latter
+  guarantees nothing). Put data under a named key instead.
+- The annotation must be on the function `@tool` wraps; the runtime reads it
+  via `tool.coroutine`/`tool.func`.
+
+Verify locally: `TOOLSET=my-toolset uv run mcp-serve`, then `tools/list`
+(via MCP Inspector or `mcp-cli`) shows each tool's `outputSchema`, and
+`tools/call` responses carry `structuredContent`.
+
+
 ## Removing a toolset
 
 ```sh
@@ -361,7 +422,7 @@ call, and the tool reads them at call time:
 from mcp_runtime.credentials import credential_from_header
 
 @tool
-def whoami() -> dict[str, Any]:
+def whoami() -> WhoamiResult:
     """Report which account the calling user's credential belongs to."""
     token = credential_from_header("x-demo-token")
     ...

packages/mcp-runtime/pyproject.toml

@@ -9,6 +9,7 @@ dependencies = [
     "langchain-core<2.0.0,>=1.4.6",
     "pydantic-settings<3.0.0,>=2.12.0",
     "fastapi<1.0.0,>=0.136.3",
+    "typing-extensions>=4.10.0",
 ]
 
 [project.scripts]

packages/mcp-runtime/src/mcp_runtime/fastmcp_output.py

@@ -0,0 +1,112 @@
+"""Upstream ``to_fastmcp`` plus an output schema from the return annotation.
+
+langchain-mcp-adapters' ``to_fastmcp`` discards the return annotation, so
+FastMCP never advertises an ``outputSchema`` and results reach clients as
+JSON text only. This wrapper derives the output model from the annotation —
+a TypedDict/BaseModel, or a union of them, with a required str ``message``
+on at least one arm (the :mod:`mcp_runtime.tool_result` contract) — so
+results also travel as ``structuredContent``. Any other annotation raises at
+conversion, aborting ``build_server``: FastMCP would wrap such values in
+``{"result": ...}`` or the schema would guarantee nothing.
+
+FastMCP validates every returned dict against the model; undeclared keys are
+dropped, so the annotation is the complete list of keys a client can see.
+"""
+
+from types import UnionType
+from typing import Any, Union, get_args, get_origin, get_type_hints, is_typeddict
+
+from langchain_core.tools import BaseTool
+from langchain_mcp_adapters.tools import to_fastmcp as _to_fastmcp
+from mcp.server.fastmcp.tools import Tool as FastMCPTool
+from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata
+from pydantic import BaseModel, RootModel
+
+
+def _return_annotation(tool: BaseTool) -> Any:
+    """The return annotation of the function behind a LangChain tool, if any."""
+    fn = getattr(tool, "coroutine", None) or getattr(tool, "func", None)
+    if fn is None:
+        return None
+    return get_type_hints(fn).get("return")
+
+
+def _arms(annotation: Any) -> tuple[Any, ...]:
+    """The annotation's union members, or the annotation itself."""
+    if get_origin(annotation) in (Union, UnionType):
+        return get_args(annotation)
+    return (annotation,)
+
+
+def _structured_dict(arm: Any) -> bool:
+    """Whether one annotation arm maps 1:1 onto a structuredContent object."""
+    return is_typeddict(arm) or (isinstance(arm, type) and issubclass(arm, BaseModel))
+
+
+def _resolve(schema: dict[str, Any], defs: dict[str, Any]) -> dict[str, Any]:
+    """A schema node with its top-level ``$ref`` resolved against ``$defs``."""
+    reference = schema.get("$ref", "")
+    if reference.startswith("#/$defs/"):
+        return defs.get(reference.removeprefix("#/$defs/"), {})
+    return schema
+
+
+def _shape_schema(schema: dict[str, Any]) -> dict[str, Any]:
+    """The advertised schema, object-rooted as MCP clients expect.
+
+    ``RootModel`` schemas root at a ``$ref`` (single model) or an ``anyOf``
+    (union): inline the former, stamp ``"type": "object"`` on the latter.
+    """
+    defs = dict(schema.get("$defs", {}))
+    if schema.get("$ref", "").startswith("#/$defs/"):
+        inlined = dict(_resolve(schema, defs))
+        defs.pop(schema["$ref"].removeprefix("#/$defs/"))
+        if defs:
+            inlined["$defs"] = defs
+        return inlined
+    if "anyOf" in schema:
+        return {"type": "object", "anyOf": schema["anyOf"], "$defs": defs}
+    return schema
+
+
+def _offers_message(schema: dict[str, Any]) -> bool:
+    """Whether some arm of the schema requires a str ``message`` property."""
+    defs: dict[str, Any] = schema.get("$defs", {})
+    arms = [_resolve(arm, defs) for arm in schema.get("anyOf", [schema])]
+    return any(
+        arm.get("properties", {}).get("message", {}).get("type") == "string"
+        and "message" in arm.get("required", [])
+        for arm in arms
+    )
+
+
+def to_fastmcp(tool: BaseTool) -> FastMCPTool:
+    """Convert a LangChain tool to FastMCP, deriving its output schema.
+
+    Raises ``RuntimeError`` (naming the tool) when the return annotation does
+    not follow the ToolResult contract — see the module docstring.
+    """
+    converted = _to_fastmcp(tool)
+    annotation = _return_annotation(tool)
+    if annotation is None or not all(
+        _structured_dict(arm) for arm in _arms(annotation)
+    ):
+        raise RuntimeError(
+            f"tool {tool.name!r} needs a ToolResult return annotation "
+            f"(a TypedDict/BaseModel or a union of them; got {annotation!r}) "
+            "— see mcp_runtime.tool_result"
+        )
+    model: type[BaseModel] = RootModel[annotation]  # type: ignore[valid-type]
+    schema = _shape_schema(model.model_json_schema())
+    if not _offers_message(schema):
+        raise RuntimeError(
+            f"tool {tool.name!r} does not follow the ToolResult contract: no arm "
+            "of its return annotation has a required str 'message' property"
+        )
+    converted.fn_metadata = FuncMetadata(
+        arg_model=converted.fn_metadata.arg_model,
+        output_model=model,
+        output_schema=schema,
+        wrap_output=False,
+    )
+    return converted

packages/mcp-runtime/src/mcp_runtime/server.py

@@ -14,13 +14,14 @@
 from ipaddress import IPv4Address
 
 from langchain_core.tools import BaseTool
-from langchain_mcp_adapters.tools import to_fastmcp
 from mcp.server.fastmcp import FastMCP
 from pydantic import Field, IPvAnyAddress
 from pydantic_settings import BaseSettings
 from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
 
+from mcp_runtime.fastmcp_output import to_fastmcp
+
 
 class RuntimeSettings(BaseSettings):
     """Runtime configuration, validated from the environment."""

packages/mcp-runtime/src/mcp_runtime/tool_result.py

@@ -0,0 +1,45 @@
+"""The base contract every tool return follows.
+
+A tool returns one dict per call, in one of two shapes:
+
+- :class:`ToolResult` — success: a ``message`` (the text the model reads)
+  plus any data keys the tool declares, captured into session state.
+- :class:`ToolError` — failure: a machine-readable ``error`` kind and a
+  ``detail`` saying what happened or what to do next.
+
+Subclass :class:`ToolResult` per tool, one ``NotRequired`` field per data
+key, and annotate the ``@tool`` function with the union::
+
+    class SearchDatasetsResult(ToolResult):
+        datasets: NotRequired[list[dict[str, Any]]]
+
+    @tool
+    async def search_datasets(query: str) -> SearchDatasetsResult | ToolError:
+        ...
+
+``mcp_runtime.fastmcp_output`` derives the tool's MCP ``outputSchema`` from
+the annotation and refuses to serve a tool that breaks the contract — a
+deploy-time gate, not a convention.
+"""
+
+from typing import Any, TypedDict
+
+from typing_extensions import TypeIs
+
+
+class ToolResult(TypedDict):
+    """A successful tool return: a ``message`` plus declared data keys."""
+
+    message: str
+
+
+class ToolError(TypedDict):
+    """A structured tool error: what kind, and what to do about it."""
+
+    error: str
+    detail: str
+
+
+def is_error(result: dict[str, Any] | ToolError) -> TypeIs[ToolError]:
+    """Whether a helper's raw dict is an error return (truthy ``error`` key)."""
+    return bool(result.get("error"))

packages/mcp-runtime/tests/test_contract.py

@@ -2,7 +2,8 @@
 
 Doubles as a per-toolset import smoke test and enforces non-empty
 descriptions (docstrings become the MCP schema, so they are part of the
-contract).
+contract) and the ToolResult return contract (annotations become the MCP
+output schema, so every tool in every toolset must convert cleanly).
 """
 
 import importlib
@@ -11,6 +12,7 @@
 import pytest
 from langchain_core.tools import BaseTool
 
+from mcp_runtime.fastmcp_output import to_fastmcp
 from mcp_runtime.server import load_credential_headers
 
 TOOLSETS_DIR = Path(__file__).resolve().parents[3] / "toolsets"
@@ -35,3 +37,6 @@ def test_toolset_contract(toolset):
         assert tool.description and tool.description.strip(), (
             f"{toolset}: tool {tool.name!r} needs a non-empty docstring"
         )
+        # The same ToolResult gate build_server applies at startup.
+        converted = to_fastmcp(tool)
+        assert converted.output_schema is not None