diff --git a/README.md b/README.md index 6f2bbc8b5..412143a9f 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,7 @@ - [Server](#server) - [Resources](#resources) - [Tools](#tools) + - [Structured Output](#structured-output) - [Prompts](#prompts) - [Images](#images) - [Context](#context) @@ -249,6 +250,127 @@ async def fetch_weather(city: str) -> str: return response.text ``` +#### Structured Output + +Tools will return structured results by default, if their return type +annotation is compatible. Otherwise, they will return unstructured results. + +Structured output supports these return types: +- Pydantic models (BaseModel subclasses) +- TypedDicts +- Dataclasses and other classes with type hints +- `dict[str, T]` (where T is any JSON-serializable type) +- Primitive types (str, int, float, bool, bytes, None) - wrapped in `{"result": value}` +- Generic types (list, tuple, Union, Optional, etc.) - wrapped in `{"result": value}` + +Classes without type hints cannot be serialized for structured output. Only +classes with properly annotated attributes will be converted to Pydantic models +for schema generation and validation. + +Structured results are automatically validated against the output schema +generated from the annotation. This ensures the tool returns well-typed, +validated data that clients can easily process. + +**Note:** For backward compatibility, unstructured results are also +returned. Unstructured results are provided for backward compatibility +with previous versions of the MCP specification, and are quirks-compatible +with previous versions of FastMCP in the current version of the SDK. + +**Note:** In cases where a tool function's return type annotation +causes the tool to be classified as structured _and this is undesirable_, +the classification can be suppressed by passing `structured_output=False` +to the `@tool` decorator. + +```python +from mcp.server.fastmcp import FastMCP +from pydantic import BaseModel, Field +from typing import TypedDict + +mcp = FastMCP("Weather Service") + + +# Using Pydantic models for rich structured data +class WeatherData(BaseModel): + temperature: float = Field(description="Temperature in Celsius") + humidity: float = Field(description="Humidity percentage") + condition: str + wind_speed: float + + +@mcp.tool() +def get_weather(city: str) -> WeatherData: + """Get structured weather data""" + return WeatherData( + temperature=22.5, humidity=65.0, condition="partly cloudy", wind_speed=12.3 + ) + + +# Using TypedDict for simpler structures +class LocationInfo(TypedDict): + latitude: float + longitude: float + name: str + + +@mcp.tool() +def get_location(address: str) -> LocationInfo: + """Get location coordinates""" + return LocationInfo(latitude=51.5074, longitude=-0.1278, name="London, UK") + + +# Using dict[str, Any] for flexible schemas +@mcp.tool() +def get_statistics(data_type: str) -> dict[str, float]: + """Get various statistics""" + return {"mean": 42.5, "median": 40.0, "std_dev": 5.2} + + +# Ordinary classes with type hints work for structured output +class UserProfile: + name: str + age: int + email: str | None = None + + def __init__(self, name: str, age: int, email: str | None = None): + self.name = name + self.age = age + self.email = email + + +@mcp.tool() +def get_user(user_id: str) -> UserProfile: + """Get user profile - returns structured data""" + return UserProfile(name="Alice", age=30, email="alice@example.com") + + +# Classes WITHOUT type hints cannot be used for structured output +class UntypedConfig: + def __init__(self, setting1, setting2): + self.setting1 = setting1 + self.setting2 = setting2 + + +@mcp.tool() +def get_config() -> UntypedConfig: + """This returns unstructured output - no schema generated""" + return UntypedConfig("value1", "value2") + + +# Lists and other types are wrapped automatically +@mcp.tool() +def list_cities() -> list[str]: + """Get a list of cities""" + return ["London", "Paris", "Tokyo"] + # Returns: {"result": ["London", "Paris", "Tokyo"]} + + +@mcp.tool() +def get_temperature(city: str) -> float: + """Get temperature as a simple float""" + return 22.5 + # Returns: {"result": 22.5} +``` + ### Prompts Prompts are reusable templates that help LLMs interact with your server effectively: diff --git a/examples/fastmcp/weather_structured.py b/examples/fastmcp/weather_structured.py new file mode 100644 index 000000000..8c26fc39e --- /dev/null +++ b/examples/fastmcp/weather_structured.py @@ -0,0 +1,225 @@ +""" +FastMCP Weather Example with Structured Output + +Demonstrates how to use structured output with tools to return +well-typed, validated data that clients can easily process. +""" + +import asyncio +import json +import sys +from dataclasses import dataclass +from datetime import datetime +from typing import TypedDict + +from pydantic import BaseModel, Field + +from mcp.server.fastmcp import FastMCP +from mcp.shared.memory import create_connected_server_and_client_session as client_session + +# Create server +mcp = FastMCP("Weather Service") + + +# Example 1: Using a Pydantic model for structured output +class WeatherData(BaseModel): + """Structured weather data response""" + + temperature: float = Field(description="Temperature in Celsius") + humidity: float = Field(description="Humidity percentage (0-100)") + condition: str = Field(description="Weather condition (sunny, cloudy, rainy, etc.)") + wind_speed: float = Field(description="Wind speed in km/h") + location: str = Field(description="Location name") + timestamp: datetime = Field(default_factory=datetime.now, description="Observation time") + + +@mcp.tool() +def get_weather(city: str) -> WeatherData: + """Get current weather for a city with full structured data""" + # In a real implementation, this would fetch from a weather API + return WeatherData(temperature=22.5, humidity=65.0, condition="partly cloudy", wind_speed=12.3, location=city) + + +# Example 2: Using TypedDict for a simpler structure +class WeatherSummary(TypedDict): + """Simple weather summary""" + + city: str + temp_c: float + description: str + + +@mcp.tool() +def get_weather_summary(city: str) -> WeatherSummary: + """Get a brief weather summary for a city""" + return WeatherSummary(city=city, temp_c=22.5, description="Partly cloudy with light breeze") + + +# Example 3: Using dict[str, Any] for flexible schemas +@mcp.tool() +def get_weather_metrics(cities: list[str]) -> dict[str, dict[str, float]]: + """Get weather metrics for multiple cities + + Returns a dictionary mapping city names to their metrics + """ + # Returns nested dictionaries with weather metrics + return { + city: {"temperature": 20.0 + i * 2, "humidity": 60.0 + i * 5, "pressure": 1013.0 + i * 0.5} + for i, city in enumerate(cities) + } + + +# Example 4: Using dataclass for weather alerts +@dataclass +class WeatherAlert: + """Weather alert information""" + + severity: str # "low", "medium", "high" + title: str + description: str + affected_areas: list[str] + valid_until: datetime + + +@mcp.tool() +def get_weather_alerts(region: str) -> list[WeatherAlert]: + """Get active weather alerts for a region""" + # In production, this would fetch real alerts + if region.lower() == "california": + return [ + WeatherAlert( + severity="high", + title="Heat Wave Warning", + description="Temperatures expected to exceed 40°C", + affected_areas=["Los Angeles", "San Diego", "Riverside"], + valid_until=datetime(2024, 7, 15, 18, 0), + ), + WeatherAlert( + severity="medium", + title="Air Quality Advisory", + description="Poor air quality due to wildfire smoke", + affected_areas=["San Francisco Bay Area"], + valid_until=datetime(2024, 7, 14, 12, 0), + ), + ] + return [] + + +# Example 5: Returning primitives with structured output +@mcp.tool() +def get_temperature(city: str, unit: str = "celsius") -> float: + """Get just the temperature for a city + + When returning primitives as structured output, + the result is wrapped in {"result": value} + """ + base_temp = 22.5 + if unit.lower() == "fahrenheit": + return base_temp * 9 / 5 + 32 + return base_temp + + +# Example 6: Weather statistics with nested models +class DailyStats(BaseModel): + """Statistics for a single day""" + + high: float + low: float + mean: float + + +class WeatherStats(BaseModel): + """Weather statistics over a period""" + + location: str + period_days: int + temperature: DailyStats + humidity: DailyStats + precipitation_mm: float = Field(description="Total precipitation in millimeters") + + +@mcp.tool() +def get_weather_stats(city: str, days: int = 7) -> WeatherStats: + """Get weather statistics for the past N days""" + return WeatherStats( + location=city, + period_days=days, + temperature=DailyStats(high=28.5, low=15.2, mean=21.8), + humidity=DailyStats(high=85.0, low=45.0, mean=65.0), + precipitation_mm=12.4, + ) + + +if __name__ == "__main__": + + async def test() -> None: + """Test the tools by calling them through the server as a client would""" + print("Testing Weather Service Tools (via MCP protocol)\n") + print("=" * 80) + + async with client_session(mcp._mcp_server) as client: + # Test get_weather + result = await client.call_tool("get_weather", {"city": "London"}) + print("\nWeather in London:") + print(json.dumps(result.structuredContent, indent=2)) + + # Test get_weather_summary + result = await client.call_tool("get_weather_summary", {"city": "Paris"}) + print("\nWeather summary for Paris:") + print(json.dumps(result.structuredContent, indent=2)) + + # Test get_weather_metrics + result = await client.call_tool("get_weather_metrics", {"cities": ["Tokyo", "Sydney", "Mumbai"]}) + print("\nWeather metrics:") + print(json.dumps(result.structuredContent, indent=2)) + + # Test get_weather_alerts + result = await client.call_tool("get_weather_alerts", {"region": "California"}) + print("\nWeather alerts for California:") + print(json.dumps(result.structuredContent, indent=2)) + + # Test get_temperature + result = await client.call_tool("get_temperature", {"city": "Berlin", "unit": "fahrenheit"}) + print("\nTemperature in Berlin:") + print(json.dumps(result.structuredContent, indent=2)) + + # Test get_weather_stats + result = await client.call_tool("get_weather_stats", {"city": "Seattle", "days": 30}) + print("\nWeather stats for Seattle (30 days):") + print(json.dumps(result.structuredContent, indent=2)) + + # Also show the text content for comparison + print("\nText content for last result:") + for content in result.content: + if content.type == "text": + print(content.text) + + async def print_schemas() -> None: + """Print all tool schemas""" + print("Tool Schemas for Weather Service\n") + print("=" * 80) + + tools = await mcp.list_tools() + for tool in tools: + print(f"\nTool: {tool.name}") + print(f"Description: {tool.description}") + print("Input Schema:") + print(json.dumps(tool.inputSchema, indent=2)) + + if tool.outputSchema: + print("Output Schema:") + print(json.dumps(tool.outputSchema, indent=2)) + else: + print("Output Schema: None (returns unstructured content)") + + print("-" * 80) + + # Check command line arguments + if len(sys.argv) > 1 and sys.argv[1] == "--schemas": + asyncio.run(print_schemas()) + else: + print("Usage:") + print(" python weather_structured.py # Run tool tests") + print(" python weather_structured.py --schemas # Print tool schemas") + print() + asyncio.run(test()) diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py index 948817140..84f873583 100644 --- a/src/mcp/client/session.py +++ b/src/mcp/client/session.py @@ -1,8 +1,10 @@ +import logging from datetime import timedelta from typing import Any, Protocol import anyio.lowlevel from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream +from jsonschema import SchemaError, ValidationError, validate from pydantic import AnyUrl, TypeAdapter import mcp.types as types @@ -13,6 +15,9 @@ DEFAULT_CLIENT_INFO = types.Implementation(name="mcp", version="0.1.0") +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger("client") + class SamplingFnT(Protocol): async def __call__( @@ -128,6 +133,7 @@ def __init__( self._list_roots_callback = list_roots_callback or _default_list_roots_callback self._logging_callback = logging_callback or _default_logging_callback self._message_handler = message_handler or _default_message_handler + self._tool_output_schemas: dict[str, dict[str, Any] | None] = {} async def initialize(self) -> types.InitializeResult: sampling = types.SamplingCapability() if self._sampling_callback is not _default_sampling_callback else None @@ -285,7 +291,7 @@ async def call_tool( ) -> types.CallToolResult: """Send a tools/call request with optional progress callback support.""" - return await self.send_request( + result = await self.send_request( types.ClientRequest( types.CallToolRequest( method="tools/call", @@ -300,6 +306,33 @@ async def call_tool( progress_callback=progress_callback, ) + if not result.isError: + await self._validate_tool_result(name, result) + + return result + + async def _validate_tool_result(self, name: str, result: types.CallToolResult) -> None: + """Validate the structured content of a tool result against its output schema.""" + if name not in self._tool_output_schemas: + # refresh output schema cache + await self.list_tools() + + output_schema = None + if name in self._tool_output_schemas: + output_schema = self._tool_output_schemas.get(name) + else: + logger.warning(f"Tool {name} not listed by server, cannot validate any structured content") + + if output_schema is not None: + if result.structuredContent is None: + raise RuntimeError(f"Tool {name} has an output schema but did not return structured content") + try: + validate(result.structuredContent, output_schema) + except ValidationError as e: + raise RuntimeError(f"Invalid structured content returned by tool {name}: {e}") + except SchemaError as e: + raise RuntimeError(f"Invalid schema for tool {name}: {e}") + async def list_prompts(self, cursor: str | None = None) -> types.ListPromptsResult: """Send a prompts/list request.""" return await self.send_request( @@ -351,7 +384,7 @@ async def complete( async def list_tools(self, cursor: str | None = None) -> types.ListToolsResult: """Send a tools/list request.""" - return await self.send_request( + result = await self.send_request( types.ClientRequest( types.ListToolsRequest( method="tools/list", @@ -361,6 +394,13 @@ async def list_tools(self, cursor: str | None = None) -> types.ListToolsResult: types.ListToolsResult, ) + # Cache tool output schemas for future validation + # Note: don't clear the cache, as we may be using a cursor + for tool in result.tools: + self._tool_output_schemas[tool.name] = tool.outputSchema + + return result + async def send_roots_list_changed(self) -> None: """Send a roots/list_changed notification.""" await self.send_notification( diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index 668c6df82..956a8aa78 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/src/mcp/server/fastmcp/server.py @@ -9,7 +9,6 @@ AbstractAsyncContextManager, asynccontextmanager, ) -from itertools import chain from typing import Any, Generic, Literal import anyio @@ -38,7 +37,6 @@ from mcp.server.fastmcp.resources import FunctionResource, Resource, ResourceManager from mcp.server.fastmcp.tools import Tool, ToolManager from mcp.server.fastmcp.utilities.logging import configure_logging, get_logger -from mcp.server.fastmcp.utilities.types import Image from mcp.server.lowlevel.helper_types import ReadResourceContents from mcp.server.lowlevel.server import LifespanResultT from mcp.server.lowlevel.server import Server as MCPServer @@ -54,7 +52,6 @@ AnyFunction, ContentBlock, GetPromptResult, - TextContent, ToolAnnotations, ) from mcp.types import Prompt as MCPPrompt @@ -235,7 +232,10 @@ def run( def _setup_handlers(self) -> None: """Set up core MCP protocol handlers.""" self._mcp_server.list_tools()(self.list_tools) - self._mcp_server.call_tool()(self.call_tool) + # Note: we disable the lowlevel server's input validation. + # FastMCP does ad hoc conversion of incoming data before validating - + # for now we preserve this for backwards compatibility. + self._mcp_server.call_tool(validate_input=False)(self.call_tool) self._mcp_server.list_resources()(self.list_resources) self._mcp_server.read_resource()(self.read_resource) self._mcp_server.list_prompts()(self.list_prompts) @@ -251,6 +251,7 @@ async def list_tools(self) -> list[MCPTool]: title=info.title, description=info.description, inputSchema=info.parameters, + outputSchema=info.output_schema, annotations=info.annotations, ) for info in tools @@ -267,12 +268,10 @@ def get_context(self) -> Context[ServerSession, object, Request]: request_context = None return Context(request_context=request_context, fastmcp=self) - async def call_tool(self, name: str, arguments: dict[str, Any]) -> Sequence[ContentBlock]: + async def call_tool(self, name: str, arguments: dict[str, Any]) -> Sequence[ContentBlock] | dict[str, Any]: """Call a tool by name with arguments.""" context = self.get_context() - result = await self._tool_manager.call_tool(name, arguments, context=context) - converted_result = _convert_to_content(result) - return converted_result + return await self._tool_manager.call_tool(name, arguments, context=context, convert_result=True) async def list_resources(self) -> list[MCPResource]: """List all available resources.""" @@ -322,6 +321,7 @@ def add_tool( title: str | None = None, description: str | None = None, annotations: ToolAnnotations | None = None, + structured_output: bool | None = None, ) -> None: """Add a tool to the server. @@ -334,8 +334,19 @@ def add_tool( title: Optional human-readable title for the tool description: Optional description of what the tool does annotations: Optional ToolAnnotations providing additional tool information + structured_output: Controls whether the tool's output is structured or unstructured + - If None, auto-detects based on the function's return type annotation + - If True, unconditionally creates a structured tool (return type annotation permitting) + - If False, unconditionally creates an unstructured tool """ - self._tool_manager.add_tool(fn, name=name, title=title, description=description, annotations=annotations) + self._tool_manager.add_tool( + fn, + name=name, + title=title, + description=description, + annotations=annotations, + structured_output=structured_output, + ) def tool( self, @@ -343,6 +354,7 @@ def tool( title: str | None = None, description: str | None = None, annotations: ToolAnnotations | None = None, + structured_output: bool | None = None, ) -> Callable[[AnyFunction], AnyFunction]: """Decorator to register a tool. @@ -355,6 +367,10 @@ def tool( title: Optional human-readable title for the tool description: Optional description of what the tool does annotations: Optional ToolAnnotations providing additional tool information + structured_output: Controls whether the tool's output is structured or unstructured + - If None, auto-detects based on the function's return type annotation + - If True, unconditionally creates a structured tool (return type annotation permitting) + - If False, unconditionally creates an unstructured tool Example: @server.tool() @@ -378,7 +394,14 @@ async def async_tool(x: int, context: Context) -> str: ) def decorator(fn: AnyFunction) -> AnyFunction: - self.add_tool(fn, name=name, title=title, description=description, annotations=annotations) + self.add_tool( + fn, + name=name, + title=title, + description=description, + annotations=annotations, + structured_output=structured_output, + ) return fn return decorator @@ -942,28 +965,6 @@ async def get_prompt(self, name: str, arguments: dict[str, Any] | None = None) - raise ValueError(str(e)) -def _convert_to_content( - result: Any, -) -> Sequence[ContentBlock]: - """Convert a result to a sequence of content objects.""" - if result is None: - return [] - - if isinstance(result, ContentBlock): - return [result] - - if isinstance(result, Image): - return [result.to_image_content()] - - if isinstance(result, list | tuple): - return list(chain.from_iterable(_convert_to_content(item) for item in result)) # type: ignore[reportUnknownVariableType] - - if not isinstance(result, str): - result = pydantic_core.to_json(result, fallback=str, indent=2).decode() - - return [TextContent(type="text", text=result)] - - class Context(BaseModel, Generic[ServerSessionT, LifespanContextT, RequestT]): """Context object providing access to MCP capabilities. diff --git a/src/mcp/server/fastmcp/tools/base.py b/src/mcp/server/fastmcp/tools/base.py index 2f7c48e8b..366a76895 100644 --- a/src/mcp/server/fastmcp/tools/base.py +++ b/src/mcp/server/fastmcp/tools/base.py @@ -3,6 +3,7 @@ import functools import inspect from collections.abc import Callable +from functools import cached_property from typing import TYPE_CHECKING, Any, get_origin from pydantic import BaseModel, Field @@ -32,6 +33,10 @@ class Tool(BaseModel): context_kwarg: str | None = Field(None, description="Name of the kwarg that should receive context") annotations: ToolAnnotations | None = Field(None, description="Optional annotations for the tool") + @cached_property + def output_schema(self) -> dict[str, Any] | None: + return self.fn_metadata.output_schema + @classmethod def from_function( cls, @@ -41,6 +46,7 @@ def from_function( description: str | None = None, context_kwarg: str | None = None, annotations: ToolAnnotations | None = None, + structured_output: bool | None = None, ) -> Tool: """Create a Tool from a function.""" from mcp.server.fastmcp.server import Context @@ -65,6 +71,7 @@ def from_function( func_arg_metadata = func_metadata( fn, skip_names=[context_kwarg] if context_kwarg is not None else [], + structured_output=structured_output, ) parameters = func_arg_metadata.arg_model.model_json_schema() @@ -84,15 +91,21 @@ async def run( self, arguments: dict[str, Any], context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None, + convert_result: bool = False, ) -> Any: """Run the tool with arguments.""" try: - return await self.fn_metadata.call_fn_with_arg_validation( + result = await self.fn_metadata.call_fn_with_arg_validation( self.fn, self.is_async, arguments, {self.context_kwarg: context} if self.context_kwarg is not None else None, ) + + if convert_result: + result = self.fn_metadata.convert_result(result) + + return result except Exception as e: raise ToolError(f"Error executing tool {self.name}: {e}") from e diff --git a/src/mcp/server/fastmcp/tools/tool_manager.py b/src/mcp/server/fastmcp/tools/tool_manager.py index b9ca1655d..bfa8b2382 100644 --- a/src/mcp/server/fastmcp/tools/tool_manager.py +++ b/src/mcp/server/fastmcp/tools/tool_manager.py @@ -49,9 +49,17 @@ def add_tool( title: str | None = None, description: str | None = None, annotations: ToolAnnotations | None = None, + structured_output: bool | None = None, ) -> Tool: """Add a tool to the server.""" - tool = Tool.from_function(fn, name=name, title=title, description=description, annotations=annotations) + tool = Tool.from_function( + fn, + name=name, + title=title, + description=description, + annotations=annotations, + structured_output=structured_output, + ) existing = self._tools.get(tool.name) if existing: if self.warn_on_duplicate_tools: @@ -65,10 +73,11 @@ async def call_tool( name: str, arguments: dict[str, Any], context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None, + convert_result: bool = False, ) -> Any: """Call a tool by name with arguments.""" tool = self.get_tool(name) if not tool: raise ToolError(f"Unknown tool: {name}") - return await tool.run(arguments, context=context) + return await tool.run(arguments, context=context, convert_result=convert_result) diff --git a/src/mcp/server/fastmcp/utilities/func_metadata.py b/src/mcp/server/fastmcp/utilities/func_metadata.py index 9f8d9177a..a6f905ee5 100644 --- a/src/mcp/server/fastmcp/utilities/func_metadata.py +++ b/src/mcp/server/fastmcp/utilities/func_metadata.py @@ -1,23 +1,43 @@ import inspect import json from collections.abc import Awaitable, Callable, Sequence -from typing import ( - Annotated, - Any, - ForwardRef, +from itertools import chain +from types import GenericAlias +from typing import Annotated, Any, ForwardRef, cast, get_args, get_origin, get_type_hints + +import pydantic_core +from pydantic import ( + BaseModel, + ConfigDict, + Field, + RootModel, + WithJsonSchema, + create_model, ) - -from pydantic import BaseModel, ConfigDict, Field, WithJsonSchema, create_model from pydantic._internal._typing_extra import eval_type_backport from pydantic.fields import FieldInfo +from pydantic.json_schema import GenerateJsonSchema, JsonSchemaWarningKind from pydantic_core import PydanticUndefined from mcp.server.fastmcp.exceptions import InvalidSignature from mcp.server.fastmcp.utilities.logging import get_logger +from mcp.server.fastmcp.utilities.types import Image +from mcp.types import ContentBlock, TextContent logger = get_logger(__name__) +class StrictJsonSchema(GenerateJsonSchema): + """A JSON schema generator that raises exceptions instead of emitting warnings. + + This is used to detect non-serializable types during schema generation. + """ + + def emit_warning(self, kind: JsonSchemaWarningKind, detail: str) -> None: + # Raise an exception instead of emitting a warning + raise ValueError(f"JSON schema warning: {kind} - {detail}") + + class ArgModelBase(BaseModel): """A model representing the arguments to a function.""" @@ -38,13 +58,13 @@ def model_dump_one_level(self) -> dict[str, Any]: class FuncMetadata(BaseModel): arg_model: Annotated[type[ArgModelBase], WithJsonSchema(None)] - # We can add things in the future like - # - Maybe some args are excluded from attempting to parse from JSON - # - Maybe some args are special (like context) for dependency injection + output_schema: dict[str, Any] | None = None + output_model: Annotated[type[BaseModel], WithJsonSchema(None)] | None = None + wrap_output: bool = False async def call_fn_with_arg_validation( self, - fn: Callable[..., Any] | Awaitable[Any], + fn: Callable[..., Any | Awaitable[Any]], fn_is_async: bool, arguments_to_validate: dict[str, Any], arguments_to_pass_directly: dict[str, Any] | None, @@ -61,12 +81,39 @@ async def call_fn_with_arg_validation( arguments_parsed_dict |= arguments_to_pass_directly or {} if fn_is_async: - if isinstance(fn, Awaitable): - return await fn return await fn(**arguments_parsed_dict) - if isinstance(fn, Callable): + else: return fn(**arguments_parsed_dict) - raise TypeError("fn must be either Callable or Awaitable") + + def convert_result(self, result: Any) -> Any: + """ + Convert the result of a function call to the appropriate format for + the lowlevel server tool call handler: + + - If output_model is None, return the unstructured content directly. + - If output_model is not None, convert the result to structured output format + (dict[str, Any]) and return both unstructured and structured content. + + Note: we return unstructured content here **even though the lowlevel server + tool call handler provides generic backwards compatibility serialization of + structured content**. This is for FastMCP backwards compatibility: we need to + retain FastMCP's ad hoc conversion logic for constructing unstructured output + from function return values, whereas the lowlevel server simply serializes + the structured output. + """ + unstructured_content = _convert_to_content(result) + + if self.output_schema is None: + return unstructured_content + else: + 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") + + return (unstructured_content, structured_content) def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]: """Pre-parse data from JSON. @@ -102,13 +149,17 @@ def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]: ) -def func_metadata(func: Callable[..., Any], skip_names: Sequence[str] = ()) -> FuncMetadata: +def func_metadata( + func: Callable[..., Any], + skip_names: Sequence[str] = (), + structured_output: bool | None = None, +) -> FuncMetadata: """Given a function, return metadata including a pydantic model representing its signature. The use case for this is ``` - meta = func_to_pyd(func) + meta = func_metadata(func) validated_args = meta.arg_model.model_validate(some_raw_data_dict) return func(**validated_args.model_dump_one_level()) ``` @@ -120,8 +171,25 @@ def func_metadata(func: Callable[..., Any], skip_names: Sequence[str] = ()) -> F func: The function to convert to a pydantic model skip_names: A list of parameter names to skip. These will not be included in the model. + structured_output: Controls whether the tool's output is structured or unstructured + - If None, auto-detects based on the function's return type annotation + - If True, unconditionally creates a structured tool (return type annotation permitting) + - If False, unconditionally creates an unstructured tool + + If structured, creates a Pydantic model for the function's result based on its annotation. + Supports various return types: + - 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 + - Dataclasses and other annotated classes - converted to Pydantic models + - Generic types (list, dict, Union, etc.) - wrapped in a model with a 'result' field + Returns: - A pydantic model representing the function's signature. + A FuncMetadata object containing: + - arg_model: A pydantic model representing the function's arguments + - output_model: A pydantic model for the return type if output is structured + - output_conversion: Records how function output should be converted before returning. """ sig = _get_typed_signature(func) params = sig.parameters @@ -162,8 +230,197 @@ def func_metadata(func: Callable[..., Any], skip_names: Sequence[str] = ()) -> F **dynamic_pydantic_model_params, __base__=ArgModelBase, ) - resp = FuncMetadata(arg_model=arguments_model) - return resp + + if structured_output is False: + return FuncMetadata(arg_model=arguments_model) + + # set up structured output support based on return type annotation + + if sig.return_annotation is inspect.Parameter.empty and structured_output is True: + raise InvalidSignature(f"Function {func.__name__}: return annotation required for structured output") + + output_info = FieldInfo.from_annotation(_get_typed_annotation(sig.return_annotation, globalns)) + annotation = output_info.annotation + + output_model, output_schema, wrap_output = _try_create_model_and_schema(annotation, func.__name__, output_info) + + if output_model is None and structured_output is True: + # Model creation failed or produced warnings - no structured output + raise InvalidSignature( + f"Function {func.__name__}: return type {annotation} is not serializable for structured output" + ) + + return FuncMetadata( + arg_model=arguments_model, + output_schema=output_schema, + output_model=output_model, + wrap_output=wrap_output, + ) + + +def _try_create_model_and_schema( + annotation: Any, func_name: str, field_info: FieldInfo +) -> tuple[type[BaseModel] | None, dict[str, Any] | None, bool]: + """Try to create a model and schema for the given annotation without warnings. + + Returns: + tuple of (model or None, schema or None, wrap_output) + Model and schema are None if warnings occur or creation fails. + wrap_output is True if the result needs to be wrapped in {"result": ...} + """ + model = None + wrap_output = False + + # First handle special case: None + if annotation is None: + model = _create_wrapped_model(func_name, annotation, field_info) + wrap_output = True + + # Handle GenericAlias types (list[str], dict[str, int], Union[str, int], etc.) + elif isinstance(annotation, GenericAlias): + origin = get_origin(annotation) + + # Special case: dict with string keys can use RootModel + if origin is dict: + args = get_args(annotation) + if len(args) == 2 and args[0] is str: + model = _create_dict_model(func_name, annotation) + else: + # dict with non-str keys needs wrapping + model = _create_wrapped_model(func_name, annotation, field_info) + wrap_output = True + else: + # All other generic types need wrapping (list, tuple, Union, Optional, etc.) + model = _create_wrapped_model(func_name, annotation, field_info) + wrap_output = True + + # Handle regular type objects + elif isinstance(annotation, type): + type_annotation: type[Any] = cast(type[Any], annotation) + + # Case 1: BaseModel subclasses (can be used directly) + if issubclass(annotation, BaseModel): + model = annotation + + # Case 2: TypedDict (special dict subclass with __annotations__) + elif hasattr(type_annotation, "__annotations__") and issubclass(annotation, dict): + model = _create_model_from_typeddict(type_annotation) + + # Case 3: Primitive types that need wrapping + elif annotation in (str, int, float, bool, bytes, type(None)): + model = _create_wrapped_model(func_name, annotation, field_info) + wrap_output = True + + # Case 4: Other class types (dataclasses, regular classes with annotations) + else: + type_hints = get_type_hints(type_annotation) + if type_hints: + # Classes with type hints can be converted to Pydantic models + model = _create_model_from_class(type_annotation) + # Classes without type hints are not serializable - model remains None + + # Handle any other types not covered above + else: + # This includes typing constructs that aren't GenericAlias in Python 3.10 + # (e.g., Union, Optional in some Python versions) + model = _create_wrapped_model(func_name, annotation, field_info) + 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 (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 + # TypeError: When Pydantic can't handle the type + # 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 {annotation} in {func_name}: {type(e).__name__}: {e}") + return None, None, False + + return model, schema, wrap_output + + return None, None, False + + +def _create_model_from_class(cls: type[Any]) -> type[BaseModel]: + """Create a Pydantic model from an ordinary class. + + The created model will: + - Have the same name as the class + - Have fields with the same names and types as the class's fields + - Include all fields whose type does not include None in the set of required fields + + Precondition: cls must have type hints (i.e., get_type_hints(cls) is non-empty) + """ + type_hints = get_type_hints(cls) + + model_fields: dict[str, Any] = {} + for field_name, field_type in type_hints.items(): + if field_name.startswith("_"): + continue + + default = getattr(cls, field_name, PydanticUndefined) + field_info = FieldInfo.from_annotated_attribute(field_type, default) + model_fields[field_name] = (field_info.annotation, field_info) + + # Create a base class with the config + class BaseWithConfig(BaseModel): + model_config = ConfigDict(from_attributes=True) + + return create_model(cls.__name__, **model_fields, __base__=BaseWithConfig) + + +def _create_model_from_typeddict(td_type: type[Any]) -> type[BaseModel]: + """Create a Pydantic model from a TypedDict. + + 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(): + field_info = FieldInfo.from_annotation(field_type) + + 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 + field_info.default = None + + model_fields[field_name] = (field_info.annotation, field_info) + + return create_model(td_type.__name__, **model_fields, __base__=BaseModel) + + +def _create_wrapped_model(func_name: str, annotation: Any, field_info: FieldInfo) -> type[BaseModel]: + """Create a model that wraps a type in a 'result' field. + + This is used for primitive types, generic types like list/dict, etc. + """ + model_name = f"{func_name}Output" + + # Pydantic needs type(None) instead of None for the type annotation + if annotation is None: + annotation = type(None) + + return create_model(model_name, result=(annotation, field_info), __base__=BaseModel) + + +def _create_dict_model(func_name: str, dict_annotation: Any) -> type[BaseModel]: + """Create a RootModel for dict[str, T] types.""" + + class DictModel(RootModel[dict_annotation]): + pass + + # Give it a meaningful name + DictModel.__name__ = f"{func_name}DictOutput" + DictModel.__qualname__ = f"{func_name}DictOutput" + + return DictModel def _get_typed_annotation(annotation: Any, globalns: dict[str, Any]) -> Any: @@ -198,5 +455,40 @@ def _get_typed_signature(call: Callable[..., Any]) -> inspect.Signature: ) for param in signature.parameters.values() ] - typed_signature = inspect.Signature(typed_params) + typed_return = _get_typed_annotation(signature.return_annotation, globalns) + typed_signature = inspect.Signature(typed_params, return_annotation=typed_return) return typed_signature + + +def _convert_to_content( + result: Any, +) -> Sequence[ContentBlock]: + """ + Convert a result to a sequence of content objects. + + Note: This conversion logic comes from previous versions of FastMCP and is being + retained for purposes of backwards compatibility. It produces different unstructured + output than the lowlevel server tool call handler, which just serializes structured + content verbatim. + """ + if result is None: + return [] + + if isinstance(result, ContentBlock): + return [result] + + if isinstance(result, Image): + return [result.to_image_content()] + + if isinstance(result, list | tuple): + return list( + chain.from_iterable( + _convert_to_content(item) + for item in result # type: ignore + ) + ) + + if not isinstance(result, str): + result = pydantic_core.to_json(result, fallback=str, indent=2).decode() + + return [TextContent(type="text", text=result)] diff --git a/tests/client/test_output_schema_validation.py b/tests/client/test_output_schema_validation.py new file mode 100644 index 000000000..242515b96 --- /dev/null +++ b/tests/client/test_output_schema_validation.py @@ -0,0 +1,198 @@ +import logging +from contextlib import contextmanager +from unittest.mock import patch + +import pytest + +from mcp.server.lowlevel import Server +from mcp.shared.memory import ( + create_connected_server_and_client_session as client_session, +) +from mcp.types import Tool + + +@contextmanager +def bypass_server_output_validation(): + """ + Context manager that bypasses server-side output validation. + This simulates a malicious or non-compliant server that doesn't validate + its outputs, allowing us to test client-side validation. + """ + # Patch jsonschema.validate in the server module to disable all validation + with patch("mcp.server.lowlevel.server.jsonschema.validate"): + # The mock will simply return None (do nothing) for all validation calls + yield + + +class TestClientOutputSchemaValidation: + """Test client-side validation of structured output from tools""" + + @pytest.mark.anyio + async def test_tool_structured_output_client_side_validation_basemodel(self): + """Test that client validates structured content against schema for BaseModel outputs""" + # Create a malicious low-level server that returns invalid structured content + server = Server("test-server") + + # Define the expected schema for our tool + output_schema = { + "type": "object", + "properties": {"name": {"type": "string", "title": "Name"}, "age": {"type": "integer", "title": "Age"}}, + "required": ["name", "age"], + "title": "UserOutput", + } + + @server.list_tools() + async def list_tools(): + return [ + Tool( + name="get_user", + description="Get user data", + inputSchema={"type": "object"}, + outputSchema=output_schema, + ) + ] + + @server.call_tool() + async def call_tool(name: str, arguments: dict): + # Return invalid structured content - age is string instead of integer + # The low-level server will wrap this in CallToolResult + return {"name": "John", "age": "invalid"} # Invalid: age should be int + + # Test that client validates the structured content + with bypass_server_output_validation(): + async with client_session(server) as client: + # The client validates structured content and should raise an error + with pytest.raises(RuntimeError) as exc_info: + await client.call_tool("get_user", {}) + # Verify it's a validation error + assert "Invalid structured content returned by tool get_user" in str(exc_info.value) + + @pytest.mark.anyio + async def test_tool_structured_output_client_side_validation_primitive(self): + """Test that client validates structured content for primitive outputs""" + server = Server("test-server") + + # Primitive types are wrapped in {"result": value} + output_schema = { + "type": "object", + "properties": {"result": {"type": "integer", "title": "Result"}}, + "required": ["result"], + "title": "calculate_Output", + } + + @server.list_tools() + async def list_tools(): + return [ + Tool( + name="calculate", + description="Calculate something", + inputSchema={"type": "object"}, + outputSchema=output_schema, + ) + ] + + @server.call_tool() + async def call_tool(name: str, arguments: dict): + # Return invalid structured content - result is string instead of integer + return {"result": "not_a_number"} # Invalid: should be int + + with bypass_server_output_validation(): + async with client_session(server) as client: + # The client validates structured content and should raise an error + with pytest.raises(RuntimeError) as exc_info: + await client.call_tool("calculate", {}) + assert "Invalid structured content returned by tool calculate" in str(exc_info.value) + + @pytest.mark.anyio + async def test_tool_structured_output_client_side_validation_dict_typed(self): + """Test that client validates dict[str, T] structured content""" + server = Server("test-server") + + # dict[str, int] schema + output_schema = {"type": "object", "additionalProperties": {"type": "integer"}, "title": "get_scores_Output"} + + @server.list_tools() + async def list_tools(): + return [ + Tool( + name="get_scores", + description="Get scores", + inputSchema={"type": "object"}, + outputSchema=output_schema, + ) + ] + + @server.call_tool() + async def call_tool(name: str, arguments: dict): + # Return invalid structured content - values should be integers + return {"alice": "100", "bob": "85"} # Invalid: values should be int + + with bypass_server_output_validation(): + async with client_session(server) as client: + # The client validates structured content and should raise an error + with pytest.raises(RuntimeError) as exc_info: + await client.call_tool("get_scores", {}) + assert "Invalid structured content returned by tool get_scores" in str(exc_info.value) + + @pytest.mark.anyio + async def test_tool_structured_output_client_side_validation_missing_required(self): + """Test that client validates missing required fields""" + server = Server("test-server") + + output_schema = { + "type": "object", + "properties": {"name": {"type": "string"}, "age": {"type": "integer"}, "email": {"type": "string"}}, + "required": ["name", "age", "email"], # All fields required + "title": "PersonOutput", + } + + @server.list_tools() + async def list_tools(): + return [ + Tool( + name="get_person", + description="Get person data", + inputSchema={"type": "object"}, + outputSchema=output_schema, + ) + ] + + @server.call_tool() + async def call_tool(name: str, arguments: dict): + # Return structured content missing required field 'email' + return {"name": "John", "age": 30} # Missing required 'email' + + with bypass_server_output_validation(): + async with client_session(server) as client: + # The client validates structured content and should raise an error + with pytest.raises(RuntimeError) as exc_info: + await client.call_tool("get_person", {}) + assert "Invalid structured content returned by tool get_person" in str(exc_info.value) + + @pytest.mark.anyio + async def test_tool_not_listed_warning(self, caplog): + """Test that client logs warning when tool is not in list_tools but has outputSchema""" + server = Server("test-server") + + @server.list_tools() + async def list_tools(): + # Return empty list - tool is not listed + return [] + + @server.call_tool() + async def call_tool(name: str, arguments: dict): + # Server still responds to the tool call with structured content + return {"result": 42} + + # Set logging level to capture warnings + caplog.set_level(logging.WARNING) + + with bypass_server_output_validation(): + async with client_session(server) as client: + # Call a tool that wasn't listed + result = await client.call_tool("mystery_tool", {}) + assert result.structuredContent == {"result": 42} + assert result.isError is False + + # Check that warning was logged + assert "Tool mystery_tool not listed" in caplog.text diff --git a/tests/issues/test_88_random_error.py b/tests/issues/test_88_random_error.py index 7ba970f0b..d595ed022 100644 --- a/tests/issues/test_88_random_error.py +++ b/tests/issues/test_88_random_error.py @@ -8,6 +8,7 @@ import pytest from anyio.abc import TaskStatus +from mcp import types from mcp.client.session import ClientSession from mcp.server.lowlevel import Server from mcp.shared.exceptions import McpError @@ -30,6 +31,21 @@ async def test_notification_validation_error(tmp_path: Path): slow_request_started = anyio.Event() slow_request_complete = anyio.Event() + @server.list_tools() + async def list_tools() -> list[types.Tool]: + return [ + types.Tool( + name="slow", + description="A slow tool", + inputSchema={"type": "object"}, + ), + types.Tool( + name="fast", + description="A fast tool", + inputSchema={"type": "object"}, + ), + ] + @server.call_tool() async def slow_tool(name: str, arg) -> Sequence[ContentBlock]: nonlocal request_count diff --git a/tests/server/fastmcp/test_func_metadata.py b/tests/server/fastmcp/test_func_metadata.py index b13685e88..aa695c762 100644 --- a/tests/server/fastmcp/test_func_metadata.py +++ b/tests/server/fastmcp/test_func_metadata.py @@ -1,4 +1,5 @@ -from typing import Annotated +from dataclasses import dataclass +from typing import Annotated, Any, TypedDict import annotated_types import pytest @@ -193,6 +194,50 @@ def func_with_many_params(keep_this: int, skip_this: str, also_keep: float, also assert model.also_keep == 2.5 # type: ignore +def test_structured_output_dict_str_types(): + """Test that dict[str, T] types are handled without wrapping.""" + + # Test dict[str, Any] + def func_dict_any() -> dict[str, Any]: + return {"a": 1, "b": "hello", "c": [1, 2, 3]} + + meta = func_metadata(func_dict_any) + assert meta.output_schema == { + "type": "object", + "title": "func_dict_anyDictOutput", + } + + # Test dict[str, str] + def func_dict_str() -> dict[str, str]: + return {"name": "John", "city": "NYC"} + + meta = func_metadata(func_dict_str) + assert meta.output_schema == { + "type": "object", + "additionalProperties": {"type": "string"}, + "title": "func_dict_strDictOutput", + } + + # Test dict[str, list[int]] + def func_dict_list() -> dict[str, list[int]]: + return {"nums": [1, 2, 3], "more": [4, 5, 6]} + + meta = func_metadata(func_dict_list) + assert meta.output_schema == { + "type": "object", + "additionalProperties": {"type": "array", "items": {"type": "integer"}}, + "title": "func_dict_listDictOutput", + } + + # Test dict[int, str] - should be wrapped since key is not str + def func_dict_int_key() -> dict[int, str]: + return {1: "a", 2: "b"} + + meta = func_metadata(func_dict_int_key) + assert meta.output_schema is not None + assert "result" in meta.output_schema["properties"] + + @pytest.mark.anyio async def test_lambda_function(): """Test lambda function schema and validation""" @@ -408,3 +453,390 @@ def func_with_str_and_int(a: str, b: int): result = meta.pre_parse_json({"a": "123", "b": 123}) assert result["a"] == "123" assert result["b"] == 123 + + +# Tests for structured output functionality + + +def test_structured_output_requires_return_annotation(): + """Test that structured_output=True requires a return annotation""" + from mcp.server.fastmcp.exceptions import InvalidSignature + + def func_no_annotation(): + return "hello" + + def func_none_annotation() -> None: + return None + + with pytest.raises(InvalidSignature) as exc_info: + func_metadata(func_no_annotation, structured_output=True) + assert "return annotation required" in str(exc_info.value) + + # None annotation should work + meta = func_metadata(func_none_annotation) + assert meta.output_schema == { + "type": "object", + "properties": {"result": {"title": "Result", "type": "null"}}, + "required": ["result"], + "title": "func_none_annotationOutput", + } + + +def test_structured_output_basemodel(): + """Test structured output with BaseModel return types""" + + class PersonModel(BaseModel): + name: str + age: int + email: str | None = None + + def func_returning_person() -> PersonModel: + return PersonModel(name="Alice", age=30) + + meta = func_metadata(func_returning_person) + assert meta.output_schema == { + "type": "object", + "properties": { + "name": {"title": "Name", "type": "string"}, + "age": {"title": "Age", "type": "integer"}, + "email": {"anyOf": [{"type": "string"}, {"type": "null"}], "default": None, "title": "Email"}, + }, + "required": ["name", "age"], + "title": "PersonModel", + } + + +def test_structured_output_primitives(): + """Test structured output with primitive return types""" + + def func_str() -> str: + return "hello" + + def func_int() -> int: + return 42 + + def func_float() -> float: + return 3.14 + + def func_bool() -> bool: + return True + + def func_bytes() -> bytes: + return b"data" + + # Test string + meta = func_metadata(func_str) + assert meta.output_schema == { + "type": "object", + "properties": {"result": {"title": "Result", "type": "string"}}, + "required": ["result"], + "title": "func_strOutput", + } + + # Test int + meta = func_metadata(func_int) + assert meta.output_schema == { + "type": "object", + "properties": {"result": {"title": "Result", "type": "integer"}}, + "required": ["result"], + "title": "func_intOutput", + } + + # Test float + meta = func_metadata(func_float) + assert meta.output_schema == { + "type": "object", + "properties": {"result": {"title": "Result", "type": "number"}}, + "required": ["result"], + "title": "func_floatOutput", + } + + # Test bool + meta = func_metadata(func_bool) + assert meta.output_schema == { + "type": "object", + "properties": {"result": {"title": "Result", "type": "boolean"}}, + "required": ["result"], + "title": "func_boolOutput", + } + + # Test bytes + meta = func_metadata(func_bytes) + assert meta.output_schema == { + "type": "object", + "properties": {"result": {"title": "Result", "type": "string", "format": "binary"}}, + "required": ["result"], + "title": "func_bytesOutput", + } + + +def test_structured_output_generic_types(): + """Test structured output with generic types (list, dict, Union, etc.)""" + + def func_list_str() -> list[str]: + return ["a", "b", "c"] + + def func_dict_str_int() -> dict[str, int]: + return {"a": 1, "b": 2} + + def func_union() -> str | int: + return "hello" + + def func_optional() -> str | None: + return None + + # Test list + meta = func_metadata(func_list_str) + assert meta.output_schema == { + "type": "object", + "properties": {"result": {"title": "Result", "type": "array", "items": {"type": "string"}}}, + "required": ["result"], + "title": "func_list_strOutput", + } + + # Test dict[str, int] - should NOT be wrapped + meta = func_metadata(func_dict_str_int) + assert meta.output_schema == { + "type": "object", + "additionalProperties": {"type": "integer"}, + "title": "func_dict_str_intDictOutput", + } + + # Test Union + meta = func_metadata(func_union) + assert meta.output_schema == { + "type": "object", + "properties": {"result": {"title": "Result", "anyOf": [{"type": "string"}, {"type": "integer"}]}}, + "required": ["result"], + "title": "func_unionOutput", + } + + # Test Optional + meta = func_metadata(func_optional) + assert meta.output_schema == { + "type": "object", + "properties": {"result": {"title": "Result", "anyOf": [{"type": "string"}, {"type": "null"}]}}, + "required": ["result"], + "title": "func_optionalOutput", + } + + +def test_structured_output_dataclass(): + """Test structured output with dataclass return types""" + + @dataclass + class PersonDataClass: + name: str + age: int + email: str | None = None + tags: list[str] | None = None + + def func_returning_dataclass() -> PersonDataClass: + return PersonDataClass(name="Bob", age=25) + + meta = func_metadata(func_returning_dataclass) + assert meta.output_schema == { + "type": "object", + "properties": { + "name": {"title": "Name", "type": "string"}, + "age": {"title": "Age", "type": "integer"}, + "email": {"anyOf": [{"type": "string"}, {"type": "null"}], "default": None, "title": "Email"}, + "tags": { + "anyOf": [{"items": {"type": "string"}, "type": "array"}, {"type": "null"}], + "default": None, + "title": "Tags", + }, + }, + "required": ["name", "age"], + "title": "PersonDataClass", + } + + +def test_structured_output_typeddict(): + """Test structured output with TypedDict return types""" + + class PersonTypedDictOptional(TypedDict, total=False): + name: str + age: int + + 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}, + }, + "title": "PersonTypedDictOptional", + } + + # Test with total=True (all required) + class PersonTypedDictRequired(TypedDict): + name: str + age: int + email: str | None + + def func_returning_typeddict_required() -> PersonTypedDictRequired: + return {"name": "Eve", "age": 40, "email": None} # Testing None value + + meta = func_metadata(func_returning_typeddict_required) + assert meta.output_schema == { + "type": "object", + "properties": { + "name": {"title": "Name", "type": "string"}, + "age": {"title": "Age", "type": "integer"}, + "email": {"anyOf": [{"type": "string"}, {"type": "null"}], "title": "Email"}, + }, + "required": ["name", "age", "email"], + "title": "PersonTypedDictRequired", + } + + +def test_structured_output_ordinary_class(): + """Test structured output with ordinary annotated classes""" + + class PersonClass: + name: str + age: int + email: str | None + + def __init__(self, name: str, age: int, email: str | None = None): + self.name = name + self.age = age + self.email = email + + def func_returning_class() -> PersonClass: + return PersonClass("Helen", 55) + + meta = func_metadata(func_returning_class) + assert meta.output_schema == { + "type": "object", + "properties": { + "name": {"title": "Name", "type": "string"}, + "age": {"title": "Age", "type": "integer"}, + "email": {"anyOf": [{"type": "string"}, {"type": "null"}], "title": "Email"}, + }, + "required": ["name", "age", "email"], + "title": "PersonClass", + } + + +def test_unstructured_output_unannotated_class(): + # Test with class that has no annotations + class UnannotatedClass: + def __init__(self, x, y): + self.x = x + self.y = y + + def func_returning_unannotated() -> UnannotatedClass: + return UnannotatedClass(1, 2) + + meta = func_metadata(func_returning_unannotated) + assert meta.output_schema is None + + +def test_structured_output_with_field_descriptions(): + """Test that Field descriptions are preserved in structured output""" + + class ModelWithDescriptions(BaseModel): + name: Annotated[str, Field(description="The person's full name")] + age: Annotated[int, Field(description="Age in years", ge=0, le=150)] + + def func_with_descriptions() -> ModelWithDescriptions: + return ModelWithDescriptions(name="Ian", age=60) + + meta = func_metadata(func_with_descriptions) + assert meta.output_schema == { + "type": "object", + "properties": { + "name": {"title": "Name", "type": "string", "description": "The person's full name"}, + "age": {"title": "Age", "type": "integer", "description": "Age in years", "minimum": 0, "maximum": 150}, + }, + "required": ["name", "age"], + "title": "ModelWithDescriptions", + } + + +def test_structured_output_nested_models(): + """Test structured output with nested models""" + + class Address(BaseModel): + street: str + city: str + zipcode: str + + class PersonWithAddress(BaseModel): + name: str + address: Address + + def func_nested() -> PersonWithAddress: + return PersonWithAddress(name="Jack", address=Address(street="123 Main St", city="Anytown", zipcode="12345")) + + meta = func_metadata(func_nested) + assert meta.output_schema == { + "type": "object", + "$defs": { + "Address": { + "type": "object", + "properties": { + "street": {"title": "Street", "type": "string"}, + "city": {"title": "City", "type": "string"}, + "zipcode": {"title": "Zipcode", "type": "string"}, + }, + "required": ["street", "city", "zipcode"], + "title": "Address", + } + }, + "properties": { + "name": {"title": "Name", "type": "string"}, + "address": {"$ref": "#/$defs/Address"}, + }, + "required": ["name", "address"], + "title": "PersonWithAddress", + } + + +def test_structured_output_unserializable_type_error(): + """Test error when structured_output=True is used with unserializable types""" + from typing import NamedTuple + + from mcp.server.fastmcp.exceptions import InvalidSignature + + # Test with a class that has non-serializable default values + class ConfigWithCallable: + name: str + # Callable defaults are not JSON serializable and will trigger Pydantic warnings + callback: Any = lambda x: x * 2 + + def func_returning_config_with_callable() -> ConfigWithCallable: + return ConfigWithCallable() + + # Should work without structured_output=True (returns None for output_schema) + meta = func_metadata(func_returning_config_with_callable) + assert meta.output_schema is None + + # Should raise error with structured_output=True + with pytest.raises(InvalidSignature) as exc_info: + func_metadata(func_returning_config_with_callable, structured_output=True) + assert "is not serializable for structured output" in str(exc_info.value) + assert "ConfigWithCallable" in str(exc_info.value) + + # Also test with NamedTuple for good measure + class Point(NamedTuple): + x: int + y: int + + def func_returning_namedtuple() -> Point: + return Point(1, 2) + + # Should work without structured_output=True (returns None for output_schema) + meta = func_metadata(func_returning_namedtuple) + assert meta.output_schema is None + + # Should raise error with structured_output=True + with pytest.raises(InvalidSignature) as exc_info: + func_metadata(func_returning_namedtuple, structured_output=True) + assert "is not serializable for structured output" in str(exc_info.value) + assert "Point" in str(exc_info.value) diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py index 8719b78d5..c30930f7b 100644 --- a/tests/server/fastmcp/test_server.py +++ b/tests/server/fastmcp/test_server.py @@ -1,10 +1,10 @@ import base64 from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from unittest.mock import patch import pytest -from pydantic import AnyUrl +from pydantic import AnyUrl, BaseModel from starlette.routing import Mount, Route from mcp.server.fastmcp import Context, FastMCP @@ -274,6 +274,9 @@ async def test_tool_return_value_conversion(self): content = result.content[0] assert isinstance(content, TextContent) assert content.text == "3" + # Check structured content - int return type should have structured output + assert result.structuredContent is not None + assert result.structuredContent == {"result": 3} @pytest.mark.anyio async def test_tool_image_helper(self, tmp_path: Path): @@ -293,6 +296,8 @@ async def test_tool_image_helper(self, tmp_path: Path): # Verify base64 encoding decoded = base64.b64decode(content.data) assert decoded == b"fake png data" + # Check structured content - Image return type should NOT have structured output + assert result.structuredContent is None @pytest.mark.anyio async def test_tool_mixed_content(self): @@ -310,6 +315,20 @@ async def test_tool_mixed_content(self): assert isinstance(content3, AudioContent) assert content3.mimeType == "audio/wav" assert content3.data == "def" + assert result.structuredContent is not None + assert "result" in result.structuredContent + structured_result = result.structuredContent["result"] + assert len(structured_result) == 3 + + expected_content = [ + {"type": "text", "text": "Hello"}, + {"type": "image", "data": "abc", "mimeType": "image/png"}, + {"type": "audio", "data": "def", "mimeType": "audio/wav"}, + ] + + for i, expected in enumerate(expected_content): + for key, value in expected.items(): + assert structured_result[i][key] == value @pytest.mark.anyio async def test_tool_mixed_list_with_image(self, tmp_path: Path): @@ -349,6 +368,169 @@ def mixed_list_fn() -> list: content4 = result.content[3] assert isinstance(content4, TextContent) assert content4.text == "direct content" + # Check structured content - untyped list with Image objects should NOT have structured output + assert result.structuredContent is None + + @pytest.mark.anyio + async def test_tool_structured_output_basemodel(self): + """Test tool with structured output returning BaseModel""" + + class UserOutput(BaseModel): + name: str + age: int + active: bool = True + + def get_user(user_id: int) -> UserOutput: + """Get user by ID""" + return UserOutput(name="John Doe", age=30) + + mcp = FastMCP() + mcp.add_tool(get_user) + + async with client_session(mcp._mcp_server) as client: + # Check that the tool has outputSchema + tools = await client.list_tools() + tool = next(t for t in tools.tools if t.name == "get_user") + assert tool.outputSchema is not None + assert tool.outputSchema["type"] == "object" + assert "name" in tool.outputSchema["properties"] + assert "age" in tool.outputSchema["properties"] + + # Call the tool and check structured output + result = await client.call_tool("get_user", {"user_id": 123}) + assert result.isError is False + assert result.structuredContent is not None + assert result.structuredContent == {"name": "John Doe", "age": 30, "active": True} + # Content should be JSON serialized version + assert len(result.content) == 1 + assert isinstance(result.content[0], TextContent) + assert '"name": "John Doe"' in result.content[0].text + + @pytest.mark.anyio + async def test_tool_structured_output_primitive(self): + """Test tool with structured output returning primitive type""" + + def calculate_sum(a: int, b: int) -> int: + """Add two numbers""" + return a + b + + mcp = FastMCP() + mcp.add_tool(calculate_sum) + + async with client_session(mcp._mcp_server) as client: + # Check that the tool has outputSchema + tools = await client.list_tools() + tool = next(t for t in tools.tools if t.name == "calculate_sum") + assert tool.outputSchema is not None + # Primitive types are wrapped + assert tool.outputSchema["type"] == "object" + assert "result" in tool.outputSchema["properties"] + assert tool.outputSchema["properties"]["result"]["type"] == "integer" + + # Call the tool + result = await client.call_tool("calculate_sum", {"a": 5, "b": 7}) + assert result.isError is False + assert result.structuredContent is not None + assert result.structuredContent == {"result": 12} + + @pytest.mark.anyio + async def test_tool_structured_output_list(self): + """Test tool with structured output returning list""" + + def get_numbers() -> list[int]: + """Get a list of numbers""" + return [1, 2, 3, 4, 5] + + mcp = FastMCP() + mcp.add_tool(get_numbers) + + async with client_session(mcp._mcp_server) as client: + result = await client.call_tool("get_numbers", {}) + assert result.isError is False + assert result.structuredContent is not None + assert result.structuredContent == {"result": [1, 2, 3, 4, 5]} + + @pytest.mark.anyio + async def test_tool_structured_output_server_side_validation_error(self): + """Test that server-side validation errors are handled properly""" + + def get_numbers() -> list[int]: + return [1, 2, 3, 4, [5]] # type: ignore + + mcp = FastMCP() + mcp.add_tool(get_numbers) + + async with client_session(mcp._mcp_server) as client: + result = await client.call_tool("get_numbers", {}) + assert result.isError is True + assert result.structuredContent is None + assert len(result.content) == 1 + assert isinstance(result.content[0], TextContent) + + @pytest.mark.anyio + async def test_tool_structured_output_dict_str_any(self): + """Test tool with dict[str, Any] structured output""" + + def get_metadata() -> dict[str, Any]: + """Get metadata dictionary""" + return { + "version": "1.0.0", + "enabled": True, + "count": 42, + "tags": ["production", "stable"], + "config": {"nested": {"value": 123}}, + } + + mcp = FastMCP() + mcp.add_tool(get_metadata) + + async with client_session(mcp._mcp_server) as client: + # Check schema + tools = await client.list_tools() + tool = next(t for t in tools.tools if t.name == "get_metadata") + assert tool.outputSchema is not None + assert tool.outputSchema["type"] == "object" + # dict[str, Any] should have minimal schema + assert ( + "additionalProperties" not in tool.outputSchema or tool.outputSchema.get("additionalProperties") is True + ) + + # Call tool + result = await client.call_tool("get_metadata", {}) + assert result.isError is False + assert result.structuredContent is not None + expected = { + "version": "1.0.0", + "enabled": True, + "count": 42, + "tags": ["production", "stable"], + "config": {"nested": {"value": 123}}, + } + assert result.structuredContent == expected + + @pytest.mark.anyio + async def test_tool_structured_output_dict_str_typed(self): + """Test tool with dict[str, T] structured output for specific T""" + + def get_settings() -> dict[str, str]: + """Get settings as string dictionary""" + return {"theme": "dark", "language": "en", "timezone": "UTC"} + + mcp = FastMCP() + mcp.add_tool(get_settings) + + async with client_session(mcp._mcp_server) as client: + # Check schema + tools = await client.list_tools() + tool = next(t for t in tools.tools if t.name == "get_settings") + assert tool.outputSchema is not None + assert tool.outputSchema["type"] == "object" + assert tool.outputSchema["additionalProperties"]["type"] == "string" + + # Call tool + result = await client.call_tool("get_settings", {}) + assert result.isError is False + assert result.structuredContent == {"theme": "dark", "language": "en", "timezone": "UTC"} class TestServerResources: diff --git a/tests/server/fastmcp/test_tool_manager.py b/tests/server/fastmcp/test_tool_manager.py index 206df42d7..4b2052da5 100644 --- a/tests/server/fastmcp/test_tool_manager.py +++ b/tests/server/fastmcp/test_tool_manager.py @@ -1,5 +1,7 @@ import json import logging +from dataclasses import dataclass +from typing import Any, TypedDict import pytest from pydantic import BaseModel @@ -10,7 +12,7 @@ from mcp.server.fastmcp.utilities.func_metadata import ArgModelBase, FuncMetadata from mcp.server.session import ServerSessionT from mcp.shared.context import LifespanContextT, RequestT -from mcp.types import ToolAnnotations +from mcp.types import TextContent, ToolAnnotations class TestAddTools: @@ -450,3 +452,184 @@ def echo(message: str) -> str: assert tools[0].annotations is not None assert tools[0].annotations.title == "Echo Tool" assert tools[0].annotations.readOnlyHint is True + + +class TestStructuredOutput: + """Test structured output functionality in tools.""" + + @pytest.mark.anyio + async def test_tool_with_basemodel_output(self): + """Test tool with BaseModel return type.""" + + class UserOutput(BaseModel): + name: str + age: int + + def get_user(user_id: int) -> UserOutput: + """Get user by ID.""" + return UserOutput(name="John", age=30) + + manager = ToolManager() + manager.add_tool(get_user) + result = await manager.call_tool("get_user", {"user_id": 1}, convert_result=True) + # don't test unstructured output here, just the structured conversion + assert len(result) == 2 and result[1] == {"name": "John", "age": 30} + + @pytest.mark.anyio + async def test_tool_with_primitive_output(self): + """Test tool with primitive return type.""" + + def double_number(n: int) -> int: + """Double a number.""" + return 10 + + manager = ToolManager() + manager.add_tool(double_number) + result = await manager.call_tool("double_number", {"n": 5}) + assert result == 10 + result = await manager.call_tool("double_number", {"n": 5}, convert_result=True) + assert isinstance(result[0][0], TextContent) and result[1] == {"result": 10} + + @pytest.mark.anyio + async def test_tool_with_typeddict_output(self): + """Test tool with TypedDict return type.""" + + class UserDict(TypedDict): + name: str + age: int + + expected_output = {"name": "Alice", "age": 25} + + def get_user_dict(user_id: int) -> UserDict: + """Get user as dict.""" + return UserDict(name="Alice", age=25) + + manager = ToolManager() + manager.add_tool(get_user_dict) + result = await manager.call_tool("get_user_dict", {"user_id": 1}) + assert result == expected_output + + @pytest.mark.anyio + async def test_tool_with_dataclass_output(self): + """Test tool with dataclass return type.""" + + @dataclass + class Person: + name: str + age: int + + expected_output = {"name": "Bob", "age": 40} + + def get_person() -> Person: + """Get a person.""" + return Person("Bob", 40) + + manager = ToolManager() + manager.add_tool(get_person) + result = await manager.call_tool("get_person", {}, convert_result=True) + # don't test unstructured output here, just the structured conversion + assert len(result) == 2 and result[1] == expected_output + + @pytest.mark.anyio + async def test_tool_with_list_output(self): + """Test tool with list return type.""" + + expected_list = [1, 2, 3, 4, 5] + expected_output = {"result": expected_list} + + def get_numbers() -> list[int]: + """Get a list of numbers.""" + return expected_list + + manager = ToolManager() + manager.add_tool(get_numbers) + result = await manager.call_tool("get_numbers", {}) + assert result == expected_list + result = await manager.call_tool("get_numbers", {}, convert_result=True) + assert isinstance(result[0][0], TextContent) and result[1] == expected_output + + @pytest.mark.anyio + async def test_tool_without_structured_output(self): + """Test that tools work normally when structured_output=False.""" + + def get_dict() -> dict: + """Get a dict.""" + return {"key": "value"} + + manager = ToolManager() + manager.add_tool(get_dict, structured_output=False) + result = await manager.call_tool("get_dict", {}) + assert isinstance(result, dict) + assert result == {"key": "value"} + + def test_tool_output_schema_property(self): + """Test that Tool.output_schema property works correctly.""" + + class UserOutput(BaseModel): + name: str + age: int + + def get_user() -> UserOutput: + return UserOutput(name="Test", age=25) + + manager = ToolManager() + tool = manager.add_tool(get_user) + + # Test that output_schema is populated + expected_schema = { + "properties": {"name": {"type": "string", "title": "Name"}, "age": {"type": "integer", "title": "Age"}}, + "required": ["name", "age"], + "title": "UserOutput", + "type": "object", + } + assert tool.output_schema == expected_schema + + @pytest.mark.anyio + async def test_tool_with_dict_str_any_output(self): + """Test tool with dict[str, Any] return type.""" + + def get_config() -> dict[str, Any]: + """Get configuration""" + return {"debug": True, "port": 8080, "features": ["auth", "logging"]} + + manager = ToolManager() + tool = manager.add_tool(get_config) + + # Check output schema + assert tool.output_schema is not None + assert tool.output_schema["type"] == "object" + assert "properties" not in tool.output_schema # dict[str, Any] has no constraints + + # Test raw result + result = await manager.call_tool("get_config", {}) + expected = {"debug": True, "port": 8080, "features": ["auth", "logging"]} + assert result == expected + + # Test converted result + result = await manager.call_tool("get_config", {}) + assert result == expected + + @pytest.mark.anyio + async def test_tool_with_dict_str_typed_output(self): + """Test tool with dict[str, T] return type for specific T.""" + + def get_scores() -> dict[str, int]: + """Get player scores""" + return {"alice": 100, "bob": 85, "charlie": 92} + + manager = ToolManager() + tool = manager.add_tool(get_scores) + + # Check output schema + assert tool.output_schema is not None + assert tool.output_schema["type"] == "object" + assert tool.output_schema["additionalProperties"]["type"] == "integer" + + # Test raw result + result = await manager.call_tool("get_scores", {}) + expected = {"alice": 100, "bob": 85, "charlie": 92} + assert result == expected + + # Test converted result + result = await manager.call_tool("get_scores", {}) + assert result == expected