diff --git a/docs/handler-authoring.md b/docs/handler-authoring.md index b9b2e98bd..d8b9780b3 100644 --- a/docs/handler-authoring.md +++ b/docs/handler-authoring.md @@ -2050,3 +2050,37 @@ with the combined server's startup/shutdown hooks or the embedding app. A `/healthz` route reports process liveness. Define `/readyz` separately from actual dependency, capacity and initialization checks; return 503 when those checks fail. A successful liveness response alone does not prove readiness. + +### MCP success text summaries + +`mcp_result_text` is an opt-in MCP-only formatter on `serve`, `ServeConfig`, +`create_mcp_server` and the public testing helpers. A callback receives the tool +name, JSON-ready result dictionary and `ToolContext | None`. Sync and async +callbacks are supported. The exported `MCPResultText` type describes both forms. + +```python +from typing import Any +from adcp.server import ToolContext, create_mcp_server + +async def summarize(name: str, result: dict[str, Any], context: ToolContext | None) -> str | None: + if name == "get_products": + return f"Found {len(result.get('products', []))} products." + return None + +mcp = create_mcp_server(handler, mcp_result_text=summarize) +``` + +The callback runs after context echo, response enhancement and response +validation. It gets a detached copy so editing a nested value while formatting +cannot change `structuredContent`. A string, including an empty string, replaces +the text block. `None` keeps the serialized JSON fallback. A field-name shorthand +such as `mcp_result_text="message"` uses a string-valued response field and falls +back to JSON when the field is absent or not a string. A2A payloads and default +MCP output stay unchanged. + +Error results retain their existing structured error/text conversion. Successful +or error `CallToolResult` values supplied by middleware retain their content and +structured data, including empty content, without wrapping them again or calling +the formatter. Callback failures follow MCP's ordinary tool-error behavior; an +invalid return type raises `TypeError` internally. Formatting happens after +handler execution and does not roll back its effects, so keep callbacks simple. diff --git a/src/adcp/server/__init__.py b/src/adcp/server/__init__.py index 412599287..6b1c3508e 100644 --- a/src/adcp/server/__init__.py +++ b/src/adcp/server/__init__.py @@ -168,6 +168,7 @@ async def get_products(params, context=None): ASGIMiddlewareEntry, ContextFactory, LifespanHook, + MCPResultText, RequestMetadata, ServeConfig, SkillMiddleware, @@ -243,6 +244,7 @@ async def get_products(params, context=None): "ProposalNotSupported", # MCP integration "ContextFactory", + "MCPResultText", "DISCOVERY_METHODS", "DISCOVERY_TOOLS", "LifespanHook", diff --git a/src/adcp/server/serve.py b/src/adcp/server/serve.py index 29a83b2b6..4d15b01c8 100644 --- a/src/adcp/server/serve.py +++ b/src/adcp/server/serve.py @@ -20,12 +20,14 @@ async def get_adcp_capabilities(self, params, context=None): import asyncio import contextlib +import inspect import logging import os import sys import warnings from collections.abc import AsyncIterator, Awaitable, Callable, Sequence from contextvars import ContextVar +from copy import deepcopy from dataclasses import dataclass from types import MethodType from typing import TYPE_CHECKING, Any, Literal @@ -298,6 +300,7 @@ class ServeConfig: # --- MCP only --- instructions: str | None = None + mcp_result_text: MCPResultText | None = None streaming_responses: bool = False stateless_http: bool = False session_idle_timeout: float | None = 1800.0 @@ -363,6 +366,7 @@ def __post_init__(self) -> None: # cleanly with the heuristic. _mcp_only = ( "instructions", + "mcp_result_text", "streaming_responses", "stateless_http", "max_active_sessions", @@ -389,6 +393,22 @@ def __post_init__(self) -> None: ) +MCPResultText = ( + str + | Callable[ + [str, dict[str, Any], ToolContext | None], + str | None | Awaitable[str | None], + ] +) +"""MCP-only success text: a response field name or sync/async callback. + +Callbacks run after response validation and context echo, on a detached copy of +JSON-ready structured data. Return a string (including an empty string) for the +text block, or None for the existing JSON fallback. Structured content is never +changed. Prebuilt CallToolResult and error results retain their own text. +""" + + SkillMiddleware = Callable[ [str, dict[str, Any], ToolContext, Callable[[], Awaitable[Any]]], Awaitable[Any], @@ -811,6 +831,7 @@ def serve( port: int | None = None, host: str | None = None, transport: str = "streamable-http", + mcp_result_text: MCPResultText | None = None, instructions: str | None = None, test_controller: TestControllerStore | None = None, test_controller_account_resolver: Any | None = None, @@ -1190,6 +1211,7 @@ async def force_account_status(self, account_id, status): allowed_hosts = config.allowed_hosts allowed_origins = config.allowed_origins enable_dns_rebinding_protection = config.enable_dns_rebinding_protection + mcp_result_text = config.mcp_result_text test_controller = config.test_controller context_factory = config.context_factory task_store = config.task_store @@ -1319,6 +1341,7 @@ async def force_account_status(self, account_id, status): host=host, transport=transport, instructions=instructions, + mcp_result_text=mcp_result_text, test_controller=test_controller, test_controller_account_resolver=test_controller_account_resolver, context_factory=context_factory, @@ -1350,6 +1373,7 @@ async def force_account_status(self, account_id, status): port=port, host=host, instructions=instructions, + mcp_result_text=mcp_result_text, test_controller=test_controller, test_controller_account_resolver=test_controller_account_resolver, context_factory=context_factory, @@ -1780,6 +1804,7 @@ def _serve_mcp( handler: ADCPHandler[Any], *, unauthenticated_routes: Sequence[Route | Mount] | None = None, + mcp_result_text: MCPResultText | None = None, name: str, port: int | None, host: str | None = None, @@ -1815,6 +1840,7 @@ def _serve_mcp( port=port, host=host, instructions=instructions, + mcp_result_text=mcp_result_text, include_test_controller=test_controller is not None, context_factory=context_factory, middleware=middleware, @@ -2146,6 +2172,7 @@ def _build_mcp_and_a2a_app( handler: ADCPHandler[Any], *, unauthenticated_routes: Sequence[Route | Mount] | None = None, + mcp_result_text: MCPResultText | None = None, name: str, port: int, host: str, @@ -2207,6 +2234,7 @@ def _build_mcp_and_a2a_app( port=port, host=host, instructions=instructions, + mcp_result_text=mcp_result_text, include_test_controller=test_controller is not None, context_factory=context_factory, middleware=middleware, @@ -2372,6 +2400,7 @@ def _serve_mcp_and_a2a( handler: ADCPHandler[Any], *, unauthenticated_routes: Sequence[Route | Mount] | None = None, + mcp_result_text: MCPResultText | None = None, name: str, port: int | None, host: str | None = None, @@ -2436,6 +2465,7 @@ def _serve_mcp_and_a2a( port=resolved_port, host=resolved_host, instructions=instructions, + mcp_result_text=mcp_result_text, test_controller=test_controller, test_controller_account_resolver=test_controller_account_resolver, context_factory=context_factory, @@ -2556,6 +2586,7 @@ def create_mcp_server( name: str = "adcp-agent", port: int | None = None, host: str | None = None, + mcp_result_text: MCPResultText | None = None, instructions: str | None = None, include_test_controller: bool = False, context_factory: ContextFactory | None = None, @@ -2583,6 +2614,9 @@ def create_mcp_server( name: Server name. port: Port to listen on. instructions: Optional system instructions. + mcp_result_text: Optional response field name or sync/async callback + for concise MCP text. None preserves the JSON fallback; + structured content uses the unchanged canonical conversion. include_test_controller: When False (default), skip registering ``comply_test_controller`` as a handler tool. Sellers who want compliance-testing support should pass ``test_controller=`` to @@ -2763,6 +2797,7 @@ def create_mcp_server( validation=validation, pre_validation_hooks=pre_validation_hooks, response_enhancer=response_enhancer, + mcp_result_text=mcp_result_text, ) if session_idle_timeout is not None and session_idle_timeout <= 0: raise ValueError( @@ -3002,6 +3037,7 @@ def _register_handler_tools( mcp: Any, handler: ADCPHandler[Any], *, + mcp_result_text: MCPResultText | None = None, include_test_controller: bool = False, context_factory: ContextFactory | None = None, middleware: Sequence[SkillMiddleware] | None = None, @@ -3056,6 +3092,7 @@ def _register_handler_tools( middleware=middleware_tuple, output_schema=output_schema, response_enhancer=response_enhancer, + mcp_result_text=mcp_result_text, ) registered.append(tool_name) @@ -3074,6 +3111,7 @@ def _register_tool( input_schema: dict[str, Any], caller: Callable[..., Any], *, + mcp_result_text: MCPResultText | None = None, context_factory: ContextFactory | None = None, middleware: tuple[SkillMiddleware, ...] = (), output_schema: dict[str, Any] | None = None, @@ -3214,10 +3252,40 @@ async def _call_handler() -> Any: if isinstance(result, CallToolResult): return result # type: ignore[return-value] if hasattr(result, "model_dump"): - return result.model_dump(mode="json", exclude_none=True) # type: ignore[no-any-return] - if isinstance(result, dict): - return result - return {"result": result} + result_dict = result.model_dump(mode="json", exclude_none=True) + elif isinstance(result, dict): + result_dict = result + else: + result_dict = {"result": result} + if mcp_result_text is not None: + # Use MCP's normal conversion before formatting so datetime, + # Decimal, aliases and nested models have exactly the baseline's + # JSON representation in both the callback and structured content. + canonical_result = FuncMetadata.convert_result(tool.fn_metadata, result_dict) + assert isinstance(canonical_result, CallToolResult) + if canonical_result.structured_content is not None: + result_dict = canonical_result.structured_content + if isinstance(mcp_result_text, str): + summary = result_dict.get(mcp_result_text) + if not isinstance(summary, str): + summary = None + else: + # A formatter cannot mutate the canonical payload, even if it + # annotates a nested product while composing its summary. + summary = mcp_result_text(name, deepcopy(result_dict), context) + if inspect.isawaitable(summary): + summary = await summary + if summary is not None: + if not isinstance(summary, str): + raise TypeError("mcp_result_text must return str or None") + from mcp.types import TextContent + + return CallToolResult( # type: ignore[return-value] + content=[TextContent(type="text", text=summary)], + structured_content=result_dict, + ) + return canonical_result # type: ignore[return-value] + return result_dict # type: ignore[no-any-return] # Create tool from function (gives us proper fn_metadata scaffolding) tool = Tool.from_function(fn, name=name, description=description, structured_output=True) @@ -3240,19 +3308,15 @@ def model_dump_one_level(self) -> dict[str, Any]: return result class _AdcpFuncMetadata(FuncMetadata): - """FuncMetadata that skips success-path output validation for error - ``CallToolResult`` returns. - - FastMCP's stock ``convert_result`` validates ``result.structuredContent`` - against the success-path ``output_model`` whenever the tool returns a - ``CallToolResult`` — but when the framework projects an ``AdcpError`` - as ``{"adcp_error": {...}}``, that payload doesn't conform to the - success schema. Skip validation for ``isError=True`` envelopes; success - envelopes still validate normally. + """Keep prebuilt success/error envelopes intact through MCP conversion. + + Handler dictionaries have already gone through SDK validation. MCP + summaries carry that same structured payload; passing through avoids + wrapping content a second time. Error envelopes use a different schema. """ def convert_result(self, result: Any) -> Any: - if isinstance(result, CallToolResult) and result.is_error: + if isinstance(result, CallToolResult): return result return super().convert_result(result) diff --git a/src/adcp/testing/decisioning.py b/src/adcp/testing/decisioning.py index d2f03252d..99476a64c 100644 --- a/src/adcp/testing/decisioning.py +++ b/src/adcp/testing/decisioning.py @@ -54,6 +54,7 @@ ASGIMiddlewareEntry, ContextFactory, LifespanHook, + MCPResultText, SkillMiddleware, ) from adcp.server.spec_compat import PreValidationHooks @@ -164,6 +165,7 @@ def build_asgi_app( discovery_base_url: str | None = None, pre_validation_hooks: PreValidationHooks | None = None, response_enhancer: ResponseEnhancer | None = None, + mcp_result_text: MCPResultText | None = None, on_startup: Sequence[LifespanHook] | None = None, on_shutdown: Sequence[LifespanHook] | None = None, **factory_kwargs: Any, @@ -345,6 +347,7 @@ def build_asgi_app( validation=validation, pre_validation_hooks=pre_validation_hooks, response_enhancer=response_enhancer, + mcp_result_text=mcp_result_text, base_url=discovery_base_url, allowed_hosts=allowed_hosts, allowed_origins=allowed_origins, @@ -396,6 +399,7 @@ def build_asgi_app( validation=validation, pre_validation_hooks=pre_validation_hooks, response_enhancer=response_enhancer, + mcp_result_text=mcp_result_text, ) # Mirror the wrapping chain from _run_mcp_http (adcp.server.serve). # auth must be innermost so its JSON-RPC body-peek runs before the @@ -449,6 +453,7 @@ async def build_test_client( discovery_base_url: str | None = None, pre_validation_hooks: PreValidationHooks | None = None, response_enhancer: ResponseEnhancer | None = None, + mcp_result_text: MCPResultText | None = None, on_startup: Sequence[LifespanHook] | None = None, on_shutdown: Sequence[LifespanHook] | None = None, **factory_kwargs: Any, @@ -581,6 +586,7 @@ async def build_test_client( discovery_base_url=discovery_base_url, pre_validation_hooks=pre_validation_hooks, response_enhancer=response_enhancer, + mcp_result_text=mcp_result_text, on_startup=on_startup, on_shutdown=on_shutdown, **factory_kwargs, diff --git a/tests/test_mcp_result_text.py b/tests/test_mcp_result_text.py new file mode 100644 index 000000000..20931f081 --- /dev/null +++ b/tests/test_mcp_result_text.py @@ -0,0 +1,281 @@ +"""MCP summaries preserve canonical payloads and prebuilt result envelopes.""" + +import importlib +import json +from copy import deepcopy + +import pytest +from mcp.server import MCPServer +from mcp.types import CallToolResult, TextContent + +from adcp.decisioning.types import AdcpError +from adcp.server import ADCPHandler, ServeConfig, ToolContext, create_mcp_server +from adcp.server.serve import _register_tool +from adcp.validation.client_hooks import ValidationHookConfig +from adcp.validation.schema_validator import ValidationIssue, ValidationOutcome + + +class Seller(ADCPHandler): + advertised_tools = {"get_products"} + + async def get_products(self, params, context=None): + return {"products": [{"product_id": "p1", "name": "Video"}], "message": "Found one product"} + + +ARGS = {"buying_mode": "brief", "brief": "video", "context": {"trace": "request-1"}} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("formatter", [None, "missing", "products", lambda name, result, ctx: None]) +async def test_default_and_none_fallback_are_json(formatter): + mcp = create_mcp_server(Seller(), validation=None, mcp_result_text=formatter) + result = await mcp.call_tool("get_products", ARGS) + assert json.loads(result.content[0].text) == result.structured_content + + +@pytest.mark.asyncio +@pytest.mark.parametrize("asynchronous", [False, True]) +async def test_summary_runs_after_context_echo_validation_and_enhancement( + monkeypatch, asynchronous +): + events = [] + context = ToolContext(caller_identity="buyer") + + def enhance(method, result, ctx): + assert result["context"] == ARGS["context"] + result["extra"] = {"validated": True} + events.append("enhance") + + def validate(method, result, **kwargs): + assert result["extra"] == {"validated": True} + events.append("validate") + return ValidationOutcome(valid=True) + + monkeypatch.setattr("adcp.validation.schema_validator.validate_response", validate) + seen = [] + + def summary(name, result, ctx): + assert events == ["enhance", "validate"] + assert name == "get_products" + assert ctx is context + seen.append(deepcopy(result)) + result["products"][0]["name"] = "formatter mutation" + result["context"]["trace"] = "formatter mutation" + result["extra"]["validated"] = False + return "One product ready" + + async def async_summary(name, result, ctx): + return summary(name, result, ctx) + + mcp = create_mcp_server( + Seller(), + validation=ValidationHookConfig(requests="off", responses="strict"), + response_enhancer=enhance, + context_factory=lambda meta: context, + mcp_result_text=async_summary if asynchronous else summary, + ) + result = await mcp.call_tool("get_products", ARGS) + assert result.content == [TextContent(type="text", text="One product ready")] + assert result.structured_content == seen[0] + assert result.structured_content["products"][0]["name"] == "Video" + assert ARGS["context"]["trace"] == "request-1" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "formatter", ["message", lambda name, result, ctx: "", lambda name, result, ctx: "Summary"] +) +async def test_summaries_preserve_baseline_data(formatter): + baseline = await create_mcp_server(Seller(), validation=None).call_tool("get_products", ARGS) + result = await create_mcp_server( + Seller(), validation=None, mcp_result_text=formatter + ).call_tool("get_products", ARGS) + assert result.structured_content == baseline.structured_content + expected = ( + "Found one product" if formatter == "message" else formatter("get_products", {}, None) + ) + assert result.content == [TextContent(type="text", text=expected)] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("is_error", [False, True]) +@pytest.mark.parametrize("empty_content", [False, True]) +async def test_prebuilt_results_are_not_double_wrapped(is_error, empty_content): + result = CallToolResult( + is_error=is_error, + content=[] if empty_content else [TextContent(type="text", text="prebuilt")], + structured_content={"canonical": [1, 2, 3]}, + ) + + async def caller(kwargs, *, context=None): + return result + + def forbidden(*args): + pytest.fail("prebuilt result must not invoke the formatter") + + mcp = MCPServer("prebuilt") + _register_tool(mcp, "example", "example", {"type": "object"}, caller, mcp_result_text=forbidden) + actual = await mcp.call_tool("example", {}) + assert actual is result + assert actual.structured_content == {"canonical": [1, 2, 3]} + + +@pytest.mark.asyncio +async def test_strict_validation_error_uses_error_conversion_without_summary(monkeypatch): + def invalid(*args, **kwargs): + return ValidationOutcome( + valid=False, issues=[ValidationIssue("/products", "invalid shape", "type", "")] + ) + + monkeypatch.setattr("adcp.validation.schema_validator.validate_response", invalid) + mcp = create_mcp_server( + Seller(), + validation=ValidationHookConfig(requests="off", responses="strict"), + mcp_result_text=lambda *args: pytest.fail("invalid response must not invoke the formatter"), + ) + result = await mcp.call_tool("get_products", ARGS) + assert result.is_error + assert result.structured_content["adcp_error"]["code"] == "VALIDATION_ERROR" + assert result.structured_content["context"] == ARGS["context"] + + +@pytest.mark.asyncio +async def test_handler_errors_keep_error_text_and_structure(): + class Failing(Seller): + async def get_products(self, params, context=None): + raise AdcpError("INVALID_REQUEST", message="Fix the brief") + + result = await create_mcp_server( + Failing(), + validation=None, + mcp_result_text=lambda *args: pytest.fail("error must not invoke the formatter"), + ).call_tool("get_products", ARGS) + assert result.is_error + assert result.structured_content["adcp_error"]["code"] == "INVALID_REQUEST" + assert "Fix the brief" in result.content[0].text + + +@pytest.mark.asyncio +async def test_wrong_callback_return_type_fails_clearly(): + mcp = create_mcp_server(Seller(), validation=None, mcp_result_text=lambda *args: 123) + with pytest.raises(Exception, match="mcp_result_text must return str or None"): + await mcp._tool_manager._tools["get_products"].fn(**ARGS) + + +@pytest.mark.parametrize("transport", ["streamable-http", "both"]) +def test_serve_config_forwards_hook(monkeypatch, transport): + module = importlib.import_module("adcp.server.serve") + captured = [] + target = "_serve_mcp" if transport == "streamable-http" else "_serve_mcp_and_a2a" + monkeypatch.setattr(module, target, lambda *args, **kwargs: captured.append(kwargs)) + module.serve(Seller(), config=ServeConfig(transport=transport, mcp_result_text="message")) + assert captured[0]["mcp_result_text"] == "message" + + +@pytest.mark.asyncio +async def test_hook_changes_only_mcp_payload_in_combined_app(): + import httpx + from asgi_lifespan import LifespanManager + + from adcp.server.serve import _build_mcp_and_a2a_app + + calls = [] + a2a_body = { + "jsonrpc": "2.0", + "id": 1, + "method": "message/send", + "params": { + "message": { + "messageId": "m1", + "role": "user", + "parts": [{"kind": "data", "data": {"skill": "get_products", "parameters": ARGS}}], + }, + }, + } + a2a_data = [] + for hook in [None, lambda name, result, ctx: calls.append(name) or "Summary"]: + app = _build_mcp_and_a2a_app( + Seller(), + name="result", + port=3001, + host="127.0.0.1", + instructions=None, + test_controller=None, + validation=None, + stateless_http=True, + allowed_hosts=["localhost"], + mcp_result_text=hook, + ) + async with LifespanManager(app): + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://localhost" + ) as client: + response = await client.post("/", json=a2a_body) + assert response.status_code == 200, response.text + a2a_data.append(response.json()["result"]["artifacts"][0]["parts"][0]["data"]) + assert calls == [] + mcp_response = await client.post( + "/mcp", + headers={"accept": "application/json, text/event-stream"}, + json={ + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": "get_products", + "arguments": ARGS, + }, + }, + ) + assert mcp_response.status_code == 200 + payload = mcp_response.json()["result"] + if hook: + assert payload["content"][0]["text"] == "Summary" + else: + assert json.loads(payload["content"][0]["text"]) == payload["structuredContent"] + assert a2a_data[0] == a2a_data[1] + assert calls == ["get_products"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("summary", ["Readable", None]) +async def test_formatter_receives_baseline_json_values(summary): + from datetime import datetime, timezone + from decimal import Decimal + + from pydantic import BaseModel, Field + + class Nested(BaseModel): + value: str = Field(serialization_alias="wire_value") + + class RichSeller(Seller): + async def get_products(self, params, context=None): + return { + "products": [], + "timestamp": datetime(2026, 10, 2, tzinfo=timezone.utc), + "price": Decimal("1.25"), + "pair": (1, 2), + "nested": Nested(value="ok"), + } + + baseline = await create_mcp_server(RichSeller(), validation=None).call_tool( + "get_products", ARGS + ) + seen = [] + + def formatter(name, result, ctx): + seen.append(result) + assert result == baseline.structured_content + assert result["price"] == "1.25" + assert result["pair"] == [1, 2] + assert isinstance(result["nested"], dict) + return summary + + result = await create_mcp_server( + RichSeller(), validation=None, mcp_result_text=formatter + ).call_tool("get_products", ARGS) + assert seen == [baseline.structured_content] + assert result.structured_content == baseline.structured_content + assert result.content == ( + baseline.content if summary is None else [TextContent(type="text", text=summary)] + ) diff --git a/tests/type_checks/mcp_result_text.py b/tests/type_checks/mcp_result_text.py new file mode 100644 index 000000000..5db310364 --- /dev/null +++ b/tests/type_checks/mcp_result_text.py @@ -0,0 +1,40 @@ +"""Typed sync/async MCP-only summary callbacks and field shorthand.""" + +from typing import Any + +from adcp.decisioning import DecisioningPlatform +from adcp.server import ( + ADCPHandler, + MCPResultText, + ServeConfig, + ToolContext, + create_mcp_server, + serve, +) +from adcp.testing import build_asgi_app, build_test_client + + +def summarize(name: str, result: dict[str, Any], context: ToolContext | None) -> str | None: + return f"{name}: {len(result.get('products', []))} products" + + +async def async_summarize( + name: str, result: dict[str, Any], context: ToolContext | None +) -> str | None: + return summarize(name, result, context) + + +def configure(handler: ADCPHandler[ToolContext], platform: DecisioningPlatform) -> None: + hook: MCPResultText = summarize + serve(handler, config=ServeConfig(transport="both", mcp_result_text=hook)) + create_mcp_server(handler, mcp_result_text=async_summarize) + create_mcp_server(handler, mcp_result_text="message") + build_asgi_app(platform, mcp_result_text=hook) + + +async def check(platform: DecisioningPlatform) -> None: + async with build_test_client( + platform, transport="both", mcp_result_text=async_summarize + ) as client: + response = await client.get("/.well-known/agent-card.json") + assert response.status_code == 200