address comments

Author: bhosmer-ant

README.md

@@ -834,6 +834,8 @@ Caution: The `mcp run` and `mcp dev` tool doesn't support low-level server.
 The low-level server supports structured output for tools, allowing you to return both human-readable content and machine-readable structured data. Tools can define an `outputSchema` to validate their structured output:
 
 ```python
+from types import Any
+
 import mcp.types as types
 from mcp.server.lowlevel import Server
 
@@ -866,29 +868,25 @@ async def list_tools() -> list[types.Tool]:
 
 
 @server.call_tool()
-async def call_tool(name: str, arguments: dict) -> tuple[list[types.TextContent], dict]:
+async def call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
     if name == "calculate":
         expression = arguments["expression"]
         try:
-            result = eval(expression)  # Note: Use a safe math parser in production
-
-            # Return both human-readable content and structured data
-            content = [
-                types.TextContent(
-                    type="text", text=f"The result of {expression} is {result}"
-                )
-            ]
+            result = eval(expression)  # Use a safe math parser
             structured = {"result": result, "expression": expression}
 
-            return (content, structured)
+            # low-level server will validate structured output against the tool's
+            # output schema, and automatically serialize it into a TextContent block
+            # for backwards compatibility with pre-2025-06-18 clients.
+            return structured
         except Exception as e:
             raise ValueError(f"Calculation error: {str(e)}")
 ```
 
 Tools can return data in three ways:
-1. **Content only**: Return a list of content blocks (default behavior)
-2. **Structured data only**: Return a dictionary that will be serialized to JSON
-3. **Both**: Return a tuple of (content, structured_data)
+1. **Content only**: Return a list of content blocks (default behavior before spec revision 2025-06-18)
+2. **Structured data only**: Return a dictionary that will be serialized to JSON (Introduced in spec revision 2025-06-18)
+3. **Both**: Return a tuple of (content, structured_data) preferred option to use for backwards compatibility
 
 When an `outputSchema` is defined, the server automatically validates the structured output against the schema. This ensures type safety and helps catch errors early.
 

examples/servers/structured_output_lowlevel.py

@@ -2,15 +2,9 @@
 """
 Example low-level MCP server demonstrating structured output support.
 
-This example shows how to use the low-level server API to return both
-human-readable content and machine-readable structured data from tools,
-with automatic validation against output schemas.
-
-The low-level API provides direct control over request handling and
-allows tools to return different types of responses:
-1. Content only (list of content blocks)
-2. Structured data only (dict that gets serialized to JSON)
-3. Both content and structured data (tuple)
+This example shows how to use the low-level server API to return
+structured data from tools, with automatic validation against output
+schemas.
 """
 
 import asyncio
@@ -30,32 +24,6 @@
 async def list_tools() -> list[types.Tool]:
     """List available tools with their schemas."""
     return [
-        types.Tool(
-            name="analyze_text",
-            description="Analyze text and return structured insights",
-            inputSchema={
-                "type": "object",
-                "properties": {"text": {"type": "string", "description": "Text to analyze"}},
-                "required": ["text"],
-            },
-            outputSchema={
-                "type": "object",
-                "properties": {
-                    "word_count": {"type": "integer"},
-                    "char_count": {"type": "integer"},
-                    "sentence_count": {"type": "integer"},
-                    "most_common_words": {
-                        "type": "array",
-                        "items": {
-                            "type": "object",
-                            "properties": {"word": {"type": "string"}, "count": {"type": "integer"}},
-                            "required": ["word", "count"],
-                        },
-                    },
-                },
-                "required": ["word_count", "char_count", "sentence_count", "most_common_words"],
-            },
-        ),
         types.Tool(
             name="get_weather",
             description="Get weather information (simulated)",
@@ -76,91 +44,16 @@ async def list_tools() -> list[types.Tool]:
                 "required": ["temperature", "conditions", "humidity", "wind_speed", "timestamp"],
             },
         ),
