mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 21:44:18 +02:00
Add structured output support
- Tools can return StructuredOutput() for explicit structured data - Tools with output_schema automatically return structured data - Client hydrates structured responses into typed objects - Maintains backward compatibility with unstructured-only tools
This commit is contained in:
Jeremiah Lowin
parent
7f0622b707
commit
21d4ce092c
5 changed files with 592 additions and 12 deletions
|
|
@ -30,6 +30,7 @@ from fastmcp.client.sampling import SamplingHandler, create_sampling_callback
|
|||
from fastmcp.exceptions import ToolError
|
||||
from fastmcp.server import FastMCP
|
||||
from fastmcp.utilities.exceptions import get_catch_handlers
|
||||
from fastmcp.utilities.json_schema_type import json_schema_to_type
|
||||
from fastmcp.utilities.mcp_config import MCPConfig
|
||||
|
||||
from .transports import (
|
||||
|
|
@ -675,7 +676,7 @@ class Client(Generic[ClientTransportT]):
|
|||
arguments: dict[str, Any] | None = None,
|
||||
timeout: datetime.timedelta | float | int | None = None,
|
||||
progress_handler: ProgressHandler | None = None,
|
||||
) -> list[ContentBlock]:
|
||||
) -> list[ContentBlock] | dict[str, Any] | type:
|
||||
"""Call a tool on the server.
|
||||
|
||||
Unlike call_tool_mcp, this method raises a ToolError if the tool call results in an error.
|
||||
|
|
@ -687,8 +688,12 @@ class Client(Generic[ClientTransportT]):
|
|||
progress_handler (ProgressHandler | None, optional): The progress handler to use for the tool call. Defaults to None.
|
||||
|
||||
Returns:
|
||||
list[mcp.types.TextContent | mcp.types.ImageContent | mcp.types.AudioContent | mcp.types.EmbeddedResource]:
|
||||
The content returned by the tool.
|
||||
list[ContentBlock] | dict[str, Any]:
|
||||
The content returned by the tool. If the tool returns structured
|
||||
outputs, they are returned as a dictionary; otherwise, a list of
|
||||
content blocks is returned. Note: to receive both structured and
|
||||
unstructured outputs, use call_tool_mcp instead and access the
|
||||
raw result object.
|
||||
|
||||
Raises:
|
||||
ToolError: If the tool call results in an error.
|
||||
|
|
@ -703,4 +708,16 @@ class Client(Generic[ClientTransportT]):
|
|||
if result.isError:
|
||||
msg = cast(mcp.types.TextContent, result.content[0]).text
|
||||
raise ToolError(msg)
|
||||
return result.content
|
||||
elif result.structuredContent:
|
||||
if name not in self.session._tool_output_schemas:
|
||||
# refresh output schema cache
|
||||
await self.session.list_tools()
|
||||
if name in self.session._tool_output_schemas:
|
||||
output_schema = self.session._tool_output_schemas.get(name)
|
||||
if output_schema:
|
||||
output_type = json_schema_to_type(output_schema)
|
||||
return output_type(**result.structuredContent)
|
||||
|
||||
return result.structuredContent
|
||||
else:
|
||||
return result.content
|
||||
|
|
|
|||
|
|
@ -63,6 +63,7 @@ from fastmcp.utilities.cache import TimedCache
|
|||
from fastmcp.utilities.components import FastMCPComponent
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
from fastmcp.utilities.mcp_config import MCPConfig
|
||||
from fastmcp.utilities.types import NotSet, NotSetT
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastmcp.client import Client
|
||||
|
|
@ -792,7 +793,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
output_schema: dict[str, Any] | None = None,
|
||||
output_schema: dict[str, Any] | NotSetT = NotSet,
|
||||
annotations: ToolAnnotations | dict[str, Any] | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
|
|
@ -806,7 +807,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
output_schema: dict[str, Any] | None = None,
|
||||
output_schema: dict[str, Any] | NotSetT = NotSet,
|
||||
annotations: ToolAnnotations | dict[str, Any] | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
|
|
@ -819,7 +820,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
output_schema: dict[str, Any] | None = None,
|
||||
output_schema: dict[str, Any] | NotSetT = NotSet,
|
||||
annotations: ToolAnnotations | dict[str, Any] | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ from fastmcp.utilities.types import (
|
|||
Image,
|
||||
NotSet,
|
||||
NotSetT,
|
||||
StructuredOutput,
|
||||
find_kwarg_by_type,
|
||||
get_cached_typeadapter,
|
||||
replace_type,
|
||||
|
|
@ -105,8 +106,18 @@ class Tool(FastMCPComponent):
|
|||
enabled=enabled,
|
||||
)
|
||||
|
||||
async def run(self, arguments: dict[str, Any]) -> list[ContentBlock]:
|
||||
"""Run the tool with arguments."""
|
||||
async def run(
|
||||
self, arguments: dict[str, Any]
|
||||
) -> list[ContentBlock] | tuple[list[ContentBlock], dict[str, Any]]:
|
||||
"""
|
||||
Run the tool with arguments.
|
||||
|
||||
This method is not implemented in the base Tool class and must be
|
||||
implemented by subclasses.
|
||||
|
||||
`run()` can EITHER return a list of ContentBlocks, or a tuple of
|
||||
(list of ContentBlocks, dict of structured output).
|
||||
"""
|
||||
raise NotImplementedError("Subclasses must implement run()")
|
||||
|
||||
@classmethod
|
||||
|
|
@ -175,7 +186,9 @@ class FunctionTool(Tool):
|
|||
enabled=enabled if enabled is not None else True,
|
||||
)
|
||||
|
||||
async def run(self, arguments: dict[str, Any]) -> list[ContentBlock]:
|
||||
async def run(
|
||||
self, arguments: dict[str, Any]
|
||||
) -> list[ContentBlock] | tuple[list[ContentBlock], dict[str, Any]]:
|
||||
"""Run the tool with arguments."""
|
||||
from fastmcp.server.context import Context
|
||||
|
||||
|
|
@ -190,7 +203,20 @@ class FunctionTool(Tool):
|
|||
if inspect.isawaitable(result):
|
||||
result = await result
|
||||
|
||||
return _convert_to_content(result, serializer=self.serializer)
|
||||
unstructured_result = _convert_to_content(result, serializer=self.serializer)
|
||||
|
||||
structured_result = None
|
||||
if isinstance(result, StructuredOutput):
|
||||
structured_result = result.to_structured_output()
|
||||
elif self.output_schema is not None:
|
||||
structured_result = pydantic_core.to_jsonable_python(result, fallback=str)
|
||||
|
||||
# return only the unstructured result if there is no structured output
|
||||
if structured_result is None:
|
||||
return unstructured_result
|
||||
|
||||
# return both the unstructured and structured results if there is structured output
|
||||
return (unstructured_result, structured_result)
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
|
|||
526
src/fastmcp/utilities/json_schema_type.py
Normal file
526
src/fastmcp/utilities/json_schema_type.py
Normal file
|
|
@ -0,0 +1,526 @@
|
|||
"""Convert JSON Schema to Python types with validation.
|
||||
|
||||
The json_schema_to_type function converts a JSON Schema into a Python type that can be used
|
||||
for validation with Pydantic. It supports:
|
||||
|
||||
- Basic types (string, number, integer, boolean, null)
|
||||
- Complex types (arrays, objects)
|
||||
- Format constraints (date-time, email, uri)
|
||||
- Numeric constraints (minimum, maximum, multipleOf)
|
||||
- String constraints (minLength, maxLength, pattern)
|
||||
- Array constraints (minItems, maxItems, uniqueItems)
|
||||
- Object properties with defaults
|
||||
- References and recursive schemas
|
||||
- Enums and constants
|
||||
- Union types
|
||||
|
||||
Example:
|
||||
```python
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string", "minLength": 1},
|
||||
"age": {"type": "integer", "minimum": 0},
|
||||
"email": {"type": "string", "format": "email"}
|
||||
},
|
||||
"required": ["name", "age"]
|
||||
}
|
||||
|
||||
# Name is optional and will be inferred from schema's "title" property if not provided
|
||||
Person = json_schema_to_type(schema)
|
||||
# Creates a validated dataclass with name, age, and optional email fields
|
||||
```
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
from collections.abc import Callable, Mapping
|
||||
from copy import deepcopy
|
||||
from dataclasses import MISSING, field, make_dataclass
|
||||
from datetime import datetime
|
||||
from enum import Enum
|
||||
from typing import (
|
||||
Annotated,
|
||||
Any,
|
||||
ForwardRef,
|
||||
Literal,
|
||||
Optional,
|
||||
Union,
|
||||
)
|
||||
|
||||
from pydantic import (
|
||||
AnyUrl,
|
||||
EmailStr,
|
||||
Field,
|
||||
Json,
|
||||
StringConstraints,
|
||||
model_validator,
|
||||
)
|
||||
from typing_extensions import NotRequired, TypedDict
|
||||
|
||||
__all__ = ["json_schema_to_type", "JSONSchema"]
|
||||
|
||||
|
||||
FORMAT_TYPES: dict[str, Any] = {
|
||||
"date-time": datetime,
|
||||
"email": EmailStr,
|
||||
"uri": AnyUrl,
|
||||
"json": Json,
|
||||
}
|
||||
|
||||
_classes: dict[tuple[str, Any], type | None] = {}
|
||||
|
||||
|
||||
class JSONSchema(TypedDict):
|
||||
type: NotRequired[str | list[str]]
|
||||
properties: NotRequired[dict[str, JSONSchema]]
|
||||
required: NotRequired[list[str]]
|
||||
additionalProperties: NotRequired[bool | JSONSchema]
|
||||
items: NotRequired[JSONSchema | list[JSONSchema]]
|
||||
enum: NotRequired[list[Any]]
|
||||
const: NotRequired[Any]
|
||||
default: NotRequired[Any]
|
||||
description: NotRequired[str]
|
||||
title: NotRequired[str]
|
||||
examples: NotRequired[list[Any]]
|
||||
format: NotRequired[str]
|
||||
allOf: NotRequired[list[JSONSchema]]
|
||||
anyOf: NotRequired[list[JSONSchema]]
|
||||
oneOf: NotRequired[list[JSONSchema]]
|
||||
not_: NotRequired[JSONSchema]
|
||||
definitions: NotRequired[dict[str, JSONSchema]]
|
||||
dependencies: NotRequired[dict[str, JSONSchema | list[str]]]
|
||||
pattern: NotRequired[str]
|
||||
minLength: NotRequired[int]
|
||||
maxLength: NotRequired[int]
|
||||
minimum: NotRequired[int | float]
|
||||
maximum: NotRequired[int | float]
|
||||
exclusiveMinimum: NotRequired[int | float]
|
||||
exclusiveMaximum: NotRequired[int | float]
|
||||
multipleOf: NotRequired[int | float]
|
||||
uniqueItems: NotRequired[bool]
|
||||
minItems: NotRequired[int]
|
||||
maxItems: NotRequired[int]
|
||||
additionalItems: NotRequired[bool | JSONSchema]
|
||||
|
||||
|
||||
def json_schema_to_type(
|
||||
schema: Mapping[str, Any],
|
||||
name: str | None = None,
|
||||
) -> type:
|
||||
"""Convert JSON schema to appropriate Python type with validation.
|
||||
|
||||
Args:
|
||||
schema: A JSON Schema dictionary defining the type structure and validation rules
|
||||
name: Optional name for object schemas. Only allowed when schema type is "object".
|
||||
If not provided for objects, name will be inferred from schema's "title"
|
||||
property or default to "Root".
|
||||
|
||||
Returns:
|
||||
A Python type (typically a dataclass for objects) with Pydantic validation
|
||||
|
||||
Raises:
|
||||
ValueError: If a name is provided for a non-object schema
|
||||
|
||||
Examples:
|
||||
Create a dataclass from an object schema:
|
||||
```python
|
||||
schema = {
|
||||
"type": "object",
|
||||
"title": "Person",
|
||||
"properties": {
|
||||
"name": {"type": "string", "minLength": 1},
|
||||
"age": {"type": "integer", "minimum": 0},
|
||||
"email": {"type": "string", "format": "email"}
|
||||
},
|
||||
"required": ["name", "age"]
|
||||
}
|
||||
|
||||
Person = json_schema_to_type(schema)
|
||||
# Creates a dataclass with name, age, and optional email fields:
|
||||
# @dataclass
|
||||
# class Person:
|
||||
# name: str
|
||||
# age: int
|
||||
# email: str | None = None
|
||||
```
|
||||
Person(name="John", age=30)
|
||||
|
||||
Create a scalar type with constraints:
|
||||
```python
|
||||
schema = {
|
||||
"type": "string",
|
||||
"minLength": 3,
|
||||
"pattern": "^[A-Z][a-z]+$"
|
||||
}
|
||||
|
||||
NameType = json_schema_to_type(schema)
|
||||
# Creates Annotated[str, StringConstraints(min_length=3, pattern="^[A-Z][a-z]+$")]
|
||||
|
||||
@dataclass
|
||||
class Name:
|
||||
name: NameType
|
||||
```
|
||||
"""
|
||||
# Always use the top-level schema for references
|
||||
if schema.get("type") == "object":
|
||||
return _create_dataclass(schema, name, schemas=schema)
|
||||
elif name:
|
||||
raise ValueError(f"Can not apply name to non-object schema: {name}")
|
||||
return _schema_to_type(schema, schemas=schema)
|
||||
|
||||
|
||||
def _hash_schema(schema: Mapping[str, Any]) -> str:
|
||||
"""Generate a deterministic hash for schema caching."""
|
||||
return hashlib.sha256(json.dumps(schema, sort_keys=True).encode()).hexdigest()
|
||||
|
||||
|
||||
def _resolve_ref(ref: str, schemas: Mapping[str, Any]) -> Mapping[str, Any]:
|
||||
"""Resolve JSON Schema reference to target schema."""
|
||||
path = ref.replace("#/", "").split("/")
|
||||
current = schemas
|
||||
for part in path:
|
||||
current = current.get(part, {})
|
||||
return current
|
||||
|
||||
|
||||
def _create_string_type(schema: Mapping[str, Any]) -> type | Annotated[Any, ...]:
|
||||
"""Create string type with optional constraints."""
|
||||
if "const" in schema:
|
||||
return Literal[schema["const"]] # type: ignore
|
||||
|
||||
if fmt := schema.get("format"):
|
||||
if fmt == "uri":
|
||||
return AnyUrl
|
||||
elif fmt == "uri-reference":
|
||||
return str
|
||||
return FORMAT_TYPES.get(fmt, str)
|
||||
|
||||
constraints = {
|
||||
k: v
|
||||
for k, v in {
|
||||
"min_length": schema.get("minLength"),
|
||||
"max_length": schema.get("maxLength"),
|
||||
"pattern": schema.get("pattern"),
|
||||
}.items()
|
||||
if v is not None
|
||||
}
|
||||
|
||||
return Annotated[str, StringConstraints(**constraints)] if constraints else str
|
||||
|
||||
|
||||
def _create_numeric_type(
|
||||
base: type[int | float], schema: Mapping[str, Any]
|
||||
) -> type | Annotated[Any, ...]:
|
||||
"""Create numeric type with optional constraints."""
|
||||
if "const" in schema:
|
||||
return Literal[schema["const"]] # type: ignore
|
||||
|
||||
constraints = {
|
||||
k: v
|
||||
for k, v in {
|
||||
"gt": schema.get("exclusiveMinimum"),
|
||||
"ge": schema.get("minimum"),
|
||||
"lt": schema.get("exclusiveMaximum"),
|
||||
"le": schema.get("maximum"),
|
||||
"multiple_of": schema.get("multipleOf"),
|
||||
}.items()
|
||||
if v is not None
|
||||
}
|
||||
|
||||
return Annotated[base, Field(**constraints)] if constraints else base
|
||||
|
||||
|
||||
def _create_enum(name: str, values: list[Any]) -> type | Enum:
|
||||
"""Create enum type from list of values."""
|
||||
if all(isinstance(v, str) for v in values):
|
||||
return Enum(name, {v.upper(): v for v in values})
|
||||
return Literal[tuple(values)] # type: ignore
|
||||
|
||||
|
||||
def _create_array_type(
|
||||
schema: Mapping[str, Any], schemas: Mapping[str, Any]
|
||||
) -> type | Annotated[Any, ...]:
|
||||
"""Create list/set type with optional constraints."""
|
||||
items = schema.get("items", {})
|
||||
if isinstance(items, list):
|
||||
# Handle positional item schemas
|
||||
item_types = [_schema_to_type(s, schemas) for s in items]
|
||||
combined = Union[tuple(item_types)] # type: ignore # noqa: UP007
|
||||
base = list[combined]
|
||||
else:
|
||||
# Handle single item schema
|
||||
item_type = _schema_to_type(items, schemas)
|
||||
base = set if schema.get("uniqueItems") else list
|
||||
base = base[item_type]
|
||||
|
||||
constraints = {
|
||||
k: v
|
||||
for k, v in {
|
||||
"min_length": schema.get("minItems"),
|
||||
"max_length": schema.get("maxItems"),
|
||||
}.items()
|
||||
if v is not None
|
||||
}
|
||||
|
||||
return Annotated[base, Field(**constraints)] if constraints else base
|
||||
|
||||
|
||||
def _return_Any() -> Any:
|
||||
return Any
|
||||
|
||||
|
||||
def _get_from_type_handler(
|
||||
schema: Mapping[str, Any], schemas: Mapping[str, Any]
|
||||
) -> Callable[..., Any]:
|
||||
"""Get the appropriate type handler for the schema."""
|
||||
|
||||
type_handlers: dict[str, Callable[..., Any]] = { # TODO
|
||||
"string": lambda s: _create_string_type(s), # type: ignore
|
||||
"integer": lambda s: _create_numeric_type(int, s), # type: ignore
|
||||
"number": lambda s: _create_numeric_type(float, s), # type: ignore
|
||||
"boolean": lambda _: bool, # type: ignore
|
||||
"null": lambda _: type(None), # type: ignore
|
||||
"array": lambda s: _create_array_type(s, schemas), # type: ignore
|
||||
"object": lambda s: _create_dataclass(s, s.get("title"), schemas), # type: ignore
|
||||
}
|
||||
return type_handlers.get(schema.get("type", None), _return_Any)
|
||||
|
||||
|
||||
def _schema_to_type(
|
||||
schema: Mapping[str, Any],
|
||||
schemas: Mapping[str, Any],
|
||||
) -> type:
|
||||
"""Convert schema to appropriate Python type."""
|
||||
if not schema:
|
||||
return object
|
||||
|
||||
if "type" not in schema and "properties" in schema:
|
||||
return _create_dataclass(schema, schema.get("title", "<unknown>"), schemas)
|
||||
|
||||
# Handle references first
|
||||
if "$ref" in schema:
|
||||
ref = schema["$ref"]
|
||||
# Handle self-reference
|
||||
if ref == "#":
|
||||
return ForwardRef(schema.get("title", "Root"))
|
||||
return _schema_to_type(_resolve_ref(ref, schemas), schemas)
|
||||
|
||||
if "const" in schema:
|
||||
return Literal[schema["const"]] # type: ignore
|
||||
|
||||
if "enum" in schema:
|
||||
return _create_enum(f"Enum_{len(_classes)}", schema["enum"])
|
||||
|
||||
schema_type = schema.get("type")
|
||||
if not schema_type:
|
||||
return Any
|
||||
|
||||
if isinstance(schema_type, list):
|
||||
# Create a copy of the schema for each type, but keep all constraints
|
||||
types: list[type | Any] = []
|
||||
for t in schema_type:
|
||||
type_schema = schema.copy()
|
||||
type_schema["type"] = t
|
||||
types.append(_schema_to_type(type_schema, schemas))
|
||||
has_null = type(None) in types
|
||||
types = [t for t in types if t is not type(None)]
|
||||
if has_null:
|
||||
return Optional[tuple(types) if len(types) > 1 else types[0]] # type: ignore # noqa: UP007
|
||||
return Union[tuple(types)] # type: ignore # noqa: UP007
|
||||
|
||||
return _get_from_type_handler(schema, schemas)(schema)
|
||||
|
||||
|
||||
def _sanitize_name(name: str) -> str:
|
||||
"""Convert string to valid Python identifier."""
|
||||
# Step 1: replace everything except [0-9a-zA-Z_] with underscores
|
||||
cleaned = re.sub(r"[^0-9a-zA-Z_]", "_", name)
|
||||
# Step 2: deduplicate underscores
|
||||
cleaned = re.sub(r"__+", "_", cleaned)
|
||||
# Step 3: if the first char of original name isn't a letter, prepend field_
|
||||
if not name or not re.match(r"[a-zA-Z]", name[0]):
|
||||
cleaned = f"field_{cleaned}"
|
||||
# Step 4: deduplicate again and strip trailing underscores
|
||||
cleaned = re.sub(r"__+", "_", cleaned).strip("_")
|
||||
return cleaned
|
||||
|
||||
|
||||
def _get_default_value(
|
||||
schema: dict[str, Any],
|
||||
prop_name: str,
|
||||
parent_default: dict[str, Any] | None = None,
|
||||
) -> Any:
|
||||
"""Get default value with proper priority ordering.
|
||||
1. Value from parent's default if it exists
|
||||
2. Property's own default if it exists
|
||||
3. None
|
||||
"""
|
||||
if parent_default is not None and prop_name in parent_default:
|
||||
return parent_default[prop_name]
|
||||
return schema.get("default")
|
||||
|
||||
|
||||
def _create_field_with_default(
|
||||
field_type: type,
|
||||
default_value: Any,
|
||||
schema: dict[str, Any],
|
||||
) -> Any:
|
||||
"""Create a field with simplified default handling."""
|
||||
# Always use None as default for complex types
|
||||
if isinstance(default_value, dict | list) or default_value is None:
|
||||
return field(default=None)
|
||||
|
||||
# For simple types, use the value directly
|
||||
return field(default=default_value)
|
||||
|
||||
|
||||
def _create_dataclass(
|
||||
schema: Mapping[str, Any],
|
||||
name: str | None = None,
|
||||
schemas: Mapping[str, Any] | None = None,
|
||||
) -> type:
|
||||
"""Create dataclass from object schema."""
|
||||
name = name or schema.get("title", "Root")
|
||||
# Sanitize name for class creation
|
||||
sanitized_name = _sanitize_name(name)
|
||||
schema_hash = _hash_schema(schema)
|
||||
cache_key = (schema_hash, sanitized_name)
|
||||
original_schema = dict(schema) # Store copy for validator
|
||||
|
||||
# Return existing class if already built
|
||||
if cache_key in _classes:
|
||||
existing = _classes[cache_key]
|
||||
if existing is None:
|
||||
return ForwardRef(sanitized_name)
|
||||
return existing
|
||||
|
||||
# Place placeholder for recursive references
|
||||
_classes[cache_key] = None
|
||||
|
||||
if "$ref" in schema:
|
||||
ref = schema["$ref"]
|
||||
if ref == "#":
|
||||
return ForwardRef(sanitized_name)
|
||||
schema = _resolve_ref(ref, schemas or {})
|
||||
|
||||
properties = schema.get("properties", {})
|
||||
required = schema.get("required", [])
|
||||
|
||||
fields: list[tuple[Any, ...]] = []
|
||||
for prop_name, prop_schema in properties.items():
|
||||
field_name = _sanitize_name(prop_name)
|
||||
|
||||
# Check for self-reference in property
|
||||
if prop_schema.get("$ref") == "#":
|
||||
field_type = ForwardRef(sanitized_name)
|
||||
else:
|
||||
field_type = _schema_to_type(prop_schema, schemas)
|
||||
|
||||
default_val = prop_schema.get("default", MISSING)
|
||||
is_required = prop_name in required
|
||||
|
||||
# Include alias in field metadata
|
||||
meta = {"alias": prop_name}
|
||||
|
||||
if default_val is not MISSING:
|
||||
if isinstance(default_val, dict | list):
|
||||
field_def = field(
|
||||
default_factory=lambda d=default_val: deepcopy(d), metadata=meta
|
||||
)
|
||||
else:
|
||||
field_def = field(default=default_val, metadata=meta)
|
||||
else:
|
||||
if is_required:
|
||||
field_def = field(metadata=meta)
|
||||
else:
|
||||
field_def = field(default=None, metadata=meta)
|
||||
|
||||
if is_required and default_val is not MISSING:
|
||||
fields.append((field_name, field_type, field_def))
|
||||
elif is_required:
|
||||
fields.append((field_name, field_type, field_def))
|
||||
else:
|
||||
fields.append((field_name, Optional[field_type], field_def))
|
||||
|
||||
cls = make_dataclass(sanitized_name, fields, kw_only=True)
|
||||
|
||||
# Add model validator for defaults
|
||||
@model_validator(mode="before")
|
||||
@classmethod
|
||||
def _apply_defaults(cls, data: Mapping[str, Any]):
|
||||
if isinstance(data, dict):
|
||||
return _merge_defaults(data, original_schema)
|
||||
return data
|
||||
|
||||
setattr(cls, "_apply_defaults", _apply_defaults)
|
||||
|
||||
# Store completed class
|
||||
_classes[cache_key] = cls
|
||||
return cls
|
||||
|
||||
|
||||
def _merge_defaults(
|
||||
data: Mapping[str, Any],
|
||||
schema: Mapping[str, Any],
|
||||
parent_default: Mapping[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Merge defaults with provided data at all levels."""
|
||||
# If we have no data
|
||||
if not data:
|
||||
# Start with parent default if available
|
||||
if parent_default:
|
||||
result = dict(parent_default)
|
||||
# Otherwise use schema default if available
|
||||
elif "default" in schema:
|
||||
result = dict(schema["default"])
|
||||
# Otherwise start empty
|
||||
else:
|
||||
result = {}
|
||||
# If we have data and a parent default, merge them
|
||||
elif parent_default:
|
||||
result = dict(parent_default)
|
||||
for key, value in data.items():
|
||||
if (
|
||||
isinstance(value, dict)
|
||||
and key in result
|
||||
and isinstance(result[key], dict)
|
||||
):
|
||||
# recursively merge nested dicts
|
||||
result[key] = _merge_defaults(value, {"properties": {}}, result[key])
|
||||
else:
|
||||
result[key] = value
|
||||
# Otherwise just use the data
|
||||
else:
|
||||
result = dict(data)
|
||||
|
||||
# For each property in the schema
|
||||
for prop_name, prop_schema in schema.get("properties", {}).items():
|
||||
# If property is missing, apply defaults in priority order
|
||||
if prop_name not in result:
|
||||
if parent_default and prop_name in parent_default:
|
||||
result[prop_name] = parent_default[prop_name]
|
||||
elif "default" in prop_schema:
|
||||
result[prop_name] = prop_schema["default"]
|
||||
|
||||
# If property exists and is an object, recursively merge
|
||||
if (
|
||||
prop_name in result
|
||||
and isinstance(result[prop_name], dict)
|
||||
and prop_schema.get("type") == "object"
|
||||
):
|
||||
# Get the appropriate default for this nested object
|
||||
nested_default = None
|
||||
if parent_default and prop_name in parent_default:
|
||||
nested_default = parent_default[prop_name]
|
||||
elif "default" in prop_schema:
|
||||
nested_default = prop_schema["default"]
|
||||
|
||||
result[prop_name] = _merge_defaults(
|
||||
result[prop_name], prop_schema, nested_default
|
||||
)
|
||||
|
||||
return result
|
||||
|
|
@ -7,7 +7,7 @@ from collections.abc import Callable
|
|||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
from types import EllipsisType, UnionType
|
||||
from typing import Annotated, TypeAlias, TypeVar, Union, get_args, get_origin
|
||||
from typing import Annotated, Any, TypeAlias, TypeVar, Union, get_args, get_origin
|
||||
|
||||
import mcp.types
|
||||
from mcp.types import Annotations
|
||||
|
|
@ -289,6 +289,16 @@ class File:
|
|||
)
|
||||
|
||||
|
||||
class StructuredOutput:
|
||||
"""Helper class for returning structured output from tools."""
|
||||
|
||||
def __init__(self, data: dict[str, Any]):
|
||||
self.data = data
|
||||
|
||||
def to_structured_output(self) -> dict[str, Any]:
|
||||
return self.data
|
||||
|
||||
|
||||
def replace_type(type_, type_map: dict[type, type]):
|
||||
"""
|
||||
Given a (possibly generic, nested, or otherwise complex) type, replaces all
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue