fix(types): type converted MCPServer handler results

Author: lavish0000

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

@@ -69,14 +69,14 @@ class Prompt(BaseModel):
     title: str | None = Field(None, description="Human-readable title of the prompt")
     description: str | None = Field(None, description="Description of what the prompt does")
     arguments: list[PromptArgument] | None = Field(None, description="Arguments that can be passed to the prompt")
-    fn: Callable[..., PromptResult | Awaitable[PromptResult]] = Field(exclude=True)
+    fn: Callable[..., PromptResult] = Field(exclude=True)
     icons: list[Icon] | None = Field(default=None, description="Optional list of icons for this prompt")
     context_kwarg: str | None = Field(None, description="Name of the kwarg that should receive context", exclude=True)
 
     @classmethod
     def from_function(
         cls,
-        fn: Callable[..., PromptResult | Awaitable[PromptResult]],
+        fn: Callable[..., PromptResult],
         name: str | None = None,
         title: str | None = None,
         description: str | None = None,

src/mcp/server/mcpserver/resources/templates.py

@@ -82,7 +82,7 @@ def from_function(
             context_kwarg=context_kwarg,
         )
 
-    def matches(self, uri: str) -> dict[str, Any] | None:
+    def matches(self, uri: str) -> dict[str, str] | None:
         """Check if URI matches template and extract parameters.
 
         Extracted parameters are URL-decoded to handle percent-encoded characters.

src/mcp/server/mcpserver/server.py

@@ -4,11 +4,10 @@
 
 import base64
 import inspect
-import json
 import re
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
-from typing import Any, Generic, Literal, TypeVar, overload
+from typing import Any, Generic, Literal, TypeVar, cast, overload
 
 import anyio
 import pydantic_core
@@ -36,6 +35,7 @@
 from mcp.server.mcpserver.resources import FunctionResource, Resource, ResourceManager
 from mcp.server.mcpserver.tools import Tool, ToolManager
 from mcp.server.mcpserver.utilities.context_injection import find_context_parameter
+from mcp.server.mcpserver.utilities.func_metadata import ConvertedToolResult
 from mcp.server.mcpserver.utilities.logging import configure_logging, get_logger
 from mcp.server.sse import SseServerTransport
 from mcp.server.stdio import stdio_server
@@ -308,18 +308,13 @@ async def _handle_call_tool(
         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]
+            unstructured_content, structured_content = cast(
+                tuple[Sequence[ContentBlock], dict[str, Any]],
+                result,
             )
-        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,
+                content=list(unstructured_content),
+                structured_content=structured_content,
             )
         return CallToolResult(content=list(result))
 
@@ -390,7 +385,7 @@ async def list_tools(self) -> list[MCPTool]:
 
     async def call_tool(
         self, name: str, arguments: dict[str, Any], context: Context[LifespanResultT, Any] | None = None
-    ) -> Sequence[ContentBlock] | dict[str, Any]:
+    ) -> ConvertedToolResult:
         """Call a tool by name with arguments."""
         if context is None:
             context = Context(mcp_server=self)

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

@@ -4,13 +4,13 @@
 import inspect
 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 ConvertedToolResult, FuncMetadata, func_metadata
 from mcp.shared.exceptions import UrlElicitationRequiredError
 from mcp.shared.tool_name_validation import validate_and_warn_tool_name
 from mcp.types import Icon, ToolAnnotations
@@ -89,6 +89,22 @@ def from_function(
             meta=meta,
         )
 
+    @overload
+    async def run(
+        self,
+        arguments: dict[str, Any],
+        context: Context[LifespanContextT, RequestT],
+        convert_result: Literal[True],
+    ) -> ConvertedToolResult: ...
+
+    @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 ConvertedToolResult
 from mcp.server.mcpserver.utilities.logging import get_logger
 from mcp.types import Icon, ToolAnnotations
 
@@ -77,6 +78,24 @@ 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],
+    ) -> ConvertedToolResult: ...
+
+    @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__)
 
+ConvertedToolResult: TypeAlias = CallToolResult | Sequence[ContentBlock] | tuple[Sequence[ContentBlock], dict[str, Any]]
+
 
 class StrictJsonSchema(GenerateJsonSchema):
     """A JSON schema generator that raises exceptions instead of emitting warnings.