-        types.Tool(
-            name="calculate_statistics",
-            description="Calculate statistics for a list of numbers",
-            inputSchema={
-                "type": "object",
-                "properties": {
-                    "numbers": {
-                        "type": "array",
-                        "items": {"type": "number"},
-                        "description": "List of numbers to analyze",
-                    }
-                },
-                "required": ["numbers"],
-            },
-            outputSchema={
-                "type": "object",
-                "properties": {
-                    "mean": {"type": "number"},
-                    "median": {"type": "number"},
-                    "min": {"type": "number"},
-                    "max": {"type": "number"},
-                    "sum": {"type": "number"},
-                    "count": {"type": "integer"},
-                },
-                "required": ["mean", "median", "min", "max", "sum", "count"],
-            },
-        ),
     ]
 
 
 @server.call_tool()
 async def call_tool(name: str, arguments: dict[str, Any]) -> Any:
     """
-    Handle tool calls with structured output.
-
-    This low-level handler demonstrates the three ways to return data:
-    1. Return a list of content blocks (traditional approach)
-    2. Return a dict (gets serialized to JSON and included as structuredContent)
-    3. Return a tuple of (content, structured_data) for both
+    Handle tool call with structured output.
     """
 
-    if name == "analyze_text":
-        text = arguments["text"]
-
-        # Analyze the text
-        words = text.split()
-        word_count = len(words)
-        char_count = len(text)
-        sentences = text.replace("?", ".").replace("!", ".").split(".")
-        sentence_count = len([s for s in sentences if s.strip()])
-
-        # Count word frequencies
-        word_freq = {}
-        for word in words:
-            word_lower = word.lower().strip('.,!?;:"')
-            if word_lower:
-                word_freq[word_lower] = word_freq.get(word_lower, 0) + 1
-
-        # Get top 5 most common words
-        most_common = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)[:5]
-        most_common_words = [{"word": word, "count": count} for word, count in most_common]
-
-        # Example 3: Return both content and structured data
-        # The low-level server will validate the structured data against outputSchema
-        content = [
-            types.TextContent(
-                type="text",
-                text=f"Text analysis complete:\n"
-                f"- {word_count} words\n"
-                f"- {char_count} characters\n"
-                f"- {sentence_count} sentences\n"
-                f"- Most common words: {', '.join(w['word'] for w in most_common_words)}",
-            )
-        ]
-
-        structured = {
-            "word_count": word_count,
-            "char_count": char_count,
-            "sentence_count": sentence_count,
-            "most_common_words": most_common_words,
-        }
-
-        return (content, structured)
-
-    elif name == "get_weather":
+    if name == "get_weather":
         # city = arguments["city"]  # Would be used with real weather API
 
         # Simulate weather data (in production, call a real weather API)
@@ -176,49 +69,10 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> Any:
             "timestamp": datetime.now().isoformat(),
         }
 
-        # Example 2: Return structured data only
+        # Return structured data only
         # The low-level server will serialize this to JSON content automatically
         return weather_data
 
-    elif name == "calculate_statistics":
-        numbers = arguments["numbers"]
-
-        if not numbers:
-            raise ValueError("Cannot calculate statistics for empty list")
-
-        sorted_nums = sorted(numbers)
-        count = len(numbers)
-
-        # Calculate statistics
-        mean = sum(numbers) / count
-
-        if count % 2 == 0:
-            median = (sorted_nums[count // 2 - 1] + sorted_nums[count // 2]) / 2
-        else:
-            median = sorted_nums[count // 2]
-
-        stats = {
-            "mean": mean,
-            "median": median,
-            "min": sorted_nums[0],
-            "max": sorted_nums[-1],
-            "sum": sum(numbers),
-            "count": count,
-        }
-
-        # Example 3: Return both content and structured data
-        content = [
-            types.TextContent(
-                type="text",
-                text=f"Statistics for {count} numbers:\n"
-                f"Mean: {stats['mean']:.2f}, Median: {stats['median']:.2f}\n"
-                f"Range: {stats['min']} to {stats['max']}\n"
-                f"Sum: {stats['sum']}",
-            )
-        ]
-
-        return (content, stats)
-
     else:
         raise ValueError(f"Unknown tool: {name}")
 

src/mcp/server/lowlevel/server.py

@@ -73,7 +73,7 @@ async def main():
 import warnings
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
-from typing import Any, Generic, cast
+from typing import Any, Generic, TypeAlias, cast
 
 import anyio
 import jsonschema
@@ -96,6 +96,11 @@ async def main():
 LifespanResultT = TypeVar("LifespanResultT")
 RequestT = TypeVar("RequestT", default=Any)
 
+# type aliases for tool call results
+StructuredContent: TypeAlias = dict[str, Any]
+UnstructuredContent: TypeAlias = Iterable[types.ContentBlock]
+CombinationContent: TypeAlias = tuple[UnstructuredContent, StructuredContent]
+
 # This will be properly typed in each Server instance's context
 request_ctx: contextvars.ContextVar[RequestContext[ServerSession, Any, Any]] = contextvars.ContextVar("request_ctx")
 
@@ -415,20 +420,19 @@ async def _get_cached_tool_definition(self, tool_name: str) -> types.Tool | None
     def call_tool(self):
         """Register a tool call handler.
 
-        The handler validates input against inputSchema, calls the tool function, and processes results:
-        - Content only: returns as-is
-        - Dict only: serializes to JSON text and returns as content with structuredContent
-        - Both: returns content and structuredContent
+        The handler validates input against inputSchema, calls the tool function,
+        and builds a CallToolResult with the results:
+        - Unstructured content (iterable of ContentBlock): returned in content
+        - Structured content (dict): returned in structuredContent, serialized JSON text returned in content
+        - Both: returned in content and structuredContent
 
         If outputSchema is defined, validates structuredContent or errors if missing.
         """
 
         def decorator(
             func: Callable[
                 ...,
-                Awaitable[
-                    Iterable[types.ContentBlock] | dict[str, Any] | tuple[Iterable[types.ContentBlock], dict[str, Any]]
-                ],
+                Awaitable[UnstructuredContent | StructuredContent | CombinationContent],
             ],
         ):
             logger.debug("Registering handler for CallToolRequest")
@@ -450,37 +454,41 @@ async def handler(req: types.CallToolRequest):
                     results = await func(tool_name, arguments)
 
                     # output normalization
-                    content: list[types.ContentBlock]
-                    structured_content: dict[str, Any] | None
-
+                    unstructured_content: UnstructuredContent
+                    maybe_structured_content: StructuredContent | None
                     if isinstance(results, tuple) and len(results) == 2:
-                        # tool returned both content and structured content
-                        structured_content = cast(dict[str, Any], results[1])
-                        content = list(cast(Iterable[types.ContentBlock], results[0]))
+                        # tool returned both structured and unstructured content
+                        unstructured_content, maybe_structured_content = cast(CombinationContent, results)
                     elif isinstance(results, dict):
                         # tool returned structured content only
-                        structured_content = cast(dict[str, Any], results)
-                        content = [types.TextContent(type="text", text=json.dumps(results, indent=2))]
+                        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
+                        unstructured_content = cast(UnstructuredContent, results)
+                        maybe_structured_content = None
                     else:
-                        # tool returned content only
-                        structured_content = None
-                        content = list(cast(Iterable[types.ContentBlock], results))
+                        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 structured_content 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=structured_content, 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(
-                        types.CallToolResult(content=content, structuredContent=structured_content, isError=False)
+                        types.CallToolResult(
+                            content=list(unstructured_content),
+                            structuredContent=maybe_structured_content,
+                            isError=False,
+                        )
                     )
                 except Exception as e:
                     return self._make_error_result(str(e))