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

Ensure transformed tools generate structured content (#1443)

Browse source
This commit is contained in:
Jeremiah Lowin 2025-08-11 12:34:43 -04:00 • committed by GitHub
commit 1df4771909
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 124 additions and 68 deletions
Download patch file Download diff file Expand all files Collapse all files

14
src/fastmcp/tools/tool.py
View file

@ -8,7 +8,6 @@ from typing import (
Annotated,
Any,
Generic,
Literal,
TypeVar,
get_type_hints,
)
@ -163,7 +162,7 @@ class Tool(FastMCPComponent):
tags: set[str] | None = None,
annotations: ToolAnnotations | None = None,
exclude_args: list[str] | None = None,
output_schema: dict[str, Any] | None | NotSetT | Literal[False] = NotSet,
output_schema: dict[str, Any] | None | NotSetT = NotSet,
serializer: Callable[[Any], str] | None = None,
meta: dict[str, Any] | None = None,
enabled: bool | None = None,
@ -199,17 +198,18 @@ class Tool(FastMCPComponent):
def from_tool(
cls,
tool: Tool,
transform_fn: Callable[..., Any] | None = None,
*,
name: str | None = None,
title: str | None | NotSetT = NotSet,
transform_args: dict[str, ArgTransform] | None = None,
description: str | None | NotSetT = NotSet,
tags: set[str] | None = None,
annotations: ToolAnnotations | None = None,
output_schema: dict[str, Any] | None | Literal[False] = None,
annotations: ToolAnnotations | None | NotSetT = NotSet,
output_schema: dict[str, Any] | None | NotSetT = NotSet,
serializer: Callable[[Any], str] | None = None,
meta: dict[str, Any] | None | NotSetT = NotSet,
transform_args: dict[str, ArgTransform] | None = None,
enabled: bool | None = None,
transform_fn: Callable[..., Any] | None = None,
) -> TransformedTool:
from fastmcp.tools.tool_transform import TransformedTool
@ -242,7 +242,7 @@ class FunctionTool(Tool):
tags: set[str] | None = None,
annotations: ToolAnnotations | None = None,
exclude_args: list[str] | None = None,
output_schema: dict[str, Any] | None | NotSetT | Literal[False] = NotSet,
output_schema: dict[str, Any] | None | NotSetT = NotSet,
serializer: Callable[[Any], str] | None = None,
meta: dict[str, Any] | None = None,
enabled: bool | None = None,

68
src/fastmcp/tools/tool_transform.py
View file

@ -6,6 +6,7 @@ from contextvars import ContextVar
from dataclasses import dataclass
from typing import Annotated, Any, Literal
import pydantic_core
from mcp.types import ToolAnnotations
from pydantic import ConfigDict
from pydantic.fields import Field
@ -312,11 +313,9 @@ class TransformedTool(Tool):
# Custom function returns ToolResult - preserve its content
return result
else:
# Forwarded call with disabled schema - strip structured content
return ToolResult(
content=result.content,
structured_content=None,
)
# Forwarded call with no explicit schema - preserve parent's structured content
# The parent tool may have generated structured content via its own fallback logic
return result
elif self.output_schema.get(
"type"
) != "object" and not self.output_schema.get("x-fastmcp-wrap-result"):
@ -334,17 +333,23 @@ class TransformedTool(Tool):
result, serializer=self.serializer
)
# Handle structured content based on output schema
structured_output = None
# First handle structured content based on output schema, if any
if self.output_schema is not None:
if self.output_schema.get("x-fastmcp-wrap-result"):
# Schema says wrap - always wrap in result key
structured_output = {"result": result}
else:
# Object schemas - use result directly
# User is responsible for returning dict-compatible data
structured_output = result
else:
structured_output = None
# 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 = pydantic_core.to_jsonable_python(result)
if not isinstance(structured_output, dict):
structured_output = None
except Exception:
pass
return ToolResult(
content=unstructured_result,
@ -363,9 +368,9 @@ class TransformedTool(Tool):
tags: set[str] | None = None,
transform_fn: Callable[..., Any] | None = None,
transform_args: dict[str, ArgTransform] | None = None,
annotations: ToolAnnotations | None = None,
output_schema: dict[str, Any] | None | Literal[False] = None,
serializer: Callable[[Any], str] | None = None,
annotations: ToolAnnotations | None | NotSetT = NotSet,
output_schema: dict[str, Any] | None | NotSetT = NotSet,
serializer: Callable[[Any], str] | None | NotSetT = NotSet,
meta: dict[str, Any] | None | NotSetT = NotSet,
enabled: bool | None = None,
) -> TransformedTool:
@ -445,6 +450,11 @@ class TransformedTool(Tool):
"""
transform_args = transform_args or {}
if transform_fn is not None:
parsed_fn = ParsedFunction.from_function(transform_fn, validate=False)
else:
parsed_fn = None
# Validate transform_args
parent_params = set(tool.parameters.get("properties", {}).keys())
unknown_args = set(transform_args.keys()) - parent_params
@ -457,16 +467,11 @@ class TransformedTool(Tool):
# Always create the forwarding transform
schema, forwarding_fn = cls._create_forwarding_transform(tool, transform_args)
# Handle output schema with smart fallback
if output_schema is False:
final_output_schema = None
elif output_schema is not None:
# Explicit schema provided - use as-is
final_output_schema = output_schema
else:
# Smart fallback: try custom function, then parent, then None
# Handle output schema
if output_schema is NotSet:
# Use smart fallback: try custom function, then parent
if transform_fn is not None:
parsed_fn = ParsedFunction.from_function(transform_fn, validate=False)
assert parsed_fn is not None
final_output_schema = parsed_fn.output_schema
if final_output_schema is None:
# Check if function returns ToolResult - if so, don't fall back to parent
@ -479,15 +484,17 @@ class TransformedTool(Tool):
final_output_schema = tool.output_schema
else:
final_output_schema = tool.output_schema
else:
assert isinstance(output_schema, dict | None)
final_output_schema = output_schema
if transform_fn is None:
# User wants pure transformation - use forwarding_fn as the main function
final_fn = forwarding_fn
final_schema = schema
else:
assert parsed_fn is not None
# User provided custom function - merge schemas
if "parsed_fn" not in locals():
parsed_fn = ParsedFunction.from_function(transform_fn, validate=False)
final_fn = transform_fn
has_kwargs = cls._function_has_kwargs(transform_fn)
@ -552,6 +559,13 @@ class TransformedTool(Tool):
)
final_title = title if not isinstance(title, NotSetT) else tool.title
final_meta = meta if not isinstance(meta, NotSetT) else tool.meta
final_annotations = (
annotations if not isinstance(annotations, NotSetT) else tool.annotations
)
final_serializer = (
serializer if not isinstance(serializer, NotSetT) else tool.serializer
)
final_enabled = enabled if enabled is not None else tool.enabled
transformed_tool = cls(
fn=final_fn,
@ -563,11 +577,11 @@ class TransformedTool(Tool):
parameters=final_schema,
output_schema=final_output_schema,
tags=tags or tool.tags,
annotations=annotations or tool.annotations,
serializer=serializer or tool.serializer,
annotations=final_annotations,
serializer=final_serializer,
meta=final_meta,
transform_args=transform_args,
enabled=enabled if enabled is not None else True,
enabled=final_enabled,
)
return transformed_tool

28
tests/server/test_tool_transformation.py
View file

@ -38,3 +38,31 @@ async def test_transformed_tool_filtering():
tools = list(await mcp._list_tools())
assert len(tools) == 1
async def test_transformed_tool_structured_output_without_annotation():
"""Test that transformed tools generate structured output when original tool has no return annotation.
Ref: https://github.com/jlowin/fastmcp/issues/1369
"""
from fastmcp.client import Client
mcp = FastMCP("Test Server")
@mcp.tool()
def tool_without_annotation(message: str): # No return annotation
"""A tool without return type annotation."""
return {"result": "processed", "input": message}
# Create a transformed tool
mcp.add_tool_transformation(
"tool_without_annotation", ToolTransformConfig(name="transformed_tool")
)
# Test with client to verify structured output is populated
async with Client(mcp) as client:
result = await client.call_tool("transformed_tool", {"message": "test"})
# Structured output should be populated even without return annotation
assert result.data is not None
assert result.data == {"result": "processed", "input": "test"}

30
tests/tools/test_tool.py
View file

@ -562,7 +562,7 @@ class TestToolFromFunctionOutputSchema:
def func() -> dict[str, str]:
return {"message": "Hello, world!"}
tool = Tool.from_function(func, output_schema=False)
tool = Tool.from_function(func, output_schema=None)
assert tool.output_schema is None
result = await tool.run({})
@ -729,17 +729,23 @@ class TestToolFromFunctionOutputSchema:
tool_none = Tool.from_function(func, output_schema=None)
assert tool_none.output_schema is None
# False should also disable
tool_false = Tool.from_function(func, output_schema=False)
assert tool_false.output_schema is None
# Default (NotSet) should infer from return type
tool_default = Tool.from_function(func)
assert (
tool_default.output_schema is not None
) # Should infer schema from dict return type
# Both should have same behavior
# Different behavior: None vs inferred
result_none = await tool_none.run({})
result_false = await tool_false.run({})
result_default = await tool_default.run({})
assert result_none.structured_content is None
assert result_false.structured_content is None
assert result_none.content[0].text == result_false.content[0].text == "123" # type: ignore[attr-defined]
# None should still try fallback generation but fail for non-dict
assert result_none.structured_content is None # Fallback fails for int
# Default should use proper schema and wrap the result
assert result_default.structured_content == {
"result": 123
} # Schema-based generation with wrapping
assert result_none.content[0].text == result_default.content[0].text == "123" # type: ignore[attr-defined]
async def test_non_object_output_schema_raises_error(self):
"""Test that providing a non-object output schema raises a ValueError."""
@ -1117,7 +1123,7 @@ class TestAutomaticStructuredContent:
return UserProfile(name="Bob", age=25, email="bob@example.com")
# No explicit output schema, but dataclass should still create structured content
tool = Tool.from_function(get_profile, output_schema=False)
tool = Tool.from_function(get_profile, output_schema=None)
result = await tool.run({"user_id": "456"})
@ -1144,8 +1150,8 @@ class TestAutomaticStructuredContent:
def get_user_stats(user_id: str) -> UserData:
return UserData(username="charlie", score=100, verified=True)
# Explicitly disable output schema to test automatic structured content
tool = Tool.from_function(get_user_stats, output_schema=False)
# Explicitly set output schema to None to test automatic structured content
tool = Tool.from_function(get_user_stats, output_schema=None)
result = await tool.run({"user_id": "789"})

52
tests/tools/test_tool_transform.py
View file

@ -1020,7 +1020,12 @@ class TestEnableDisable:
new_add = Tool.from_tool(add, name="new_add")
mcp.add_tool(new_add)
assert new_add.enabled
# the new tool inherits the disabled state from the parent tool
assert new_add.enabled is False
new_add.enable()
assert new_add.enabled is True
assert add.enabled is False
async with Client(mcp) as client:
tools = await client.list_tools()
@ -1118,15 +1123,15 @@ class TestTransformToolOutputSchema:
assert new_tool.output_schema == expected_schema
assert new_tool.output_schema == base_string_tool.output_schema
def test_transform_with_explicit_output_schema_false(self, base_string_tool):
"""Test that output_schema=False disables structured output."""
new_tool = Tool.from_tool(base_string_tool, output_schema=False)
def test_transform_with_explicit_output_schema_none(self, base_string_tool):
"""Test that output_schema=None sets output schema to None."""
new_tool = Tool.from_tool(base_string_tool, output_schema=None)
assert new_tool.output_schema is None
async def test_transform_output_schema_false_runtime(self, base_string_tool):
"""Test runtime behavior with output_schema=False."""
new_tool = Tool.from_tool(base_string_tool, output_schema=False)
async def test_transform_output_schema_none_runtime(self, base_string_tool):
"""Test runtime behavior with output_schema=None."""
new_tool = Tool.from_tool(base_string_tool, output_schema=None)
# Debug: check that output_schema is actually None
assert new_tool.output_schema is None, (
@ -1134,7 +1139,8 @@ class TestTransformToolOutputSchema:
)
result = await new_tool.run({"x": 5})
assert result.structured_content is None
# Even with output_schema=None, structured content should be generated via fallback logic
assert result.structured_content == {"result": "Result: 5"}
assert result.content[0].text == "Result: 5" # type: ignore[attr-defined]
def test_transform_with_explicit_output_schema_dict(self, base_string_tool):
@ -1306,25 +1312,27 @@ class TestTransformToolOutputSchema:
expected_schema = TypeAdapter(dict[str, str]).json_schema()
assert new_tool.output_schema == expected_schema
async def test_transform_output_schema_none_vs_false(self, base_string_tool):
"""Test None vs False behavior for output_schema in transforms."""
# None (default) should use smart fallback (inherit from parent)
tool_none = Tool.from_tool(base_string_tool) # default output_schema=None
assert tool_none.output_schema == base_string_tool.output_schema # Inherits
async def test_transform_output_schema_default_vs_none(self, base_string_tool):
"""Test default (NotSet) vs explicit None behavior for output_schema in transforms."""
# Default (NotSet) should use smart fallback (inherit from parent)
tool_default = Tool.from_tool(base_string_tool) # default output_schema=NotSet
assert tool_default.output_schema == base_string_tool.output_schema # Inherits
# False should explicitly disable
tool_false = Tool.from_tool(base_string_tool, output_schema=False)
assert tool_false.output_schema is None
# None should explicitly set output_schema to None but still generate structured content via fallback
tool_explicit_none = Tool.from_tool(base_string_tool, output_schema=None)
assert tool_explicit_none.output_schema is None
# Different behavior at runtime
result_none = await tool_none.run({"x": 5})
result_false = await tool_false.run({"x": 5})
# Both should generate structured content now (via different paths)
result_default = await tool_default.run({"x": 5})
result_explicit_none = await tool_explicit_none.run({"x": 5})
assert result_none.structured_content == {
assert result_default.structured_content == {
"result": "Result: 5"
} # Inherits wrapping
assert result_false.structured_content is None # Disabled
assert result_none.content[0].text == result_false.content[0].text # type: ignore[attr-defined]
assert result_explicit_none.structured_content == {
"result": "Result: 5"
} # Generated via fallback logic
assert result_default.content[0].text == result_explicit_none.content[0].text # type: ignore[attr-defined]
async def test_transform_output_schema_with_tool_result_return(
self, base_string_tool