Preserve raw CallToolResult returns (#4587)

Co-authored-by: nate nowack <thrast36@gmail.com>

fastmcp_slim/fastmcp/tools/base.py

@@ -21,7 +21,7 @@ from mcp_types import (
     ToolExecution,
 )
 from mcp_types import Tool as MCPTool
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, Field, PrivateAttr, model_validator
 from pydantic.json_schema import SkipJsonSchema
 
 from fastmcp.utilities.authorization import AuthCheck
@@ -87,6 +87,8 @@ def default_serializer(data: Any) -> str:
 
 
 class ToolResult(BaseModel):
+    _raw_mcp_result: CallToolResult | None = PrivateAttr(default=None)
+
     content: list[ContentBlock] = Field(
         description="List of content blocks for the tool result"
     )
@@ -152,11 +154,26 @@ class ToolResult(BaseModel):
             is_error=is_error,
         )
 
+    @classmethod
+    def from_mcp_result(cls, result: CallToolResult) -> ToolResult:
+        """Wrap a protocol result while preserving its exact wire representation."""
+        tool_result = cls(
+            content=result.content,
+            structured_content=result.structured_content,
+            meta=result.meta,
+            is_error=result.is_error,
+        )
+        tool_result._raw_mcp_result = result
+        return tool_result
+
     def to_mcp_result(
         self,
     ) -> (
         list[ContentBlock] | tuple[list[ContentBlock], dict[str, Any]] | CallToolResult
     ):
+        if self._raw_mcp_result is not None:
+            return self._raw_mcp_result
+
         # An error result must round-trip through CallToolResult so isError
         # reaches the client; the plain content/tuple returns can't carry it.
         if self.meta is not None or self.is_error:
@@ -346,6 +363,9 @@ class Tool(FastMCPComponent):
         if isinstance(raw_value, ToolResult):
             return raw_value
 
+        if isinstance(raw_value, CallToolResult):
+            return ToolResult.from_mcp_result(raw_value)
+
         if _HAS_PREFAB:
             if isinstance(raw_value, _PrefabApp):
                 return _prefab_to_tool_result(

fastmcp_slim/fastmcp/tools/function_parsing.py

@@ -393,6 +393,13 @@ class ParsedFunction:
             if is_class_member_of_type(output_type, ToolResult):
                 output_type = _UnserializableType
 
+            # A bare CallToolResult gives the tool full protocol-level control
+            # over its response, so there is no FastMCP output schema to infer.
+            if isinstance(output_type, type) and issubclass(
+                output_type, mcp_types.CallToolResult
+            ):
+                output_type = _UnserializableType
+
             # If InputRequiredResult survives stripping in any wrapping — bare,
             # via a `type X = ...` alias, Annotated, or a subclass — it is a
             # guard-only return with no output data (a union would have had its

tests/tools/tool/test_output_schema.py

@@ -3,7 +3,13 @@ from typing import Annotated, Any
 
 import pytest
 from inline_snapshot import snapshot
-from mcp_types import AudioContent, EmbeddedResource, ImageContent, TextContent
+from mcp_types import (
+    AudioContent,
+    CallToolResult,
+    EmbeddedResource,
+    ImageContent,
+    TextContent,
+)
 from pydantic import AnyUrl, BaseModel, Field, TypeAdapter
 from typing_extensions import TypedDict
 
@@ -130,6 +136,13 @@ class TestToolFromFunctionOutputSchema:
         tool = Tool.from_function(func)
         assert tool.output_schema is None
 
+    async def test_call_tool_result_return_annotation_no_output_schema(self):
+        def func() -> CallToolResult:
+            return CallToolResult(content=[])
+
+        tool = Tool.from_function(func)
+        assert tool.output_schema is None
+
     async def test_tool_result_subclass_return_annotation_no_output_schema(self):
         class MyToolResult(ToolResult):
             def __init__(self, data: str):

tests/tools/tool/test_results.py

@@ -123,6 +123,38 @@ class TestToolResultIsError:
         assert result.is_error is True
         assert result.content[0].text == "upstream boom"
 
+    def test_raw_call_tool_result_is_preserved(self):
+        tool = Tool.from_function(lambda: None, name="test_tool")
+        raw_result = CallToolResult(
+            content=[TextContent(type="text", text="upstream boom")],
+            structured_content={"code": 42},
+            is_error=True,
+            _meta={"source": "upstream"},
+        )
+
+        result = tool.convert_result(raw_result)
+
+        assert result.to_mcp_result() is raw_result
+
+    async def test_raw_call_tool_result_preserves_protocol_fields(self):
+        mcp = FastMCP()
+
+        raw_result = CallToolResult(
+            content=[TextContent(type="text", text="upstream boom")],
+            structured_content={"code": 42},
+            is_error=True,
+            _meta={"source": "upstream"},
+        )
+
+        @mcp.tool
+        def failing() -> CallToolResult:
+            return raw_result
+
+        async with Client(mcp) as client:
+            result = await client.call_tool_mcp("failing", {})
+
+        assert result.model_dump(by_alias=True) == raw_result.model_dump(by_alias=True)
+
 
 class TestUnionReturnTypes:
     """Tests for tools with union return types."""