From adf9a65eed427f91cbe04ceec23f316cf03d1f2a Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Thu, 6 Aug 2026 11:44:01 -0400 Subject: [PATCH] Simplify typed output serialization --- fastmcp_slim/fastmcp/tools/base.py | 34 ++- fastmcp_slim/fastmcp/tools/function_tool.py | 6 - fastmcp_slim/fastmcp/tools/tool_transform.py | 40 +--- tests/tools/tool/test_output_schema.py | 4 - tests/tools/tool/test_results.py | 199 +----------------- .../tool_transform/test_tool_transform.py | 19 -- 6 files changed, 19 insertions(+), 283 deletions(-) diff --git a/fastmcp_slim/fastmcp/tools/base.py b/fastmcp_slim/fastmcp/tools/base.py index d4dfa8a81..9e11ad5c9 100644 --- a/fastmcp_slim/fastmcp/tools/base.py +++ b/fastmcp_slim/fastmcp/tools/base.py @@ -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"] diff --git a/fastmcp_slim/fastmcp/tools/function_tool.py b/fastmcp_slim/fastmcp/tools/function_tool.py index 82466b5d3..e5787f466 100644 --- a/fastmcp_slim/fastmcp/tools/function_tool.py +++ b/fastmcp_slim/fastmcp/tools/function_tool.py @@ -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, diff --git a/fastmcp_slim/fastmcp/tools/tool_transform.py b/fastmcp_slim/fastmcp/tools/tool_transform.py index 06ed382ee..436f3f2d3 100644 --- a/fastmcp_slim/fastmcp/tools/tool_transform.py +++ b/fastmcp_slim/fastmcp/tools/tool_transform.py @@ -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) diff --git a/tests/tools/tool/test_output_schema.py b/tests/tools/tool/test_output_schema.py index 6a769899f..c650134d9 100644 --- a/tests/tools/tool/test_output_schema.py +++ b/tests/tools/tool/test_output_schema.py @@ -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, ): diff --git a/tests/tools/tool/test_results.py b/tests/tools/tool/test_results.py index a68fcf384..9edcbb327 100644 --- a/tests/tools/tool/test_results.py +++ b/tests/tools/tool/test_results.py @@ -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] diff --git a/tests/tools/tool_transform/test_tool_transform.py b/tests/tools/tool_transform/test_tool_transform.py index 7f62f744e..3dc89239b 100644 --- a/tests/tools/tool_transform/test_tool_transform.py +++ b/tests/tools/tool_transform/test_tool_transform.py @@ -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: