github-featured/modelcontextprotocol--python-sdk
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

feat: Add structured output support for tool functions (#993)

Browse Source
This commit is contained in:
bhosmer-ant
2025-06-26 04:36:17 -04:00
committed by GitHub
parent c8bbfc034d
commit 43bb24f62b
12 changed files with 1774 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+122
View File
@@ -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:
examples/fastmcp/weather_structured.py
+225
View File
@@ -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())
src/mcp/client/session.py
+42 -2
View File
@@ -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 @@ from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
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 @@ class ClientSession(
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 @@ class ClientSession(
) -> 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 @@ class ClientSession(
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 @@ class ClientSession(
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 @@ class ClientSession(
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(
src/mcp/server/fastmcp/server.py
+33 -32
View File
@@ -9,7 +9,6 @@ from contextlib import (
AbstractAsyncContextManager,
asynccontextmanager,
)
from itertools import chain
from typing import Any, Generic, Literal
import anyio
@@ -38,7 +37,6 @@ from mcp.server.fastmcp.prompts import Prompt, PromptManager
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 @@ from mcp.types import (
AnyFunction,
ContentBlock,
GetPromptResult,
TextContent,
ToolAnnotations,
)
from mcp.types import Prompt as MCPPrompt
@@ -235,7 +232,10 @@ class FastMCP:
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 @@ class FastMCP:
title=info.title,
description=info.description,
inputSchema=info.parameters,
outputSchema=info.output_schema,
annotations=info.annotations,
)
for info in tools
@@ -267,12 +268,10 @@ class FastMCP:
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 @@ class FastMCP:
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 @@ class FastMCP:
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 @@ class FastMCP:
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 @@ class FastMCP:
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 @@ class FastMCP:
)
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 @@ class FastMCP:
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.
src/mcp/server/fastmcp/tools/base.py
+14 -1
View File
@@ -3,6 +3,7 @@ from __future__ import annotations as _annotations
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 @@ class Tool(BaseModel):
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 @@ class Tool(BaseModel):
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 @@ class Tool(BaseModel):
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
src/mcp/server/fastmcp/tools/tool_manager.py
+11 -2
View File
@@ -49,9 +49,17 @@ class ToolManager:
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 @@ class ToolManager:
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)
src/mcp/server/fastmcp/utilities/func_metadata.py
+312 -20
View File
@@ -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
from pydantic import BaseModel, ConfigDict, Field, WithJsonSchema, create_model
import pydantic_core
from pydantic import (
BaseModel,
ConfigDict,
Field,
RootModel,
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 @@ class ArgModelBase(BaseModel):
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 @@ class FuncMetadata(BaseModel):
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 @@ class FuncMetadata(BaseModel):
)
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)]
tests/client/test_output_schema_validation.py
+198
View File
@@ -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
tests/issues/test_88_random_error.py
+16
View File
@@ -8,6 +8,7 @@ import anyio
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
tests/server/fastmcp/test_func_metadata.py
+433 -1
View File
@@ -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 test_skip_names():
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 test_str_vs_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)
tests/server/fastmcp/test_server.py
+184 -2
View File
@@ -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 @@ class TestServerTools:
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 @@ class TestServerTools:
# 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 @@ class TestServerTools:
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 @@ class TestServerTools:
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:
tests/server/fastmcp/test_tool_manager.py
+184 -1
View File
@@ -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.tools import Tool, ToolManager
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 @@ class TestToolAnnotations:
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