diff --git a/README.md b/README.md index 48d1c1742f..823ab4d4ea 100644 --- a/README.md +++ b/README.md @@ -383,6 +383,61 @@ causes the tool to be classified as structured _and this is undesirable_, the classification can be suppressed by passing `structured_output=False` to the `@tool` decorator. +##### Advanced: Direct CallToolResult + +For full control over tool responses including the `_meta` field (for passing data to client applications without exposing it to the model), you can return `CallToolResult` directly: + + +```python +"""Example showing direct CallToolResult return for advanced control.""" + +from typing import Annotated + +from pydantic import BaseModel + +from mcp.server.fastmcp import FastMCP +from mcp.types import CallToolResult, TextContent + +mcp = FastMCP("CallToolResult Example") + + +class ValidationModel(BaseModel): + """Model for validating structured output.""" + + status: str + data: dict[str, int] + + +@mcp.tool() +def advanced_tool() -> CallToolResult: + """Return CallToolResult directly for full control including _meta field.""" + return CallToolResult( + content=[TextContent(type="text", text="Response visible to the model")], + _meta={"hidden": "data for client applications only"}, + ) + + +@mcp.tool() +def validated_tool() -> Annotated[CallToolResult, ValidationModel]: + """Return CallToolResult with structured output validation.""" + return CallToolResult( + content=[TextContent(type="text", text="Validated response")], + structuredContent={"status": "success", "data": {"result": 42}}, + _meta={"internal": "metadata"}, + ) + + +@mcp.tool() +def empty_result_tool() -> CallToolResult: + """For empty results, return CallToolResult with empty content.""" + return CallToolResult(content=[]) +``` + +_Full example: [examples/snippets/servers/direct_call_tool_result.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/direct_call_tool_result.py)_ + + +**Important:** `CallToolResult` must always be returned (no `Optional` or `Union`). For empty results, use `CallToolResult(content=[])`. For optional simple types, use `str | None` without `CallToolResult`. + ```python """Example showing structured output with tools.""" @@ -1769,14 +1824,93 @@ if __name__ == "__main__": _Full example: [examples/snippets/servers/lowlevel/structured_output.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/lowlevel/structured_output.py)_ -Tools can return data in three ways: +Tools can return data in four ways: 1. **Content only**: Return a list of content blocks (default behavior before spec revision 2025-06-18) 2. **Structured data only**: Return a dictionary that will be serialized to JSON (Introduced in spec revision 2025-06-18) 3. **Both**: Return a tuple of (content, structured_data) preferred option to use for backwards compatibility +4. **Direct CallToolResult**: Return `CallToolResult` directly for full control (including `_meta` field) When an `outputSchema` is defined, the server automatically validates the structured output against the schema. This ensures type safety and helps catch errors early. +##### Returning CallToolResult Directly + +For full control over the response including the `_meta` field (for passing data to client applications without exposing it to the model), return `CallToolResult` directly: + + +```python +""" +Run from the repository root: + uv run examples/snippets/servers/lowlevel/direct_call_tool_result.py +""" + +import asyncio +from typing import Any + +import mcp.server.stdio +import mcp.types as types +from mcp.server.lowlevel import NotificationOptions, Server +from mcp.server.models import InitializationOptions + +server = Server("example-server") + + +@server.list_tools() +async def list_tools() -> list[types.Tool]: + """List available tools.""" + return [ + types.Tool( + name="advanced_tool", + description="Tool with full control including _meta field", + inputSchema={ + "type": "object", + "properties": {"message": {"type": "string"}}, + "required": ["message"], + }, + ) + ] + + +@server.call_tool() +async def handle_call_tool(name: str, arguments: dict[str, Any]) -> types.CallToolResult: + """Handle tool calls by returning CallToolResult directly.""" + if name == "advanced_tool": + message = str(arguments.get("message", "")) + return types.CallToolResult( + content=[types.TextContent(type="text", text=f"Processed: {message}")], + structuredContent={"result": "success", "message": message}, + _meta={"hidden": "data for client applications only"}, + ) + + raise ValueError(f"Unknown tool: {name}") + + +async def run(): + """Run the server.""" + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="example", + server_version="0.1.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions(), + experimental_capabilities={}, + ), + ), + ) + + +if __name__ == "__main__": + asyncio.run(run()) +``` + +_Full example: [examples/snippets/servers/lowlevel/direct_call_tool_result.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/lowlevel/direct_call_tool_result.py)_ + + +**Note:** When returning `CallToolResult`, you bypass the automatic content/structured conversion. You must construct the complete response yourself. + ### Pagination (Advanced) For servers that need to handle large datasets, the low-level server provides paginated versions of list operations. This is an optional optimization - most servers won't need pagination unless they're dealing with hundreds or thousands of items. diff --git a/examples/fastmcp/direct_call_tool_result_return.py b/examples/fastmcp/direct_call_tool_result_return.py new file mode 100644 index 0000000000..a441769b2a --- /dev/null +++ b/examples/fastmcp/direct_call_tool_result_return.py @@ -0,0 +1,24 @@ +""" +FastMCP Echo Server with direct CallToolResult return +""" + +from typing import Annotated + +from pydantic import BaseModel + +from mcp.server.fastmcp import FastMCP +from mcp.types import CallToolResult, TextContent + +mcp = FastMCP("Echo Server") + + +class EchoResponse(BaseModel): + text: str + + +@mcp.tool() +def echo(text: str) -> Annotated[CallToolResult, EchoResponse]: + """Echo the input text with structure and metadata""" + return CallToolResult( + content=[TextContent(type="text", text=text)], structuredContent={"text": text}, _meta={"some": "metadata"} + ) diff --git a/examples/snippets/servers/direct_call_tool_result.py b/examples/snippets/servers/direct_call_tool_result.py new file mode 100644 index 0000000000..54d49b2f66 --- /dev/null +++ b/examples/snippets/servers/direct_call_tool_result.py @@ -0,0 +1,42 @@ +"""Example showing direct CallToolResult return for advanced control.""" + +from typing import Annotated + +from pydantic import BaseModel + +from mcp.server.fastmcp import FastMCP +from mcp.types import CallToolResult, TextContent + +mcp = FastMCP("CallToolResult Example") + + +class ValidationModel(BaseModel): + """Model for validating structured output.""" + + status: str + data: dict[str, int] + + +@mcp.tool() +def advanced_tool() -> CallToolResult: + """Return CallToolResult directly for full control including _meta field.""" + return CallToolResult( + content=[TextContent(type="text", text="Response visible to the model")], + _meta={"hidden": "data for client applications only"}, + ) + + +@mcp.tool() +def validated_tool() -> Annotated[CallToolResult, ValidationModel]: + """Return CallToolResult with structured output validation.""" + return CallToolResult( + content=[TextContent(type="text", text="Validated response")], + structuredContent={"status": "success", "data": {"result": 42}}, + _meta={"internal": "metadata"}, + ) + + +@mcp.tool() +def empty_result_tool() -> CallToolResult: + """For empty results, return CallToolResult with empty content.""" + return CallToolResult(content=[]) diff --git a/examples/snippets/servers/lowlevel/direct_call_tool_result.py b/examples/snippets/servers/lowlevel/direct_call_tool_result.py new file mode 100644 index 0000000000..496eaad105 --- /dev/null +++ b/examples/snippets/servers/lowlevel/direct_call_tool_result.py @@ -0,0 +1,65 @@ +""" +Run from the repository root: + uv run examples/snippets/servers/lowlevel/direct_call_tool_result.py +""" + +import asyncio +from typing import Any + +import mcp.server.stdio +import mcp.types as types +from mcp.server.lowlevel import NotificationOptions, Server +from mcp.server.models import InitializationOptions + +server = Server("example-server") + + +@server.list_tools() +async def list_tools() -> list[types.Tool]: + """List available tools.""" + return [ + types.Tool( + name="advanced_tool", + description="Tool with full control including _meta field", + inputSchema={ + "type": "object", + "properties": {"message": {"type": "string"}}, + "required": ["message"], + }, + ) + ] + + +@server.call_tool() +async def handle_call_tool(name: str, arguments: dict[str, Any]) -> types.CallToolResult: + """Handle tool calls by returning CallToolResult directly.""" + if name == "advanced_tool": + message = str(arguments.get("message", "")) + return types.CallToolResult( + content=[types.TextContent(type="text", text=f"Processed: {message}")], + structuredContent={"result": "success", "message": message}, + _meta={"hidden": "data for client applications only"}, + ) + + raise ValueError(f"Unknown tool: {name}") + + +async def run(): + """Run the server.""" + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="example", + server_version="0.1.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions(), + experimental_capabilities={}, + ), + ), + ) + + +if __name__ == "__main__": + asyncio.run(run()) diff --git a/src/mcp/server/fastmcp/utilities/func_metadata.py b/src/mcp/server/fastmcp/utilities/func_metadata.py index 3289a5aa62..873b1ae19f 100644 --- a/src/mcp/server/fastmcp/utilities/func_metadata.py +++ b/src/mcp/server/fastmcp/utilities/func_metadata.py @@ -1,9 +1,10 @@ import inspect import json +import types from collections.abc import Awaitable, Callable, Sequence from itertools import chain from types import GenericAlias -from typing import Annotated, Any, ForwardRef, cast, get_args, get_origin, get_type_hints +from typing import Annotated, Any, ForwardRef, Union, cast, get_args, get_origin, get_type_hints import pydantic_core from pydantic import ( @@ -22,7 +23,7 @@ from mcp.server.fastmcp.exceptions import InvalidSignature from mcp.server.fastmcp.utilities.logging import get_logger from mcp.server.fastmcp.utilities.types import Audio, Image -from mcp.types import ContentBlock, TextContent +from mcp.types import CallToolResult, ContentBlock, TextContent logger = get_logger(__name__) @@ -104,6 +105,12 @@ def convert_result(self, result: Any) -> Any: from function return values, whereas the lowlevel server simply serializes the structured output. """ + if isinstance(result, CallToolResult): + if self.output_schema is not None: + assert self.output_model is not None, "Output model must be set if output schema is defined" + self.output_model.model_validate(result.structuredContent) + return result + unstructured_content = _convert_to_content(result) if self.output_schema is None: @@ -268,6 +275,26 @@ def func_metadata( output_info = FieldInfo.from_annotation(_get_typed_annotation(sig.return_annotation, globalns)) annotation = output_info.annotation + # Reject CallToolResult in Union types (including Optional) + # Handle both typing.Union (Union[X, Y]) and types.UnionType (X | Y) + origin = get_origin(annotation) + if origin is Union or origin is types.UnionType: + args = get_args(annotation) + # Check if CallToolResult appears in the union (excluding None for Optional check) + if any(isinstance(arg, type) and issubclass(arg, CallToolResult) for arg in args if arg is not type(None)): + raise InvalidSignature( + f"Function {func.__name__}: CallToolResult cannot be used in Union or Optional types. " + "To return empty results, use: CallToolResult(content=[])" + ) + + # if the typehint is CallToolResult, the user either intends to return without validation + # or they provided validation as Annotated metadata + if isinstance(annotation, type) and issubclass(annotation, CallToolResult): + if output_info.metadata: + annotation = output_info.metadata[0] + else: + return FuncMetadata(arg_model=arguments_model) + output_model, output_schema, wrap_output = _try_create_model_and_schema(annotation, func.__name__, output_info) if output_model is None and structured_output is True: diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py index 2fec3381bc..9a4ae9c89f 100644 --- a/src/mcp/server/lowlevel/server.py +++ b/src/mcp/server/lowlevel/server.py @@ -480,7 +480,7 @@ def call_tool(self, *, validate_input: bool = True): def decorator( func: Callable[ ..., - Awaitable[UnstructuredContent | StructuredContent | CombinationContent], + Awaitable[UnstructuredContent | StructuredContent | CombinationContent | types.CallToolResult], ], ): logger.debug("Registering handler for CallToolRequest") @@ -504,7 +504,9 @@ async def handler(req: types.CallToolRequest): # output normalization unstructured_content: UnstructuredContent maybe_structured_content: StructuredContent | None - if isinstance(results, tuple) and len(results) == 2: + if isinstance(results, types.CallToolResult): + return types.ServerResult(results) + elif isinstance(results, tuple) and len(results) == 2: # tool returned both structured and unstructured content unstructured_content, maybe_structured_content = cast(CombinationContent, results) elif isinstance(results, dict): diff --git a/tests/server/fastmcp/test_func_metadata.py b/tests/server/fastmcp/test_func_metadata.py index 830cf816b0..793dfc3245 100644 --- a/tests/server/fastmcp/test_func_metadata.py +++ b/tests/server/fastmcp/test_func_metadata.py @@ -13,6 +13,7 @@ from pydantic import BaseModel, Field from mcp.server.fastmcp.utilities.func_metadata import func_metadata +from mcp.types import CallToolResult class SomeInputModelA(BaseModel): @@ -834,6 +835,93 @@ def func_returning_unannotated() -> UnannotatedClass: assert meta.output_schema is None +def test_tool_call_result_is_unstructured_and_not_converted(): + def func_returning_call_tool_result() -> CallToolResult: + return CallToolResult(content=[]) + + meta = func_metadata(func_returning_call_tool_result) + + assert meta.output_schema is None + assert isinstance(meta.convert_result(func_returning_call_tool_result()), CallToolResult) + + +def test_tool_call_result_annotated_is_structured_and_converted(): + class PersonClass(BaseModel): + name: str + + def func_returning_annotated_tool_call_result() -> Annotated[CallToolResult, PersonClass]: + return CallToolResult(content=[], structuredContent={"name": "Brandon"}) + + meta = func_metadata(func_returning_annotated_tool_call_result) + + assert meta.output_schema == { + "type": "object", + "properties": { + "name": {"title": "Name", "type": "string"}, + }, + "required": ["name"], + "title": "PersonClass", + } + assert isinstance(meta.convert_result(func_returning_annotated_tool_call_result()), CallToolResult) + + +def test_tool_call_result_annotated_is_structured_and_invalid(): + class PersonClass(BaseModel): + name: str + + def func_returning_annotated_tool_call_result() -> Annotated[CallToolResult, PersonClass]: + return CallToolResult(content=[], structuredContent={"person": "Brandon"}) + + meta = func_metadata(func_returning_annotated_tool_call_result) + + with pytest.raises(ValueError): + meta.convert_result(func_returning_annotated_tool_call_result()) + + +def test_tool_call_result_in_optional_is_rejected(): + """Test that Optional[CallToolResult] raises InvalidSignature""" + + from mcp.server.fastmcp.exceptions import InvalidSignature + + def func_optional_call_tool_result() -> CallToolResult | None: + return CallToolResult(content=[]) + + with pytest.raises(InvalidSignature) as exc_info: + func_metadata(func_optional_call_tool_result) + + assert "Union or Optional" in str(exc_info.value) + assert "CallToolResult" in str(exc_info.value) + + +def test_tool_call_result_in_union_is_rejected(): + """Test that Union[str, CallToolResult] raises InvalidSignature""" + + from mcp.server.fastmcp.exceptions import InvalidSignature + + def func_union_call_tool_result() -> str | CallToolResult: + return CallToolResult(content=[]) + + with pytest.raises(InvalidSignature) as exc_info: + func_metadata(func_union_call_tool_result) + + assert "Union or Optional" in str(exc_info.value) + assert "CallToolResult" in str(exc_info.value) + + +def test_tool_call_result_in_pipe_union_is_rejected(): + """Test that str | CallToolResult raises InvalidSignature""" + from mcp.server.fastmcp.exceptions import InvalidSignature + + def func_pipe_union_call_tool_result() -> str | CallToolResult: + return CallToolResult(content=[]) + + with pytest.raises(InvalidSignature) as exc_info: + func_metadata(func_pipe_union_call_tool_result) + + assert "Union or Optional" in str(exc_info.value) + assert "CallToolResult" in str(exc_info.value) + + def test_structured_output_with_field_descriptions(): """Test that Field descriptions are preserved in structured output""" diff --git a/tests/server/test_lowlevel_output_validation.py b/tests/server/test_lowlevel_output_validation.py index 7bcdf59d3d..04e8a93a96 100644 --- a/tests/server/test_lowlevel_output_validation.py +++ b/tests/server/test_lowlevel_output_validation.py @@ -391,6 +391,47 @@ async def test_callback(client_session: ClientSession) -> CallToolResult: assert result.structuredContent == {"sentiment": "positive", "confidence": 0.95} +@pytest.mark.anyio +async def test_tool_call_result(): + """Test returning ToolCallResult when no outputSchema is defined.""" + tools = [ + Tool( + name="get_info", + description="Get structured information", + inputSchema={ + "type": "object", + "properties": {}, + }, + # No outputSchema for direct return of tool call result + ) + ] + + async def call_tool_handler(name: str, arguments: dict[str, Any]) -> CallToolResult: + if name == "get_info": + return CallToolResult( + content=[TextContent(type="text", text="Results calculated")], + structuredContent={"status": "ok", "data": {"value": 42}}, + _meta={"some": "metadata"}, + ) + else: + raise ValueError(f"Unknown tool: {name}") + + async def test_callback(client_session: ClientSession) -> CallToolResult: + return await client_session.call_tool("get_info", {}) + + result = await run_tool_test(tools, call_tool_handler, test_callback) + + # Verify results + assert result is not None + assert not result.isError + assert len(result.content) == 1 + assert result.content[0].type == "text" + assert result.content[0].text == "Results calculated" + assert isinstance(result.content[0], TextContent) + assert result.structuredContent == {"status": "ok", "data": {"value": 42}} + assert result.meta == {"some": "metadata"} + + @pytest.mark.anyio async def test_output_schema_type_validation(): """Test outputSchema validates types correctly.""" diff --git a/tests/test_examples.py b/tests/test_examples.py index 59063f122f..78f6d34020 100644 --- a/tests/test_examples.py +++ b/tests/test_examples.py @@ -44,6 +44,23 @@ async def test_complex_inputs(): assert result.content[2].text == "charlie" +@pytest.mark.anyio +async def test_direct_call_tool_result_return(): + """Test the CallToolResult echo server""" + from examples.fastmcp.direct_call_tool_result_return import mcp + + async with client_session(mcp._mcp_server) as client: + result = await client.call_tool("echo", {"text": "hello"}) + assert len(result.content) == 1 + content = result.content[0] + assert isinstance(content, TextContent) + assert content.text == "hello" + assert result.structuredContent + assert result.structuredContent["text"] == "hello" + assert isinstance(result.meta, dict) + assert result.meta["some"] == "metadata" + + @pytest.mark.anyio async def test_desktop(monkeypatch: pytest.MonkeyPatch): """Test the desktop server"""