Return CallToolResult from MCPServer.call_tool

Move the conversion of the internal tool-result shapes (bare content
sequence and (content, structured_content) tuple) out of the low-level
handler and into call_tool, so the public API has a single CallToolResult
return type. Type the conversion chain with a ToolResult alias and
convert_result overloads keyed on the convert_result flag, removing the
cast and type: ignore the previous approach needed.

Author: Kludex

docs/migration.md

@@ -8,16 +8,16 @@ Version 2 of the MCP Python SDK introduces several breaking changes to improve t
 
 ## Breaking Changes
 
-### `MCPServer.call_tool()` return annotation corrected
+### `MCPServer.call_tool()` returns `CallToolResult`
 
-`MCPServer.call_tool()` no longer advertises a raw `dict[str, Any]`
-return. On v2 it returns exactly the shapes produced by the MCPServer
-tool conversion path: a direct `CallToolResult`, a sequence of
-`ContentBlock` values for unstructured tools, or a
-`(content, structured_content)` tuple for structured tools.
+`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 subclass `MCPServer` or annotate wrappers around `call_tool()`,
-update those annotations to match the corrected return shape.
+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
 

src/mcp/server/mcpserver/server.py

@@ -5,9 +5,9 @@
 import base64
 import inspect
 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, TypeAlias, TypeVar, cast, overload
+from typing import Any, Generic, Literal, TypeVar, overload
 
 import anyio
 import pydantic_core
@@ -50,7 +50,6 @@
     CompleteRequestParams,
     CompleteResult,
     Completion,
-    ContentBlock,
     GetPromptRequestParams,
     GetPromptResult,
     Icon,
@@ -75,8 +74,6 @@
 
 _CallableT = TypeVar("_CallableT", bound=Callable[..., Any])
 
-ToolResult: TypeAlias = CallToolResult | Sequence[ContentBlock] | tuple[Sequence[ContentBlock], dict[str, Any]]
-
 
 class Settings(BaseSettings, Generic[LifespanResultT]):
     """MCPServer settings.
@@ -310,20 +307,11 @@ async def _handle_call_tool(
     ) -> 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 = cast(tuple[Sequence[ContentBlock], dict[str, Any]], result)
-            return CallToolResult(
-                content=list(unstructured_content),
-                structured_content=structured_content,
-            )
-        return CallToolResult(content=list(result))
 
     async def _handle_list_resources(
         self, ctx: ServerRequestContext[LifespanResultT], params: PaginatedRequestParams | None
@@ -392,18 +380,17 @@ async def list_tools(self) -> list[MCPTool]:
 
     async def call_tool(
         self, name: str, arguments: dict[str, Any], context: Context[LifespanResultT, Any] | None = None
-    ) -> ToolResult:
-        """Call a tool by name with arguments.
-
-        Returns:
-            The tool result converted for the low-level handler:
-            - a `CallToolResult` returned directly by the tool,
-            - a sequence of content blocks for unstructured tools, or
-            - a `(content, structured_content)` tuple for tools with structured output.
-        """
+    ) -> CallToolResult:
+        """Call a tool by name with arguments."""
         if context is None:
             context = Context(mcp_server=self)
-        return await self._tool_manager.call_tool(name, arguments, context, convert_result=True)
+        result = await self._tool_manager.call_tool(name, arguments, context, convert_result=True)
+        if isinstance(result, CallToolResult):
+            return result
+        if isinstance(result, tuple):
+            content, structured_content = result
+            return CallToolResult(content=list(content), structured_content=structured_content)
+        return CallToolResult(content=list(result))
 
     async def list_resources(self) -> list[MCPResource]:
         """List all available resources."""

src/mcp/server/mcpserver/tools/base.py

@@ -2,13 +2,13 @@
 
 from collections.abc import Callable
 from functools import cached_property
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal, overload
 
 from pydantic import BaseModel, Field
 
 from mcp.server.mcpserver.exceptions import ToolError
 from mcp.server.mcpserver.utilities.context_injection import find_context_parameter
-from mcp.server.mcpserver.utilities.func_metadata import FuncMetadata, func_metadata
+from mcp.server.mcpserver.utilities.func_metadata import FuncMetadata, ToolResult, func_metadata
 from mcp.shared._callable_inspection import is_async_callable
 from mcp.shared.exceptions import UrlElicitationRequiredError
 from mcp.shared.tool_name_validation import validate_and_warn_tool_name
@@ -88,6 +88,17 @@ def from_function(
             meta=meta,
         )
 
+    @overload
+    async def run(
+        self, arguments: dict[str, Any], context: Context[LifespanContextT, RequestT], convert_result: Literal[True]
+    ) -> ToolResult: ...
+    @overload
+    async def run(
+        self,
+        arguments: dict[str, Any],
+        context: Context[LifespanContextT, RequestT],
+        convert_result: Literal[False] = False,
+    ) -> Any: ...
     async def run(
         self,
         arguments: dict[str, Any],

src/mcp/server/mcpserver/tools/tool_manager.py

@@ -1,10 +1,11 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal, overload
 
 from mcp.server.mcpserver.exceptions import ToolError
 from mcp.server.mcpserver.tools.base import Tool
+from mcp.server.mcpserver.utilities.func_metadata import ToolResult
 from mcp.server.mcpserver.utilities.logging import get_logger
 from mcp.types import Icon, ToolAnnotations
 
@@ -71,6 +72,22 @@ def remove_tool(self, name: str) -> None:
             raise ToolError(f"Unknown tool: {name}")
         del self._tools[name]
 
+    @overload
+    async def call_tool(
+        self,
+        name: str,
+        arguments: dict[str, Any],
+        context: Context[LifespanContextT, RequestT],
+        convert_result: Literal[True],
+    ) -> ToolResult: ...
+    @overload
+    async def call_tool(
+        self,
+        name: str,
+        arguments: dict[str, Any],
+        context: Context[LifespanContextT, RequestT],
+        convert_result: Literal[False] = False,
+    ) -> Any: ...
     async def call_tool(
         self,
         name: str,

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

@@ -4,7 +4,7 @@
 from collections.abc import Awaitable, Callable, Sequence
 from itertools import chain
 from types import GenericAlias
-from typing import Annotated, Any, cast, get_args, get_origin, get_type_hints
+from typing import Annotated, Any, TypeAlias, cast, get_args, get_origin, get_type_hints
 
 import anyio
 import anyio.to_thread
@@ -28,6 +28,8 @@
 
 logger = get_logger(__name__)
 
+ToolResult: TypeAlias = CallToolResult | list[ContentBlock] | tuple[list[ContentBlock], dict[str, Any]]
+
 
 class StrictJsonSchema(GenerateJsonSchema):
     """A JSON schema generator that raises exceptions instead of emitting warnings.
@@ -88,7 +90,7 @@ async def call_fn_with_arg_validation(
         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) -> ToolResult:
         """Convert a function call result to the format for the lowlevel tool call handler.
 
         - If output_model is None, return the unstructured content directly.
@@ -496,7 +498,7 @@ class DictModel(RootModel[dict_annotation]):
     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,9 @@ def func_with_aliases() -> ModelWithAliases:  # pragma: no cover
 
     # Check that the actual output uses aliases too
     result = ModelWithAliases(**{"first": "hello", "second": "world"})
-    _, structured_content = meta.convert_result(result)
+    converted = meta.convert_result(result)
+    assert isinstance(converted, tuple)
+    _, structured_content = converted
 
     # The structured content should use aliases to match the schema
     assert "first" in structured_content
@@ -1050,7 +1052,9 @@ def func_with_aliases() -> ModelWithAliases:  # pragma: no cover
 
     # 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)
+    converted_defaults = meta.convert_result(result_with_defaults)
+    assert isinstance(converted_defaults, tuple)
+    _, structured_content_defaults = converted_defaults
 
     # Even with defaults, should use aliases in output
     assert "first" in structured_content_defaults

tests/server/mcpserver/test_server.py

@@ -305,12 +305,12 @@ async def test_tool_return_value_conversion(self):
             assert result.structured_content is not None
             assert result.structured_content == {"result": 3}
 
-    async def test_call_tool_returns_declared_result_shapes(self):
+    async def test_call_tool_always_returns_call_tool_result(self):
         mcp = MCPServer()
 
         @mcp.tool()
-        def direct_result() -> CallToolResult:
-            return CallToolResult(content=[TextContent(text="direct")])
+        def direct() -> CallToolResult:
+            return CallToolResult(content=[TextContent(type="text", text="direct")])
 
         @mcp.tool(structured_output=False)
         def unstructured() -> str:
@@ -320,14 +320,13 @@ def unstructured() -> str:
         def structured() -> int:
             return 3
 
-        direct = await mcp.call_tool("direct_result", {})
-        assert direct == CallToolResult(content=[TextContent(text="direct")])
-
-        bare_content = await mcp.call_tool("unstructured", {})
-        assert bare_content == [TextContent(text="plain")]
-
-        structured_result = await mcp.call_tool("structured", {})
-        assert structured_result == ([TextContent(text="3")], {"result": 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

tests/server/mcpserver/test_tool_manager.py

@@ -455,7 +455,8 @@ def get_user(user_id: int) -> UserOutput:
         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, tuple)
+        assert result[1] == {"name": "John", "age": 30}
 
     @pytest.mark.anyio
     async def test_tool_with_primitive_output(self):
@@ -470,6 +471,7 @@ def double_number(n: int) -> int:
         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, tuple)
         assert isinstance(result[0][0], TextContent) and result[1] == {"result": 10}
 
     @pytest.mark.anyio
@@ -510,7 +512,8 @@ def get_person() -> Person:
         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, tuple)
+        assert result[1] == expected_output
 
     @pytest.mark.anyio
     async def test_tool_with_list_output(self):
@@ -528,6 +531,7 @@ def get_numbers() -> list[int]:
         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, tuple)
         assert isinstance(result[0][0], TextContent) and result[1] == expected_output
 
     @pytest.mark.anyio