Hand TypedDict tool results to pydantic natively

docs/servers/structured-output.md

@@ -100,7 +100,7 @@
 --8<-- "docs_src/structured_output/tutorial003.py"
 ```
 
-A `TypedDict` is a plain `dict` at runtime, so that is what you build and return. The schema, the validation, and `structured_content` are identical to the `BaseModel` version (minus the descriptions, which `TypedDict` has no place for).
+A `TypedDict` is a plain `dict` at runtime, so that is what you build and return. The schema, the validation, and `structured_content` are identical to the `BaseModel` version: the class docstring and `Annotated[..., Field(description=...)]` carry the descriptions, and a `NotRequired` key you leave out of the dict stays out of `structured_content`.
 
 ## A dataclass
 

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

@@ -1,19 +1,20 @@
 import functools
 import inspect
 import json
+import sys
 from collections.abc import Awaitable, Callable, Sequence
 from itertools import chain
 from types import GenericAlias
-from typing import Annotated, Any, Union, cast, get_args, get_origin, get_type_hints
+from typing import Annotated, Any, Union, cast, get_args, get_origin
 
 import anyio
 import anyio.to_thread
 import pydantic_core
 from mcp_types import CallToolResult, ContentBlock, InputRequiredResult, TextContent
-from pydantic import BaseModel, ConfigDict, Field, PydanticUserError, WithJsonSchema, create_model
+from pydantic import BaseModel, ConfigDict, Field, PydanticUserError, TypeAdapter, WithJsonSchema, create_model
 from pydantic.fields import FieldInfo
 from pydantic.json_schema import GenerateJsonSchema, JsonSchemaWarningKind
-from typing_extensions import is_typeddict
+from typing_extensions import NotRequired, TypedDict, get_type_hints, is_typeddict
 from typing_inspection.introspection import (
     UNKNOWN,
     AnnotationSource,
@@ -85,9 +86,15 @@
 class FuncMetadata(BaseModel):
     arg_model: Annotated[type[ArgModelBase], WithJsonSchema(None)]
     output_schema: dict[str, Any] | None = None
-    output_model: Annotated[type[BaseModel], WithJsonSchema(None)] | None = None
+    output_model: Annotated[type[Any], WithJsonSchema(None)] | None = None
     wrap_output: bool = False
 
+    @functools.cached_property
+    def output_adapter(self) -> TypeAdapter[Any]:
+        """Validates and serializes structured output against `output_model`."""
+        assert self.output_model is not None, "Output model must be set if output schema is defined"
+        return TypeAdapter(self.output_model)
+
     def validate_arguments(self, arguments_to_validate: dict[str, Any]) -> dict[str, Any]:
         """Validate raw arguments into a one-level kwargs dict (no function call).
 
@@ -144,8 +151,7 @@
             return result
         if isinstance(result, CallToolResult):
             if self.output_schema is not None:
-                assert self.output_model is not None, "Output model must be set if output schema is defined"
-                self.output_model.model_validate(result.structured_content)
+                self.output_adapter.validate_python(result.structured_content)
             return result
 
         unstructured_content = _convert_to_content(result)
@@ -156,9 +162,12 @@
         if self.wrap_output:
             result = {"result": result}
 
-        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)
+        validated = self.output_adapter.validate_python(result)
+        if isinstance(validated, BaseModel):
+            # Dump via the instance so a returned subclass keeps its own fields.
+            structured_content = validated.model_dump(mode="json", by_alias=True)
+        else:
+            structured_content = self.output_adapter.dump_python(validated, mode="json", by_alias=True)
 
         return CallToolResult(content=unstructured_content, structured_content=structured_content)
 
@@ -238,7 +247,7 @@
             - BaseModel subclasses (used directly)
             - Primitive types (str, int, float, bool, bytes, None) - wrapped in a
                 model with a 'result' field
-            - TypedDict - converted to a Pydantic model with same fields
+            - TypedDict - used directly
             - Dataclasses and other annotated classes - converted to Pydantic models
             - Generic types (list, dict, Union, etc.) - wrapped in a model with a 'result' field
             - Content blocks (TextContent, EmbeddedResource, ...), Image and Audio, bare or inside a
@@ -374,30 +383,41 @@
         # structured_output=True still forces one.
         return FuncMetadata(arg_model=arguments_model)
 
-    output_model, output_schema, wrap_output = _try_create_model_and_schema(
-        original_annotation, return_type_expr, func.__name__
-    )
+    output_model, wrap_output = _create_output_model(original_annotation, return_type_expr, func.__name__)
+
+    if output_model is not None:
+        meta = FuncMetadata(arg_model=arguments_model, output_model=output_model, wrap_output=wrap_output)
+        try:
+            # Building the validator here surfaces unsupported types at registration rather than on the
+            # first call. StrictJsonSchema raises instead of emitting warnings.
+            meta.output_schema = meta.output_adapter.json_schema(schema_generator=StrictJsonSchema)
+            return meta
+        except (
+            PydanticUserError,
+            TypeError,
+            ValueError,
+            pydantic_core.SchemaError,
+            pydantic_core.ValidationError,
+        ) as e:
+            # These are expected errors when a type can't be converted to a Pydantic schema
+            # PydanticUserError: When Pydantic can't handle the type (e.g. PydanticInvalidForJsonSchema);
+            #   subclasses TypeError on pydantic <2.13 and RuntimeError on pydantic >=2.13
+            # ValueError: When there are issues with the type definition (including our custom warnings)
+            # SchemaError: When Pydantic can't build a schema
+            # ValidationError: When validation fails
+            logger.info(f"Cannot create schema for type {return_type_expr} in {func.__name__}: {type(e).__name__}: {e}")
 
-    if output_model is None and structured_output is True:
+    if structured_output is True:
         # Model creation failed or produced warnings - no structured output
         raise InvalidSignature(
             f"Function {func.__name__}: return type {return_type_expr} is not serializable for structured output"
         )
 
-    return FuncMetadata(
-        arg_model=arguments_model,
-        output_schema=output_schema,
-        output_model=output_model,
-        wrap_output=wrap_output,
-    )
+    return FuncMetadata(arg_model=arguments_model)
 
 
-def _try_create_model_and_schema(
-    original_annotation: Any,
-    type_expr: Any,
-    func_name: str,
-) -> tuple[type[BaseModel] | None, dict[str, Any] | None, bool]:
-    """Try to create a model and schema for the given annotation without warnings.
+def _create_output_model(original_annotation: Any, type_expr: Any, func_name: str) -> tuple[type[Any] | None, bool]:
+    """Pick the type structured output is validated against for the given return annotation.
 
     Args:
         original_annotation: The original return annotation (may be wrapped in `Annotated`).
@@ -406,11 +426,11 @@
         func_name: The name of the function.
 
     Returns:
-        tuple of (model or None, schema or None, wrap_output)
-        Model and schema are None if warnings occur or creation fails.
+        tuple of (model or None, wrap_output)
+        Model is None if the type cannot carry structured output.
         wrap_output is True if the result needs to be wrapped in {"result": ...}
     """
-    model = None
+    model: type[Any] | None = None
     wrap_output = False
 
     # First handle special case: None
@@ -446,9 +466,9 @@
         if issubclass(type_annotation, BaseModel):
             model = type_annotation
 
-        # Case 2: TypedDicts:
+        # Case 2: TypedDicts (pydantic reads qualifiers, totality, docstring and `Annotated` metadata natively)
         elif is_typeddict(type_annotation):
-            model = _create_model_from_typeddict(type_annotation)
+            model = _pydantic_readable_typeddict(type_annotation)
 
         # Case 3: Primitive types that need wrapping
         elif type_annotation in (str, int, float, bool, bytes, type(None)):
@@ -470,30 +490,7 @@
         model = _create_wrapped_model(func_name, original_annotation)
         wrap_output = True
 
-    if model:
-        # If we successfully created a model, try to get its schema
-        # Use StrictJsonSchema to raise exceptions instead of warnings
-        try:
-            schema = model.model_json_schema(schema_generator=StrictJsonSchema)
-        except (
-            PydanticUserError,
-            TypeError,
-            ValueError,
-            pydantic_core.SchemaError,
-            pydantic_core.ValidationError,
-        ) as e:
-            # These are expected errors when a type can't be converted to a Pydantic schema
-            # PydanticUserError: When Pydantic can't handle the type (e.g. PydanticInvalidForJsonSchema);
-            #   subclasses TypeError on pydantic <2.13 and RuntimeError on pydantic >=2.13
-            # ValueError: When there are issues with the type definition (including our custom warnings)
-            # SchemaError: When Pydantic can't build a schema
-            # ValidationError: When validation fails
-            logger.info(f"Cannot create schema for type {type_expr} in {func_name}: {type(e).__name__}: {e}")
-            return None, None, False
-
-        return model, schema, wrap_output
-
-    return None, None, False
+    return model, wrap_output
 
 
 _no_default = object()
@@ -523,25 +520,29 @@
     return create_model(cls.__name__, __config__=ConfigDict(from_attributes=True), **model_fields)
 
 
-def _create_model_from_typeddict(td_type: type[Any]) -> type[BaseModel]:
-    """Create a Pydantic model from a TypedDict.
+def _pydantic_readable_typeddict(td_type: type[Any]) -> type[Any]:
+    """pydantic refuses `typing.TypedDict` below Python 3.12 (it needs `__orig_bases__`); rebuild those as an
+    equivalent `typing_extensions.TypedDict` so tool authors don't have to know. Delete once 3.11 support goes."""
+    if sys.version_info >= (3, 12) or type(td_type).__module__ != "typing":
+        return td_type
+    return _as_typing_extensions_typeddict(td_type)  # pragma: lax no cover
 
-    The created model will have the same name and fields as the TypedDict.
-    """
-    type_hints = get_type_hints(td_type)
-    required_keys = getattr(td_type, "__required_keys__", set(type_hints.keys()))
 
-    model_fields: dict[str, Any] = {}
-    for field_name, field_type in type_hints.items():
-        if field_name not in required_keys:
-            # For optional TypedDict fields, set default=None
-            # This makes them not required in the Pydantic model
-            # The model should use exclude_unset=True when dumping to get TypedDict semantics
-            model_fields[field_name] = (field_type, None)
-        else:
-            model_fields[field_name] = field_type
-
-    return create_model(td_type.__name__, **model_fields)
+def _as_typing_extensions_typeddict(td_type: type[Any]) -> type[Any]:  # pragma: lax no cover
+    items: dict[str, Any] = {}
+    for name, hint in get_type_hints(td_type, include_extras=True).items():
+        key = inspect_annotation(hint, annotation_source=AnnotationSource.TYPED_DICT)
+        item: Any = Annotated[(key.type, *key.metadata)] if key.metadata else key.type
+        # pydantic's rule: an explicit qualifier wins over class totality. Needed because a stdlib TypedDict
+        # this old computes `__required_keys__` without seeing `typing_extensions` qualifiers.
+        required = (name in td_type.__required_keys__ or "required" in key.qualifiers) and (
+            "not_required" not in key.qualifiers
+        )
+        items[name] = item if required else NotRequired[item]
+    # The functional form, spelled so type checkers don't try to evaluate it statically.
+    rebuilt = cast("Callable[[str, dict[str, Any]], type[Any]]", TypedDict)(td_type.__name__, items)
+    rebuilt.__doc__ = td_type.__doc__
+    return rebuilt
 
 
 def _create_wrapped_model(func_name: str, annotation: Any) -> type[BaseModel]:

tests/server/mcpserver/test_func_metadata.py

@@ -9,9 +9,11 @@
 
 import annotated_types
 import pytest
+import typing_extensions
 from dirty_equals import IsPartialDict
 from mcp_types import CallToolResult, ContentBlock, EmbeddedResource, InputRequiredResult, TextContent
 from pydantic import BaseModel, Field
+from typing_extensions import NotRequired, ReadOnly, Required
 
 from mcp.server.mcpserver import Audio, Image
 from mcp.server.mcpserver.exceptions import InvalidSignature
@@ -773,22 +775,28 @@ def func_returning_dataclass() -> PersonDataClass:  # pragma: no cover
 def test_structured_output_typeddict():
     """Test structured output with TypedDict return types"""
 
+    # stdlib TypedDict with a qualifier: exercises the typing_extensions rebuild below Python 3.12
     class PersonTypedDictOptional(TypedDict, total=False):
-        name: str
+        name: Required[str]
         age: int
 
-    def func_returning_typeddict_optional() -> PersonTypedDictOptional:  # pragma: no cover
+    def func_returning_typeddict_optional() -> PersonTypedDictOptional:
         return {"name": "Dave"}  # Only returning one field to test partial dict
 
     meta = func_metadata(func_returning_typeddict_optional)
     assert meta.output_schema == {
         "type": "object",
         "properties": {
-            "name": {"title": "Name", "type": "string", "default": None},
-            "age": {"title": "Age", "type": "integer", "default": None},
+            "name": {"title": "Name", "type": "string"},
+            "age": {"title": "Age", "type": "integer"},
         },
+        "required": ["name"],
         "title": "PersonTypedDictOptional",
     }
+    # An optional key the tool leaves out is absent, not null, so it validates against the schema above
+    result = meta.convert_result(func_returning_typeddict_optional())
+    assert isinstance(result, CallToolResult)
+    assert result.structured_content == {"name": "Dave"}
 
     # Test with total=True (all required)
     class PersonTypedDictRequired(TypedDict):
@@ -812,6 +820,40 @@ def func_returning_typeddict_required() -> PersonTypedDictRequired:  # pragma: n
     }
 
 
+def test_structured_output_typeddict_qualifiers_and_metadata():
+    """PEP 655/705 qualifiers register on every supported Python and decide `required`; the docstring
+    and `Annotated` field metadata reach the schema like they do for a BaseModel."""
+
+    class Forecast(typing_extensions.TypedDict, total=False):
+        """Tomorrow's weather."""
+
+        city: Required[Annotated[str, Field(description="City name")]]
+        high: ReadOnly[Required[float]]
+        low: float
+        summary: NotRequired[Annotated[str, Field(max_length=80)]]
+
+    def forecast() -> Forecast:
+        return {"city": "Berlin", "high": 21.5}
+
+    with pytest.warns(UserWarning, match="ReadOnly"):  # pydantic notes it won't enforce ReadOnly
+        meta = func_metadata(forecast)
+    result = meta.convert_result(forecast())
+    assert isinstance(result, CallToolResult)
+    assert result.structured_content == {"city": "Berlin", "high": 21.5}
+    assert meta.output_schema == {
+        "type": "object",
+        "title": "Forecast",
+        "description": "Tomorrow's weather.",
+        "properties": {
+            "city": {"title": "City", "type": "string", "description": "City name"},
+            "high": {"title": "High", "type": "number"},
+            "low": {"title": "Low", "type": "number"},
+            "summary": {"title": "Summary", "type": "string", "maxLength": 80},
+        },
+        "required": ["city", "high"],
+    }
+
+
 def test_structured_output_ordinary_class():
     """Test structured output with ordinary annotated classes"""
 

tests/server/mcpserver/test_server.py

@@ -1,7 +1,7 @@
 import base64
 from pathlib import Path
 from types import SimpleNamespace
-from typing import Any
+from typing import Any, TypedDict
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import anyio
@@ -44,6 +44,7 @@
 from pydantic import BaseModel
 from starlette.applications import Starlette
 from starlette.routing import Mount, Route
+from typing_extensions import NotRequired
 
 from mcp.client import Client
 from mcp.server.context import ServerRequestContext
@@ -715,6 +716,33 @@ async def test_remove_tool_and_call(self):
             assert "Unknown tool" in content.text
 
 
+@pytest.mark.anyio
+async def test_typeddict_tool_omitting_optional_keys_passes_client_validation():
+    """The client validates structured content against the tool's output schema, so a `NotRequired`
+    key the tool leaves out must be absent from `structured_content` rather than null."""
+
+    class Person(TypedDict):
+        name: str
+        age: NotRequired[int]
+
+    mcp = MCPServer()
+
+    @mcp.tool()
+    def get_person() -> Person:
+        return {"name": "Dave"}
+
+    async with Client(mcp) as client:
+        (tool,) = (await client.list_tools()).tools
+        assert tool.output_schema == {
+            "type": "object",
+            "title": "Person",
+            "properties": {"name": {"title": "Name", "type": "string"}, "age": {"title": "Age", "type": "integer"}},
+            "required": ["name"],
+        }
+        result = await client.call_tool("get_person", {})
+        assert result.structured_content == {"name": "Dave"}
+
+
 class TestServerResources:
     async def test_init_with_resources(self):
         def get_text() -> str: