Simplify with early return because output schemas won't work with a CallToolResult return type

Author: BrandonShar

examples/fastmcp/direct_call_tool_result_return.py

@@ -0,0 +1,16 @@
+"""
+FastMCP Echo Server with direct CallToolResult return
+"""
+
+from mcp.server.fastmcp import FastMCP
+from mcp.types import CallToolResult, TextContent
+
+mcp = FastMCP("Echo Server")
+
+
+@mcp.tool()
+def echo(text: str) -> CallToolResult:
+    """Echo the input text with structure and metadata"""
+    return CallToolResult(
+        content=[TextContent(type="text", text=text)], structuredContent={"text": text}, _meta={"some": "metadata"}
+    )

src/mcp/server/fastmcp/utilities/func_metadata.py

@@ -22,7 +22,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 +104,9 @@ def convert_result(self, result: Any) -> Any:
         from function return values, whereas the lowlevel server simply serializes
         the structured output.
         """
+        if isinstance(result, CallToolResult):
+            return result
+
         unstructured_content = _convert_to_content(result)
 
         if self.output_schema is None:
@@ -268,6 +271,10 @@ def func_metadata(
     output_info = FieldInfo.from_annotation(_get_typed_annotation(sig.return_annotation, globalns))
     annotation = output_info.annotation
 
+    # if the typehint is CallToolResult, the user intends to return it directly
+    if isinstance(annotation, type) and issubclass(annotation, CallToolResult):
+        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:

src/mcp/server/lowlevel/server.py

@@ -502,46 +502,45 @@ async def handler(req: types.CallToolRequest):
                     results = await func(tool_name, arguments)
 
                     # output normalization
-                    casted_results: types.CallToolResult
                     unstructured_content: UnstructuredContent
                     maybe_structured_content: StructuredContent | None
                     if isinstance(results, types.CallToolResult):
-                        casted_results = results
+                        # tool returned a CallToolResult so we'll skip further validation and return it directly
+                        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)
-                        casted_results = types.CallToolResult(
-                            content=list(unstructured_content),
-                            structuredContent=maybe_structured_content,
-                        )
                     elif isinstance(results, dict):
                         # tool returned structured content only
-                        casted_results = types.CallToolResult(
-                            content=[types.TextContent(type="text", text=json.dumps(results, indent=2))],
-                            structuredContent=cast(StructuredContent, results),
-                        )
+                        maybe_structured_content = cast(StructuredContent, results)
+                        unstructured_content = [types.TextContent(type="text", text=json.dumps(results, indent=2))]
                     elif hasattr(results, "__iter__"):
                         # tool returned unstructured content only
-                        casted_results = types.CallToolResult(
-                            content=list(cast(UnstructuredContent, results)),
-                        )
+                        unstructured_content = cast(UnstructuredContent, results)
+                        maybe_structured_content = None
                     else:
                         return self._make_error_result(f"Unexpected return type from tool: {type(results).__name__}")
 
                     # output validation
                     if tool and tool.outputSchema is not None:
-                        if casted_results.structuredContent is None:
+                        if maybe_structured_content is None:
                             return self._make_error_result(
                                 "Output validation error: outputSchema defined but no structured output returned"
                             )
                         else:
                             try:
-                                jsonschema.validate(instance=casted_results.structuredContent, schema=tool.outputSchema)
+                                jsonschema.validate(instance=maybe_structured_content, schema=tool.outputSchema)
                             except jsonschema.ValidationError as e:
                                 return self._make_error_result(f"Output validation error: {e.message}")
 
                     # result
-                    return types.ServerResult(casted_results)
+                    return types.ServerResult(
+                        types.CallToolResult(
+                            content=list(unstructured_content),
+                            structuredContent=maybe_structured_content,
+                            isError=False,
+                        )
+                    )
                 except Exception as e:
                     return self._make_error_result(str(e))
 

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,16 @@ 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_structured_output_with_field_descriptions():
     """Test that Field descriptions are preserved in structured output"""
 

tests/server/test_lowlevel_output_validation.py

@@ -392,7 +392,7 @@ async def test_callback(client_session: ClientSession) -> CallToolResult:
 
 
 @pytest.mark.anyio
-async def test_tool_call_result_without_output_schema():
+async def test_tool_call_result():
     """Test returning ToolCallResult when no outputSchema is defined."""
     tools = [
         Tool(
@@ -402,7 +402,7 @@ async def test_tool_call_result_without_output_schema():
                 "type": "object",
                 "properties": {},
             },
-            # No outputSchema defined
+            # No outputSchema for direct return of tool call result
         )
     ]
 
@@ -432,100 +432,6 @@ async def test_callback(client_session: ClientSession) -> CallToolResult:
     assert result.meta == {"some": "metadata"}
 
 
-@pytest.mark.anyio
-async def test_valid_tool_call_result_with_output_schema():
-    """Test returning ToolCallResult when no outputSchema is defined."""
-    tools = [
-        Tool(
-            name="get_info",
-            description="Get structured information",
-            inputSchema={
-                "type": "object",
-                "properties": {},
-            },
-            outputSchema={
-                "type": "object",
-                "properties": {
-                    "name": {"type": "string"},
-                    "number": {"type": "integer"},
-                },
-                "required": ["name", "number"],
-            },
-        )
-    ]
-
-    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={"name": "Brandon", "number": 29},
-                _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 == {"name": "Brandon", "number": 29}
-    assert result.meta == {"some": "metadata"}
-
-
-@pytest.mark.anyio
-async def test_invalid_tool_call_result_with_output_schema():
-    """Test returning ToolCallResult when no outputSchema is defined."""
-    tools = [
-        Tool(
-            name="get_info",
-            description="Get structured information",
-            inputSchema={
-                "type": "object",
-                "properties": {},
-            },
-            outputSchema={
-                "type": "object",
-                "properties": {
-                    "name": {"type": "string"},
-                    "number": {"type": "integer"},
-                },
-                "required": ["name", "number"],
-            },
-        )
-    ]
-
-    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={"name": "Brandon"},
-                _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)
-
-    assert result is not None
-    assert result.isError
-    assert len(result.content) == 1
-    assert result.content[0].type == "text"
-    assert isinstance(result.content[0], TextContent)
-    assert "Output validation error:" in result.content[0].text
-    assert "'number' is a required property" in result.content[0].text
-
-
 @pytest.mark.anyio
 async def test_output_schema_type_validation():
     """Test outputSchema validates types correctly."""

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"""