Harden typed output serialization

fastmcp_slim/fastmcp/tools/base.py

@@ -21,7 +21,14 @@ from mcp_types import (
     ToolExecution,
 )
 from mcp_types import Tool as MCPTool
-from pydantic import BaseModel, Field, PrivateAttr, TypeAdapter, model_validator
+from pydantic import (
+    BaseModel,
+    Field,
+    PrivateAttr,
+    PydanticSchemaGenerationError,
+    TypeAdapter,
+    model_validator,
+)
 from pydantic.json_schema import SkipJsonSchema
 
 from fastmcp.utilities.authorization import AuthCheck
@@ -68,7 +75,7 @@ def default_serializer(data: Any) -> str:
 
 
 def _serialize_to_jsonable(data: Any, annotation: Any = Any) -> Any:
-    """Serialize through Pydantic while preserving each model's configuration."""
+    """Serialize through Pydantic, falling back for unsupported annotations."""
     if (
         annotation is inspect.Signature.empty
         or annotation is None
@@ -78,7 +85,10 @@ def _serialize_to_jsonable(data: Any, annotation: Any = Any) -> Any:
     ):
         adapter = _JSONABLE_ADAPTER
     else:
-        adapter = get_cached_typeadapter(annotation)
+        try:
+            return get_cached_typeadapter(annotation).dump_python(data, mode="json")
+        except PydanticSchemaGenerationError:
+            adapter = _JSONABLE_ADAPTER
 
     return adapter.dump_python(data, mode="json")
 
@@ -389,13 +399,7 @@ class Tool(FastMCPComponent):
 
         # 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 (
-            isinstance(raw_value, ContentBlock | Audio | Image | File)
-            or (
-                isinstance(raw_value, list | tuple)
-                and any(isinstance(item, ContentBlock) for item in raw_value)
-            )
-        ):
+        if self.output_schema is None and _is_content_result(raw_value):
             return ToolResult(content=content)
 
         try:
@@ -403,6 +407,9 @@ class Tool(FastMCPComponent):
         except (pydantic_core.PydanticSerializationError, UnicodeDecodeError):
             return ToolResult(content=content)
 
+        if not _is_content_result(raw_value):
+            content = _convert_to_content(structured)
+
         if self.output_schema is None:
             # No schema - only use structured_content for dicts
             if isinstance(structured, dict):
@@ -585,4 +592,14 @@ 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"]

fastmcp_slim/fastmcp/tools/tool_transform.py

@@ -19,6 +19,7 @@ from fastmcp.tools.base import (
     Tool,
     ToolResult,
     _convert_to_content,
+    _is_content_result,
     _serialize_to_jsonable,
 )
 from fastmcp.tools.function_parsing import ParsedFunction
@@ -405,15 +406,21 @@ class TransformedTool(Tool):
             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": self._serialize_output(result)}
+                    structured_output = {"result": serialized_result}
                 else:
-                    structured_output = self._serialize_output(result)
+                    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:

tests/tools/tool/test_output_schema.py

@@ -231,6 +231,10 @@ 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_mixed_unserializable_return_annotation(self):
         class Unserializable:
             def __init__(self, data: Any):
@@ -242,6 +246,10 @@ 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,
     ):

tests/tools/tool/test_results.py

@@ -480,6 +480,7 @@ class TestSerializeByAlias:
             "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."""
@@ -503,6 +504,7 @@ class TestSerializeByAlias:
             "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."""
@@ -527,6 +529,7 @@ class TestSerializeByAlias:
         ]
         assert set(item_schema["properties"]) == {"dataValue"}
         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."""

tests/tools/tool_transform/test_tool_transform.py

@@ -1,5 +1,6 @@
 """Core tool transform functionality."""
 
+import json
 import re
 from dataclasses import dataclass
 from typing import Annotated, Any
@@ -742,6 +743,27 @@ async def test_transform_fn_configured_dataclass_respects_serialize_by_alias():
     result = await transformed.run({})
 
     assert result.structured_content == {"result": [{"itemId": "42"}]}
+    assert isinstance(result.content[0], TextContent)
+    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: