ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-21 13:04:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Simplify typed output serialization

Browse source
This commit is contained in:
Jeremiah Lowin 2026-08-06 11:44:01 -04:00
commit adf9a65eed
No known key found for this signature in database
6 changed files with 19 additions and 283 deletions
Download patch file Download diff file Expand all files Collapse all files

34
fastmcp_slim/fastmcp/tools/base.py
View file

@ -26,7 +26,6 @@ from pydantic import (
Field,
PrivateAttr,
PydanticSchemaGenerationError,
TypeAdapter,
model_validator,
)
from pydantic.json_schema import SkipJsonSchema
@ -57,7 +56,7 @@ if TYPE_CHECKING:
logger = get_logger(__name__)
_JSONABLE_ADAPTER = TypeAdapter(Any)
_JSONABLE_ADAPTER = get_cached_typeadapter(Any)
def _default_title(name: str) -> str:
@ -234,6 +233,7 @@ class Tool(FastMCPComponent):
KEY_PREFIX: ClassVar[str] = "tool"
return_type: Annotated[SkipJsonSchema[Any], Field(exclude=True)] = None
parameters: Annotated[
dict[str, Any], Field(description="JSON schema for tool parameters")
]
@ -364,10 +364,6 @@ class Tool(FastMCPComponent):
"""
raise NotImplementedError("Subclasses must implement run()")
def _serialize_output(self, raw_value: Any) -> Any:
"""Serialize a tool result to JSON-compatible Python values."""
return _serialize_to_jsonable(raw_value)
def convert_result(self, raw_value: Any) -> ToolResult:
"""Convert a raw result to ToolResult.
@ -397,17 +393,27 @@ class Tool(FastMCPComponent):
if isinstance(raw_value, bytes):
return ToolResult(content=content)
is_content_result = isinstance(
raw_value, ContentBlock | Audio | Image | File
) or (
isinstance(raw_value, list | tuple)
and any(
isinstance(item, ContentBlock | Audio | Image | File)
for item in raw_value
)
)
# Skip structured content for ContentBlock types only if no output_schema
# (if output_schema exists, MCP SDK requires structured_content)
if self.output_schema is None and _is_content_result(raw_value):
if self.output_schema is None and is_content_result:
return ToolResult(content=content)
try:
structured = self._serialize_output(raw_value)
structured = _serialize_to_jsonable(raw_value, self.return_type)
except (pydantic_core.PydanticSerializationError, UnicodeDecodeError):
return ToolResult(content=content)
if not _is_content_result(raw_value):
if not is_content_result:
content = _convert_to_content(structured)
if self.output_schema is None:
@ -592,14 +598,4 @@ def _convert_to_content(
return [TextContent(type="text", text=default_serializer(result))]
def _is_content_result(result: Any) -> bool:
"""Whether a result contains content that must retain its MCP representation."""
return isinstance(result, ContentBlock | Audio | Image | File) or (
isinstance(result, list | tuple)
and any(
isinstance(item, ContentBlock | Audio | Image | File) for item in result
)
)
__all__ = ["InputRequiredToolResult", "Tool", "ToolResult"]

6
fastmcp_slim/fastmcp/tools/function_tool.py
View file

@ -35,7 +35,6 @@ from fastmcp.tools.base import (
InputRequiredToolResult,
Tool,
ToolResult,
_serialize_to_jsonable,
)
from fastmcp.tools.function_parsing import ParsedFunction, _is_object_schema
from fastmcp.utilities.async_utils import (
@ -198,7 +197,6 @@ def _resolve_param_hints(fn: Callable[..., Any]) -> dict[str, Any]:
class FunctionTool(Tool):
fn: SkipJsonSchema[Callable[..., Any]]
return_type: Annotated[SkipJsonSchema[Any], Field(exclude=True)] = None
run_in_thread: Annotated[
bool,
Field(
@ -215,10 +213,6 @@ class FunctionTool(Tool):
),
] = True
def _serialize_output(self, raw_value: Any) -> Any:
"""Serialize using the return annotation that produced the output schema."""
return _serialize_to_jsonable(raw_value, self.return_type)
@classmethod
def from_function(
cls,

40
fastmcp_slim/fastmcp/tools/tool_transform.py
View file

@ -18,9 +18,6 @@ from fastmcp.tools.base import (
InputRequiredToolResult,
Tool,
ToolResult,
_convert_to_content,
_is_content_result,
_serialize_to_jsonable,
)
from fastmcp.tools.function_parsing import ParsedFunction
from fastmcp.utilities.async_utils import (
@ -310,16 +307,11 @@ class TransformedTool(Tool):
parent_tool: SkipJsonSchema[Tool]
fn: SkipJsonSchema[Callable[..., Any]]
return_type: Annotated[SkipJsonSchema[Any], Field(exclude=True)] = None
forwarding_fn: SkipJsonSchema[
Callable[..., Any]
] # Always present, handles arg transformation
transform_args: dict[str, ArgTransform]
def _serialize_output(self, raw_value: Any) -> Any:
"""Serialize using the custom transform's declared return type."""
return _serialize_to_jsonable(raw_value, self.return_type)
async def run(self, arguments: dict[str, Any]) -> ToolResult:
"""Run the tool with context set for forward() functions.
@ -399,37 +391,7 @@ class TransformedTool(Tool):
else:
return result
# Otherwise convert to content and create ToolResult with proper structured content
unstructured_result = _convert_to_content(result)
structured_output = None
# First handle structured content based on output schema, if any
if self.output_schema is not None:
serialized_result = self._serialize_output(result)
if self.output_schema.get("x-fastmcp-wrap-result"):
structured_output = {"result": serialized_result}
else:
structured_output = serialized_result
if not _is_content_result(result):
unstructured_result = _convert_to_content(serialized_result)
# If no output schema, try to serialize the result. If it is a dict, use
# it as structured content. If it is not a dict, ignore it.
if structured_output is None:
try:
structured_output = self._serialize_output(result)
if not _is_content_result(result):
unstructured_result = _convert_to_content(structured_output)
if not isinstance(structured_output, dict):
structured_output = None
except Exception:
pass
return ToolResult(
content=unstructured_result,
structured_content=structured_output,
)
return self.convert_result(result)
finally:
_current_tool.reset(token)

4
tests/tools/tool/test_output_schema.py
View file

@ -246,10 +246,6 @@ class TestToolFromFunctionOutputSchema:
tool = Tool.from_function(func)
assert tool.output_schema is None
result = await tool.run({})
assert result.structured_content is None
assert len(result.content) == 1
async def test_provided_output_schema_takes_precedence_over_json_compatible_annotation(
self,
):

199
tests/tools/tool/test_results.py
View file

@ -5,8 +5,6 @@ from typing import Annotated, Any
import pytest
from mcp_types import CallToolResult, TextContent
from pydantic import BaseModel, ConfigDict, Field, with_config
from pydantic.dataclasses import dataclass as pydantic_dataclass
from typing_extensions import TypedDict
from fastmcp import Client, FastMCP
from fastmcp.tools.base import Tool, ToolResult
@ -281,12 +279,7 @@ class TestSerializationAlias:
class TestSerializeByAlias:
"""Tests that a model's serialize_by_alias config is honored at runtime.
pydantic_core's serialization helpers default by_alias to True, which
silently ignores serialize_by_alias=False. The serialized result and the
generated output schema must both reflect the model's configured behavior.
"""
"""Tests that typed results use Pydantic's serialization behavior."""
async def test_serialize_by_alias_false_uses_field_names(self):
"""serialize_by_alias=False emits field names in schema, structured, and text."""
@ -360,29 +353,6 @@ class TestSerializeByAlias:
assert set(value_schema["properties"]) == {"id"}
assert result.structured_content == {"first": {"id": "1"}}
async def test_model_in_list_respects_config(self):
"""A list's wrapped schema and result both use the model's field names."""
class Biofile(BaseModel):
model_config = ConfigDict(serialize_by_alias=False)
id: str = Field(alias="_id")
mcp = FastMCP()
@mcp.tool
def get_biofiles() -> list[Biofile]:
return [Biofile(_id="1")]
async with Client(mcp) as client:
tools = {tool.name: tool for tool in await client.list_tools()}
result = await client.call_tool("get_biofiles", {})
item_schema = tools["get_biofiles"].output_schema["properties"]["result"][ # type: ignore[index]
"items"
]
assert set(item_schema["properties"]) == {"id"}
assert result.structured_content == {"result": [{"id": "1"}]}
async def test_nested_models_use_their_own_alias_configs(self):
"""Nested models can independently enable and disable aliases."""
@ -419,75 +389,8 @@ class TestSerializeByAlias:
"aliased": {"aliasedValue": "aliased"},
}
async def test_dataclass_uses_pydantic_alias_default(self):
"""A dataclass schema uses field names when aliases are not enabled."""
@dataclass
class Output:
value: Annotated[str, Field(serialization_alias="dataValue")]
mcp = FastMCP()
@mcp.tool
def get_output() -> Output:
return Output(value="data")
async with Client(mcp) as client:
tools = {tool.name: tool for tool in await client.list_tools()}
result = await client.call_tool("get_output", {})
assert set(tools["get_output"].output_schema["properties"]) == {"value"} # type: ignore[index]
assert result.structured_content == {"value": "data"}
async def test_pydantic_dataclass_can_enable_aliases(self):
"""A Pydantic dataclass can opt in to serialization aliases."""
@pydantic_dataclass(config=ConfigDict(serialize_by_alias=True))
class Output:
value: Annotated[str, Field(serialization_alias="dataValue")]
mcp = FastMCP()
@mcp.tool
def get_output() -> Output:
return Output(value="data")
async with Client(mcp) as client:
tools = {tool.name: tool for tool in await client.list_tools()}
result = await client.call_tool("get_output", {})
assert set(tools["get_output"].output_schema["properties"]) == { # type: ignore[index]
"dataValue"
}
assert result.structured_content == {"dataValue": "data"}
assert json.loads(result.content[0].text) == {"dataValue": "data"} # type: ignore[union-attr]
async def test_configured_standard_dataclass_can_enable_aliases(self):
"""A configured stdlib dataclass uses aliases in schema and output."""
@with_config(ConfigDict(serialize_by_alias=True))
@dataclass
class Output:
value: Annotated[str, Field(serialization_alias="dataValue")]
mcp = FastMCP()
@mcp.tool
def get_output() -> Output:
return Output(value="data")
async with Client(mcp) as client:
tools = {tool.name: tool for tool in await client.list_tools()}
result = await client.call_tool("get_output", {})
assert set(tools["get_output"].output_schema["properties"]) == { # type: ignore[index]
"dataValue"
}
assert result.structured_content == {"dataValue": "data"}
assert json.loads(result.content[0].text) == {"dataValue": "data"} # type: ignore[union-attr]
async def test_nested_configured_standard_dataclass_can_enable_aliases(self):
"""Typed containers preserve a nested dataclass's alias configuration."""
async def test_typed_dataclass_container_uses_declared_adapter(self):
"""A typed container preserves its dataclass's alias configuration."""
@with_config(ConfigDict(serialize_by_alias=True))
@dataclass
@ -511,25 +414,6 @@ class TestSerializeByAlias:
assert result.structured_content == {"result": [{"dataValue": "data"}]}
assert json.loads(result.content[0].text) == [{"dataValue": "data"}] # type: ignore[union-attr]
async def test_typed_dict_uses_pydantic_alias_default(self):
"""A TypedDict schema uses the field names emitted by Pydantic."""
class Output(TypedDict):
value: Annotated[str, Field(serialization_alias="dataValue")]
mcp = FastMCP()
@mcp.tool
def get_output() -> Output:
return {"value": "data"}
async with Client(mcp) as client:
tools = {tool.name: tool for tool in await client.list_tools()}
result = await client.call_tool("get_output", {})
assert set(tools["get_output"].output_schema["properties"]) == {"value"} # type: ignore[index]
assert result.structured_content == {"value": "data"}
async def test_serialize_by_alias_true_uses_alias(self):
"""serialize_by_alias=True emits aliases."""
@ -549,80 +433,3 @@ class TestSerializeByAlias:
assert result.structured_content == {"_id": "123"}
assert set(tools["get_biofile"].output_schema["properties"]) == {"_id"} # type: ignore[index]
async def test_nested_models_respect_config(self):
"""serialize_by_alias=False propagates through nested models."""
class Inner(BaseModel):
model_config = ConfigDict(serialize_by_alias=False)
inner_id: str = Field(alias="_iid")
class Outer(BaseModel):
model_config = ConfigDict(serialize_by_alias=False)
id: str = Field(alias="_id")
inner: Inner
mcp = FastMCP()
@mcp.tool
def get_outer() -> Outer:
return Outer(_id="1", inner=Inner(_iid="2"))
async with Client(mcp) as client:
result = await client.call_tool("get_outer", {})
assert result.structured_content == {"id": "1", "inner": {"inner_id": "2"}}
async def test_annotated_optional_return_stays_consistent(self):
"""Annotated[Model, ...] | None resolves the model inside the union arm.
Regression: the union arm is a typing.Annotated object, so a naive
isinstance check skipped the model and the schema fell back to aliases
while the runtime serialized field names, breaking client validation.
"""
class Biofile(BaseModel):
model_config = ConfigDict(serialize_by_alias=False)
id: str = Field(alias="_id")
mcp = FastMCP()
@mcp.tool
def get_biofile() -> Annotated[Biofile, Field(description="x")] | None:
return Biofile(_id="1")
async with Client(mcp) as client:
tools = {t.name: t for t in await client.list_tools()}
# client-side validation of structured content against the schema
# raises if they disagree
result = await client.call_tool("get_biofile", {})
schema_props = set(tools["get_biofile"].output_schema["properties"]) # type: ignore[index]
assert schema_props == set(result.structured_content) # type: ignore[arg-type]
assert result.structured_content == {"result": {"id": "1"}}
@pytest.mark.parametrize("serialize_by_alias", [True, False, None])
async def test_schema_and_structured_content_agree(self, serialize_by_alias):
"""The output schema field names always match the structured content keys."""
if serialize_by_alias is None:
config = ConfigDict()
else:
config = ConfigDict(serialize_by_alias=serialize_by_alias)
class Model(BaseModel):
model_config = config
id: str = Field(alias="_id")
name: str
mcp = FastMCP()
@mcp.tool
def get_model() -> Model:
return Model(_id="1", name="x")
async with Client(mcp) as client:
tools = {t.name: t for t in await client.list_tools()}
result = await client.call_tool("get_model", {})
schema_props = set(tools["get_model"].output_schema["properties"]) # type: ignore[index]
assert schema_props == set(result.structured_content) # type: ignore[arg-type]

19
tests/tools/tool_transform/test_tool_transform.py
View file

@ -746,25 +746,6 @@ async def test_transform_fn_configured_dataclass_respects_serialize_by_alias():
assert json.loads(result.content[0].text) == [{"itemId": "42"}]
async def test_transform_fn_unsupported_return_annotation_falls_back_to_content():
"""An unsupported transform annotation does not become a runtime failure."""
class Unsupported:
pass
def base() -> None:
pass
async def transform() -> Unsupported:
return Unsupported()
transformed = Tool.from_tool(base, transform_fn=transform, output_schema=None)
result = await transformed.run({})
assert result.structured_content is None
assert len(result.content) == 1
class TestProxy:
@pytest.fixture
def mcp_server(self) -> FastMCP: