fix: correct MCPServer call_tool result type (#2816)

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

docs/migration.md

@@ -8,6 +8,17 @@ Version 2 of the MCP Python SDK introduces several breaking changes to improve t
 
 ## Breaking Changes
 
+### `MCPServer.call_tool()` returns `CallToolResult`
+
+`MCPServer.call_tool()` now always returns a `CallToolResult`. It previously
+advertised `Sequence[ContentBlock] | dict[str, Any]` and leaked the internal
+conversion shapes (a bare content sequence or a `(content, structured_content)`
+tuple), forcing callers to re-assemble a `CallToolResult` themselves.
+
+If you call `MCPServer.call_tool()` directly, read `.content` and
+`.structured_content` off the returned `CallToolResult` instead of branching on
+the result type.
+
 ### `streamablehttp_client` removed
 
 The deprecated `streamablehttp_client` function has been removed. Use `streamable_http_client` instead.

src/mcp/server/mcpserver/server.py

@@ -4,9 +4,8 @@ from __future__ import annotations
 
 import base64
 import inspect
-import json
 import re
-from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from typing import Any, Generic, Literal, TypeVar, overload
 
@@ -53,7 +52,6 @@ from mcp.types import (
     CompleteRequestParams,
     CompleteResult,
     Completion,
-    ContentBlock,
     GetPromptRequestParams,
     GetPromptResult,
     Icon,
@@ -311,28 +309,11 @@ class MCPServer(Generic[LifespanResultT]):
     ) -> CallToolResult:
         context = Context(request_context=ctx, mcp_server=self)
         try:
-            result = await self.call_tool(params.name, params.arguments or {}, context)
+            return await self.call_tool(params.name, params.arguments or {}, context)
         except MCPError:
             raise
         except Exception as e:
             return CallToolResult(content=[TextContent(type="text", text=str(e))], is_error=True)
-        if isinstance(result, CallToolResult):
-            return result
-        if isinstance(result, tuple) and len(result) == 2:
-            unstructured_content, structured_content = result
-            return CallToolResult(
-                content=list(unstructured_content),  # type: ignore[arg-type]
-                structured_content=structured_content,  # type: ignore[arg-type]
-            )
-        if isinstance(result, dict):  # pragma: no cover
-            # TODO: this code path is unreachable — convert_result never returns a raw dict.
-            # The call_tool return type (Sequence[ContentBlock] | dict[str, Any]) is wrong
-            # and needs to be cleaned up.
-            return CallToolResult(
-                content=[TextContent(type="text", text=json.dumps(result, indent=2))],
-                structured_content=result,
-            )
-        return CallToolResult(content=list(result))
 
     async def _handle_list_resources(
         self, ctx: ServerRequestContext[LifespanResultT], params: PaginatedRequestParams | None
@@ -406,7 +387,7 @@ class MCPServer(Generic[LifespanResultT]):
 
     async def call_tool(
         self, name: str, arguments: dict[str, Any], context: Context[LifespanResultT, Any] | None = None
-    ) -> Sequence[ContentBlock] | dict[str, Any]:
+    ) -> CallToolResult:
         """Call a tool by name with arguments."""
         if context is None:
             context = Context(mcp_server=self)

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

@@ -88,14 +88,10 @@ class FuncMetadata(BaseModel):
         else:
             return await anyio.to_thread.run_sync(functools.partial(fn, **arguments_parsed_dict))
 
-    def convert_result(self, result: Any) -> Any:
+    def convert_result(self, result: Any) -> CallToolResult:
-        """Convert a function call result to the format for the lowlevel tool call handler.
+        """Convert a function call result into a `CallToolResult`.
 
-        - If output_model is None, return the unstructured content directly.
+        Note: we build unstructured content here **even though the lowlevel server
-        - If output_model is not None, convert the result to structured output format
-            (dict[str, Any]) and return both unstructured and structured content.
-
-        Note: we return unstructured content here **even though the lowlevel server
         tool call handler provides generic backwards compatibility serialization of
         structured content**. This is for MCPServer backwards compatibility: we need to
         retain MCPServer's ad hoc conversion logic for constructing unstructured output
@@ -111,16 +107,16 @@ class FuncMetadata(BaseModel):
         unstructured_content = _convert_to_content(result)
 
         if self.output_schema is None:
-            return unstructured_content
+            return CallToolResult(content=unstructured_content)
-        else:
-            if self.wrap_output:
-                result = {"result": result}
 
-            assert self.output_model is not None, "Output model must be set if output schema is defined"
+        if self.wrap_output:
-            validated = self.output_model.model_validate(result)
+            result = {"result": result}
-            structured_content = validated.model_dump(mode="json", by_alias=True)
 
-            return (unstructured_content, structured_content)
+        assert self.output_model is not None, "Output model must be set if output schema is defined"
+        validated = self.output_model.model_validate(result)
+        structured_content = validated.model_dump(mode="json", by_alias=True)
+
+        return CallToolResult(content=unstructured_content, structured_content=structured_content)
 
     def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]:
         """Pre-parse data from JSON.
@@ -496,7 +492,7 @@ def _create_dict_model(func_name: str, dict_annotation: Any) -> type[BaseModel]:
     return DictModel
 
 
-def _convert_to_content(result: Any) -> Sequence[ContentBlock]:
+def _convert_to_content(result: Any) -> list[ContentBlock]:
     """Convert a result to a sequence of content objects.
 
     Note: This conversion logic comes from previous versions of MCPServer and is being

tests/server/mcpserver/test_func_metadata.py

@@ -1038,7 +1038,8 @@ def test_structured_output_aliases():
 
     # Check that the actual output uses aliases too
     result = ModelWithAliases(**{"first": "hello", "second": "world"})
-    _, structured_content = meta.convert_result(result)
+    structured_content = meta.convert_result(result).structured_content
+    assert structured_content is not None
 
     # The structured content should use aliases to match the schema
     assert "first" in structured_content
@@ -1050,7 +1051,8 @@ def test_structured_output_aliases():
 
     # Also test the case where we have a model with defaults to ensure aliases work in all cases
     result_with_defaults = ModelWithAliases()  # Uses default None values
-    _, structured_content_defaults = meta.convert_result(result_with_defaults)
+    structured_content_defaults = meta.convert_result(result_with_defaults).structured_content
+    assert structured_content_defaults is not None
 
     # Even with defaults, should use aliases in output
     assert "first" in structured_content_defaults

tests/server/mcpserver/test_server.py

@@ -23,6 +23,7 @@ from mcp.types import (
     INVALID_PARAMS,
     AudioContent,
     BlobResourceContents,
+    CallToolResult,
     Completion,
     CompletionArgument,
     CompletionContext,
@@ -306,6 +307,29 @@ class TestServerTools:
             assert result.structured_content is not None
             assert result.structured_content == {"result": 3}
 
+    async def test_call_tool_always_returns_call_tool_result(self):
+        mcp = MCPServer()
+
+        @mcp.tool()
+        def direct() -> CallToolResult:
+            return CallToolResult(content=[TextContent(type="text", text="direct")])
+
+        @mcp.tool(structured_output=False)
+        def unstructured() -> str:
+            return "plain"
+
+        @mcp.tool()
+        def structured() -> int:
+            return 3
+
+        assert await mcp.call_tool("direct", {}) == CallToolResult(content=[TextContent(type="text", text="direct")])
+        assert await mcp.call_tool("unstructured", {}) == CallToolResult(
+            content=[TextContent(type="text", text="plain")]
+        )
+        assert await mcp.call_tool("structured", {}) == CallToolResult(
+            content=[TextContent(type="text", text="3")], structured_content={"result": 3}
+        )
+
     async def test_tool_image_helper(self, tmp_path: Path):
         # Create a test image
         image_path = tmp_path / "test.png"

tests/server/mcpserver/test_tool_manager.py

@@ -11,7 +11,7 @@ from mcp.server.mcpserver import Context, MCPServer
 from mcp.server.mcpserver.exceptions import ToolError
 from mcp.server.mcpserver.tools import Tool, ToolManager
 from mcp.server.mcpserver.utilities.func_metadata import ArgModelBase, FuncMetadata
-from mcp.types import TextContent, ToolAnnotations
+from mcp.types import CallToolResult, TextContent, ToolAnnotations
 
 
 class TestAddTools:
@@ -455,7 +455,8 @@ class TestStructuredOutput:
         manager.add_tool(get_user)
         result = await manager.call_tool("get_user", {"user_id": 1}, Context(), convert_result=True)
         # don't test unstructured output here, just the structured conversion
-        assert len(result) == 2 and result[1] == {"name": "John", "age": 30}
+        assert isinstance(result, CallToolResult)
+        assert result.structured_content == {"name": "John", "age": 30}
 
     @pytest.mark.anyio
     async def test_tool_with_primitive_output(self):
@@ -470,7 +471,8 @@ class TestStructuredOutput:
         result = await manager.call_tool("double_number", {"n": 5}, Context())
         assert result == 10
         result = await manager.call_tool("double_number", {"n": 5}, Context(), convert_result=True)
-        assert isinstance(result[0][0], TextContent) and result[1] == {"result": 10}
+        assert isinstance(result, CallToolResult)
+        assert isinstance(result.content[0], TextContent) and result.structured_content == {"result": 10}
 
     @pytest.mark.anyio
     async def test_tool_with_typeddict_output(self):
@@ -510,7 +512,8 @@ class TestStructuredOutput:
         manager.add_tool(get_person)
         result = await manager.call_tool("get_person", {}, Context(), convert_result=True)
         # don't test unstructured output here, just the structured conversion
-        assert len(result) == 2 and result[1] == expected_output
+        assert isinstance(result, CallToolResult)
+        assert result.structured_content == expected_output
 
     @pytest.mark.anyio
     async def test_tool_with_list_output(self):
@@ -528,7 +531,8 @@ class TestStructuredOutput:
         result = await manager.call_tool("get_numbers", {}, Context())
         assert result == expected_list
         result = await manager.call_tool("get_numbers", {}, Context(), convert_result=True)
-        assert isinstance(result[0][0], TextContent) and result[1] == expected_output
+        assert isinstance(result, CallToolResult)
+        assert isinstance(result.content[0], TextContent) and result.structured_content == expected_output
 
     @pytest.mark.anyio
     async def test_tool_without_structured_output(self):