From 1e90f40a0bb20e07f6f9ae824b5c90349f970a96 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Tue, 18 Aug 2026 12:50:54 +0000 Subject: [PATCH] Hand TypedDict tool results to pydantic natively MCPServer used to mirror a TypedDict return type into a synthesized BaseModel by hand. That mirror gave optional keys a `None` default and dumped them as `null`, so a tool omitting a `NotRequired`/`total=False` key produced structuredContent that violated its own outputSchema and was rejected by the client (#3224); it fed un-stripped `NotRequired`/`Required` (3.10) and `ReadOnly` (3.10-3.12) qualifiers to `create_model`, which raised at registration (#3227); and it dropped the TypedDict's docstring and `Annotated[..., Field(...)]` metadata from the schema. TypedDict returns are now validated and serialized through a `TypeAdapter` over the TypedDict itself, so pydantic's own handling of qualifiers, totality, docstrings and field metadata applies and omitted keys stay absent. Below Python 3.12 pydantic refuses `typing.TypedDict`, so those are rebuilt as an equivalent `typing_extensions.TypedDict` first. The validator is built once at registration, inside the existing "not serializable" fallback, and cached on `FuncMetadata` as `output_adapter`; `output_model` is now the TypedDict class for such tools. Observable schema change for TypedDict tools: optional keys no longer carry `"default": null`, and docstring/Field descriptions and constraints now appear. Fixes #3224 Fixes #3227 --- docs/servers/structured-output.md | 2 +- .../mcpserver/utilities/func_metadata.py | 145 +++++++++--------- tests/server/mcpserver/test_func_metadata.py | 50 +++++- tests/server/mcpserver/test_server.py | 30 +++- 4 files changed, 149 insertions(+), 78 deletions(-) diff --git a/docs/servers/structured-output.md b/docs/servers/structured-output.md index 510e750faa..d67a68d47c 100644 --- a/docs/servers/structured-output.md +++ b/docs/servers/structured-output.md @@ -100,7 +100,7 @@ Not every shape deserves a class. A `TypedDict` produces the same schema: --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 diff --git a/src/mcp/server/mcpserver/utilities/func_metadata.py b/src/mcp/server/mcpserver/utilities/func_metadata.py index 2037b860a1..c06cf747b1 100644 --- a/src/mcp/server/mcpserver/utilities/func_metadata.py +++ b/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 @@ def model_dump_one_level(self) -> dict[str, Any]: 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 @@ def convert_result(self, result: Any) -> CallToolResult | InputRequiredResult: 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 @@ def convert_result(self, result: Any) -> CallToolResult | InputRequiredResult: 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 @@ def func_metadata( - 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 @@ def func_metadata( # 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 @@ def _try_create_model_and_schema( 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 @@ def _try_create_model_and_schema( 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 @@ def _try_create_model_and_schema( 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 @@ def _create_model_from_class(cls: type[Any], type_hints: dict[str, Any]) -> type 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]: diff --git a/tests/server/mcpserver/test_func_metadata.py b/tests/server/mcpserver/test_func_metadata.py index 2dfe5d389d..c0ca3734d2 100644 --- a/tests/server/mcpserver/test_func_metadata.py +++ b/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""" diff --git a/tests/server/mcpserver/test_server.py b/tests/server/mcpserver/test_server.py index 81b490c544..9700f37adb 100644 --- a/tests/server/mcpserver/test_server.py +++ b/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: