from dataclasses import dataclass from datetime import timedelta from typing import Annotated, Any import pytest from dirty_equals import HasName from inline_snapshot import snapshot from mcp.types import ( AudioContent, BlobResourceContents, EmbeddedResource, ImageContent, ResourceLink, TextContent, TextResourceContents, ToolExecution, ) from pydantic import AnyUrl, BaseModel, Field, TypeAdapter from typing_extensions import TypedDict from fastmcp.tools.tool import Tool, ToolResult, _convert_to_content from fastmcp.utilities.json_schema import compress_schema from fastmcp.utilities.types import Audio, File, Image class TestToolFromFunction: def test_basic_function(self): """Test registering and running a basic function.""" def add(a: int, b: int) -> int: """Add two numbers.""" return a + b tool = Tool.from_function(add) assert tool.model_dump(exclude_none=True) == snapshot( { "name": "add", "description": "Add two numbers.", "tags": set(), "parameters": { "properties": { "a": {"type": "integer"}, "b": {"type": "integer"}, }, "required": ["a", "b"], "type": "object", }, "output_schema": { "properties": {"result": {"type": "integer"}}, "required": ["result"], "type": "object", "x-fastmcp-wrap-result": True, }, "fn": HasName("add"), "task_config": { "mode": "forbidden", "poll_interval": timedelta(seconds=5), }, } ) def test_meta_parameter(self): """Test that meta parameter is properly handled.""" def multiply(a: int, b: int) -> int: """Multiply two numbers.""" return a * b meta_data = {"version": "1.0", "author": "test"} tool = Tool.from_function(multiply, meta=meta_data) assert tool.meta == meta_data mcp_tool = tool.to_mcp_tool() # MCP tool includes fastmcp meta, so check that our meta is included assert mcp_tool.meta is not None assert meta_data.items() <= mcp_tool.meta.items() async def test_async_function(self): """Test registering and running an async function.""" async def fetch_data(url: str) -> str: """Fetch data from URL.""" return f"Data from {url}" tool = Tool.from_function(fetch_data) assert tool.model_dump(exclude_none=True) == snapshot( { "name": "fetch_data", "description": "Fetch data from URL.", "tags": set(), "parameters": { "properties": {"url": {"type": "string"}}, "required": ["url"], "type": "object", }, "output_schema": { "properties": {"result": {"type": "string"}}, "required": ["result"], "type": "object", "x-fastmcp-wrap-result": True, }, "fn": HasName("fetch_data"), "task_config": { "mode": "forbidden", "poll_interval": timedelta(seconds=5), }, } ) def test_callable_object(self): class Adder: """Adds two numbers.""" def __call__(self, x: int, y: int) -> int: """ignore this""" return x + y tool = Tool.from_function(Adder()) assert tool.model_dump(exclude_none=True, exclude={"fn"}) == snapshot( { "name": "Adder", "description": "Adds two numbers.", "tags": set(), "parameters": { "properties": { "x": {"type": "integer"}, "y": {"type": "integer"}, }, "required": ["x", "y"], "type": "object", }, "output_schema": { "properties": {"result": {"type": "integer"}}, "required": ["result"], "type": "object", "x-fastmcp-wrap-result": True, }, "task_config": { "mode": "forbidden", "poll_interval": timedelta(seconds=5), }, } ) def test_async_callable_object(self): class Adder: """Adds two numbers.""" async def __call__(self, x: int, y: int) -> int: """ignore this""" return x + y tool = Tool.from_function(Adder()) assert tool.model_dump(exclude_none=True, exclude={"fn"}) == snapshot( { "name": "Adder", "description": "Adds two numbers.", "tags": set(), "parameters": { "properties": { "x": {"type": "integer"}, "y": {"type": "integer"}, }, "required": ["x", "y"], "type": "object", }, "output_schema": { "properties": {"result": {"type": "integer"}}, "required": ["result"], "type": "object", "x-fastmcp-wrap-result": True, }, "task_config": { "mode": "forbidden", "poll_interval": timedelta(seconds=5), }, } ) def test_pydantic_model_function(self): """Test registering a function that takes a Pydantic model.""" class UserInput(BaseModel): name: str age: int def create_user(user: UserInput, flag: bool) -> dict: """Create a new user.""" return {"id": 1, **user.model_dump()} tool = Tool.from_function(create_user) assert tool.model_dump(exclude_none=True) == snapshot( { "name": "create_user", "description": "Create a new user.", "tags": set(), "parameters": { "properties": { "user": { "properties": { "name": {"type": "string"}, "age": {"type": "integer"}, }, "required": ["name", "age"], "type": "object", }, "flag": {"type": "boolean"}, }, "required": ["user", "flag"], "type": "object", }, "output_schema": {"additionalProperties": True, "type": "object"}, "fn": HasName("create_user"), "task_config": { "mode": "forbidden", "poll_interval": timedelta(seconds=5), }, } ) async def test_tool_with_image_return(self): def image_tool(data: bytes) -> Image: return Image(data=data) tool = Tool.from_function(image_tool) assert tool.parameters["properties"]["data"]["type"] == "string" assert tool.output_schema is None result = await tool.run({"data": "test.png"}) assert isinstance(result.content[0], ImageContent) async def test_tool_with_audio_return(self): def audio_tool(data: bytes) -> Audio: return Audio(data=data) tool = Tool.from_function(audio_tool) assert tool.parameters["properties"]["data"]["type"] == "string" assert tool.output_schema is None result = await tool.run({"data": "test.wav"}) assert isinstance(result.content[0], AudioContent) async def test_tool_with_file_return(self): def file_tool(data: bytes) -> File: return File(data=data, format="octet-stream") tool = Tool.from_function(file_tool) assert tool.parameters["properties"]["data"]["type"] == "string" assert tool.output_schema is None result: ToolResult = await tool.run({"data": "test.bin"}) assert result.content[0].model_dump(exclude_none=True) == snapshot( { "type": "resource", "resource": { "uri": AnyUrl("file:///resource.octet-stream"), "mimeType": "application/octet-stream", "blob": "dGVzdC5iaW4=", }, } ) def test_non_callable_fn(self): with pytest.raises(TypeError, match="not a callable object"): Tool.from_function(1) # type: ignore def test_lambda(self): tool = Tool.from_function(lambda x: x, name="my_tool") assert tool.model_dump(exclude_none=True, exclude={"fn"}) == snapshot( { "name": "my_tool", "tags": set(), "parameters": { "properties": {"x": {"title": "X"}}, "required": ["x"], "type": "object", }, "task_config": { "mode": "forbidden", "poll_interval": timedelta(seconds=5), }, } ) def test_lambda_with_no_name(self): with pytest.raises( ValueError, match="You must provide a name for lambda functions" ): Tool.from_function(lambda x: x) def test_private_arguments(self): def add(_a: int, _b: int) -> int: """Add two numbers.""" return _a + _b tool = Tool.from_function(add) assert tool.model_dump( exclude_none=True, exclude={"output_schema", "fn"} ) == snapshot( { "name": "add", "description": "Add two numbers.", "tags": set(), "parameters": { "properties": { "_a": {"type": "integer"}, "_b": {"type": "integer"}, }, "required": ["_a", "_b"], "type": "object", }, "task_config": { "mode": "forbidden", "poll_interval": timedelta(seconds=5), }, } ) def test_tool_with_varargs_not_allowed(self): def func(a: int, b: int, *args: int) -> int: """Add two numbers.""" return a + b with pytest.raises( ValueError, match=r"Functions with \*args are not supported as tools" ): Tool.from_function(func) def test_tool_with_varkwargs_not_allowed(self): def func(a: int, b: int, **kwargs: int) -> int: """Add two numbers.""" return a + b with pytest.raises( ValueError, match=r"Functions with \*\*kwargs are not supported as tools" ): Tool.from_function(func) async def test_instance_method(self): class MyClass: def add(self, x: int, y: int) -> int: """Add two numbers.""" return x + y obj = MyClass() tool = Tool.from_function(obj.add) assert "self" not in tool.parameters["properties"] assert tool.model_dump(exclude_none=True, exclude={"fn"}) == snapshot( { "name": "add", "description": "Add two numbers.", "tags": set(), "parameters": { "properties": { "x": {"type": "integer"}, "y": {"type": "integer"}, }, "required": ["x", "y"], "type": "object", }, "output_schema": { "properties": {"result": {"type": "integer"}}, "required": ["result"], "type": "object", "x-fastmcp-wrap-result": True, }, "task_config": { "mode": "forbidden", "poll_interval": timedelta(seconds=5), }, } ) async def test_instance_method_with_varargs_not_allowed(self): class MyClass: def add(self, x: int, y: int, *args: int) -> int: """Add two numbers.""" return x + y obj = MyClass() with pytest.raises( ValueError, match=r"Functions with \*args are not supported as tools" ): Tool.from_function(obj.add) async def test_instance_method_with_varkwargs_not_allowed(self): class MyClass: def add(self, x: int, y: int, **kwargs: int) -> int: """Add two numbers.""" return x + y obj = MyClass() with pytest.raises( ValueError, match=r"Functions with \*\*kwargs are not supported as tools" ): Tool.from_function(obj.add) async def test_classmethod(self): class MyClass: x: int = 10 @classmethod def call(cls, x: int, y: int) -> int: """Add two numbers.""" return x + y tool = Tool.from_function(MyClass.call) assert tool.name == "call" assert tool.description == "Add two numbers." assert "x" in tool.parameters["properties"] assert "y" in tool.parameters["properties"] class TestToolFromFunctionOutputSchema: async def test_no_return_annotation(self): def func(): pass tool = Tool.from_function(func) assert tool.output_schema is None @pytest.mark.parametrize( "annotation", [ int, float, bool, str, int | float, list, list[int], list[int | float], dict, dict[str, Any], dict[str, int | None], tuple[int, str], set[int], list[tuple[int, str]], ], ) async def test_simple_return_annotation(self, annotation): def func() -> annotation: return 1 tool = Tool.from_function(func) base_schema = TypeAdapter(annotation).json_schema() # Non-object types get wrapped schema_type = base_schema.get("type") is_object_type = schema_type == "object" if not is_object_type: # Non-object types get wrapped expected_schema = { "type": "object", "properties": {"result": base_schema}, "required": ["result"], "x-fastmcp-wrap-result": True, } assert tool.output_schema == expected_schema # # Note: Parameterized test - keeping original assertion for multiple parameter values else: # Object types remain unwrapped assert tool.output_schema == base_schema @pytest.mark.parametrize( "annotation", [ AnyUrl, Annotated[int, Field(ge=1)], Annotated[int, Field(ge=1)], ], ) async def test_complex_return_annotation(self, annotation): def func() -> annotation: return 1 tool = Tool.from_function(func) base_schema = TypeAdapter(annotation).json_schema() expected_schema = { "type": "object", "properties": {"result": base_schema}, "required": ["result"], "x-fastmcp-wrap-result": True, } assert tool.output_schema == expected_schema async def test_none_return_annotation(self): def func() -> None: pass tool = Tool.from_function(func) assert tool.output_schema is None async def test_any_return_annotation(self): def func() -> Any: return 1 tool = Tool.from_function(func) assert tool.output_schema is None @pytest.mark.parametrize( "annotation, expected", [ (Image, ImageContent), (Audio, AudioContent), (File, EmbeddedResource), (Image | int, ImageContent | int), (Image | Audio, ImageContent | AudioContent), (list[Image | Audio], list[ImageContent | AudioContent]), ], ) async def test_converted_return_annotation(self, annotation, expected): def func() -> annotation: return 1 tool = Tool.from_function(func) # Image, Audio, File types don't generate output schemas since they're converted to content directly assert tool.output_schema is None async def test_dataclass_return_annotation(self): @dataclass class Person: name: str age: int def func() -> Person: return Person(name="John", age=30) tool = Tool.from_function(func) expected_schema = compress_schema( TypeAdapter(Person).json_schema(), prune_titles=True ) assert tool.output_schema == expected_schema async def test_base_model_return_annotation(self): class Person(BaseModel): name: str age: int def func() -> Person: return Person(name="John", age=30) tool = Tool.from_function(func) assert tool.output_schema == snapshot( { "properties": { "name": {"type": "string"}, "age": {"type": "integer"}, }, "required": ["name", "age"], "type": "object", } ) async def test_typeddict_return_annotation(self): class Person(TypedDict): name: str age: int def func() -> Person: return Person(name="John", age=30) tool = Tool.from_function(func) assert tool.output_schema == snapshot( { "properties": { "name": {"type": "string"}, "age": {"type": "integer"}, }, "required": ["name", "age"], "type": "object", } ) async def test_unserializable_return_annotation(self): class Unserializable: def __init__(self, data: Any): self.data = data def func() -> Unserializable: return Unserializable(data="test") tool = Tool.from_function(func) assert tool.output_schema is None async def test_mixed_unserializable_return_annotation(self): class Unserializable: def __init__(self, data: Any): self.data = data def func() -> Unserializable | int: return Unserializable(data="test") tool = Tool.from_function(func) assert tool.output_schema is None async def test_provided_output_schema_takes_precedence_over_json_compatible_annotation( self, ): """Test that provided output_schema takes precedence over inferred schema from JSON-compatible annotation.""" def func() -> dict[str, int]: return {"a": 1, "b": 2} # Provide a custom output schema that differs from the inferred one custom_schema = {"type": "object", "description": "Custom schema"} tool = Tool.from_function(func, output_schema=custom_schema) assert tool.output_schema == custom_schema async def test_provided_output_schema_takes_precedence_over_complex_annotation( self, ): """Test that provided output_schema takes precedence over inferred schema from complex annotation.""" def func() -> list[dict[str, int | float]]: return [{"a": 1, "b": 2.5}] # Provide a custom output schema that differs from the inferred one custom_schema = {"type": "object", "properties": {"custom": {"type": "string"}}} tool = Tool.from_function(func, output_schema=custom_schema) assert tool.output_schema == custom_schema async def test_provided_output_schema_takes_precedence_over_unserializable_annotation( self, ): """Test that provided output_schema takes precedence over None schema from unserializable annotation.""" class Unserializable: def __init__(self, data: Any): self.data = data def func() -> Unserializable: return Unserializable(data="test") # Provide a custom output schema even though the annotation is unserializable custom_schema = { "type": "object", "properties": {"items": {"type": "array", "items": {"type": "string"}}}, } tool = Tool.from_function(func, output_schema=custom_schema) assert tool.output_schema == custom_schema async def test_provided_output_schema_takes_precedence_over_no_annotation(self): """Test that provided output_schema takes precedence over None schema from no annotation.""" def func(): return "hello" # Provide a custom output schema even though there's no return annotation custom_schema = { "type": "object", "properties": {"value": {"type": "number", "minimum": 0}}, } tool = Tool.from_function(func, output_schema=custom_schema) assert tool.output_schema == custom_schema async def test_provided_output_schema_takes_precedence_over_converted_annotation( self, ): """Test that provided output_schema takes precedence over converted schema from Image/Audio/File annotations.""" def func() -> Image: return Image(data=b"test") # Provide a custom output schema that differs from the converted ImageContent schema custom_schema = { "type": "object", "properties": {"custom_image": {"type": "string"}}, } tool = Tool.from_function(func, output_schema=custom_schema) assert tool.output_schema == custom_schema async def test_provided_output_schema_takes_precedence_over_union_annotation(self): """Test that provided output_schema takes precedence over inferred schema from union annotation.""" def func() -> str | int | None: return "hello" # Provide a custom output schema that differs from the inferred union schema custom_schema = {"type": "object", "properties": {"flag": {"type": "boolean"}}} tool = Tool.from_function(func, output_schema=custom_schema) assert tool.output_schema == custom_schema async def test_provided_output_schema_takes_precedence_over_pydantic_annotation( self, ): """Test that provided output_schema takes precedence over inferred schema from Pydantic model annotation.""" class Person(BaseModel): name: str age: int def func() -> Person: return Person(name="John", age=30) # Provide a custom output schema that differs from the inferred Person schema custom_schema = { "type": "object", "properties": {"numbers": {"type": "array", "items": {"type": "number"}}}, } tool = Tool.from_function(func, output_schema=custom_schema) assert tool.output_schema == custom_schema async def test_output_schema_false_allows_automatic_structured_content(self): """Test that output_schema=False still allows automatic structured content for dict-like objects.""" def func() -> dict[str, str]: return {"message": "Hello, world!"} tool = Tool.from_function(func, output_schema=None) assert tool.output_schema is None result = await tool.run({}) # Dict objects automatically become structured content even without schema assert result.structured_content == {"message": "Hello, world!"} assert len(result.content) == 1 assert isinstance(result.content[0], TextContent) assert result.content[0].text == '{"message":"Hello, world!"}' async def test_output_schema_none_disables_structured_content(self): """Test that output_schema=None explicitly disables structured content.""" def func() -> int: return 42 tool = Tool.from_function(func, output_schema=None) assert tool.output_schema is None result = await tool.run({}) assert result.structured_content is None assert len(result.content) == 1 assert isinstance(result.content[0], TextContent) assert result.content[0].text == "42" async def test_output_schema_inferred_when_not_specified(self): """Test that output schema is inferred when not explicitly specified.""" def func() -> int: return 42 # Don't specify output_schema - should infer and wrap tool = Tool.from_function(func) assert tool.output_schema == snapshot( { "properties": {"result": {"type": "integer"}}, "required": ["result"], "type": "object", "x-fastmcp-wrap-result": True, } ) result = await tool.run({}) assert result.structured_content == {"result": 42} async def test_explicit_object_schema_with_dict_return(self): """Test that explicit object schemas work when function returns a dict.""" def func() -> dict[str, int]: return {"value": 42} # Provide explicit object schema explicit_schema = { "type": "object", "properties": {"value": {"type": "integer", "minimum": 0}}, } tool = Tool.from_function(func, output_schema=explicit_schema) assert tool.output_schema == explicit_schema # Schema not wrapped assert tool.output_schema and "x-fastmcp-wrap-result" not in tool.output_schema result = await tool.run({}) # Dict result with object schema is used directly assert result.structured_content == {"value": 42} assert isinstance(result.content[0], TextContent) assert result.content[0].text == '{"value":42}' async def test_explicit_object_schema_with_non_dict_return_fails(self): """Test that explicit object schemas fail when function returns non-dict.""" def func() -> int: return 42 # Provide explicit object schema but return non-dict explicit_schema = { "type": "object", "properties": {"value": {"type": "integer"}}, } tool = Tool.from_function(func, output_schema=explicit_schema) # Should fail because int is not dict-compatible with object schema with pytest.raises(ValueError, match="structured_content must be a dict"): await tool.run({}) async def test_object_output_schema_not_wrapped(self): """Test that object-type output schemas are never wrapped.""" def func() -> dict[str, int]: return {"value": 42} # Object schemas should never be wrapped, even when inferred tool = Tool.from_function(func) expected_schema = TypeAdapter(dict[str, int]).json_schema() assert tool.output_schema == expected_schema # Not wrapped assert tool.output_schema and "x-fastmcp-wrap-result" not in tool.output_schema result = await tool.run({}) assert result.structured_content == {"value": 42} # Direct value async def test_structured_content_interaction_with_wrapping(self): """Test that structured content works correctly with schema wrapping.""" def func() -> str: return "hello" # Inferred schema should wrap string type tool = Tool.from_function(func) assert tool.output_schema == snapshot( { "properties": {"result": {"type": "string"}}, "required": ["result"], "type": "object", "x-fastmcp-wrap-result": True, } ) result = await tool.run({}) # Unstructured content assert len(result.content) == 1 assert isinstance(result.content[0], TextContent) assert result.content[0].text == "hello" # Structured content should be wrapped assert result.structured_content == {"result": "hello"} async def test_structured_content_with_explicit_object_schema(self): """Test structured content with explicit object schema.""" def func() -> dict[str, str]: return {"greeting": "hello"} # Provide explicit object schema explicit_schema = { "type": "object", "properties": {"greeting": {"type": "string"}}, "required": ["greeting"], } tool = Tool.from_function(func, output_schema=explicit_schema) assert tool.output_schema == explicit_schema result = await tool.run({}) # Should use direct value since explicit schema doesn't have wrap marker assert result.structured_content == {"greeting": "hello"} async def test_structured_content_with_custom_wrapper_schema(self): """Test structured content with custom schema that includes wrap marker.""" def func() -> str: return "world" # Custom schema with wrap marker custom_schema = { "type": "object", "properties": {"message": {"type": "string"}}, "x-fastmcp-wrap-result": True, } tool = Tool.from_function(func, output_schema=custom_schema) assert tool.output_schema == custom_schema result = await tool.run({}) # Should wrap with "result" key due to wrap marker assert result.structured_content == {"result": "world"} async def test_none_vs_false_output_schema_behavior(self): """Test the difference between None and False for output_schema.""" def func() -> int: return 123 # None should disable tool_none = Tool.from_function(func, output_schema=None) assert tool_none.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 # Different behavior: None vs inferred result_none = await tool_none.run({}) result_default = await tool_default.run({}) # 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 isinstance(result_none.content[0], TextContent) assert isinstance(result_default.content[0], TextContent) assert result_none.content[0].text == result_default.content[0].text == "123" async def test_non_object_output_schema_raises_error(self): """Test that providing a non-object output schema raises a ValueError.""" def func() -> int: return 42 # Test various non-object schemas that should raise errors non_object_schemas = [ {"type": "string"}, {"type": "integer", "minimum": 0}, {"type": "number"}, {"type": "boolean"}, {"type": "array", "items": {"type": "string"}}, ] for schema in non_object_schemas: with pytest.raises( ValueError, match="Output schemas must represent object types" ): Tool.from_function(func, output_schema=schema) class SampleModel(BaseModel): x: int y: str class TestConvertResultToContent: """Tests for the _convert_to_content helper function.""" @pytest.mark.parametrize( argnames=("result", "expected"), argvalues=[ (True, "true"), ("hello", "hello"), (123, "123"), (123.45, "123.45"), ({"key": "value"}, '{"key":"value"}'), ( SampleModel(x=1, y="hello"), '{"x":1,"y":"hello"}', ), ], ids=[ "boolean", "string", "integer", "float", "object", "basemodel", ], ) def test_convert_singular(self, result, expected): """Test that a single item is converted to a TextContent.""" converted = _convert_to_content(result) assert converted == [TextContent(type="text", text=expected)] @pytest.mark.parametrize( argnames=("result", "expected_text"), argvalues=[ ([None], "[null]"), ([None, None], "[null,null]"), ([True], "[true]"), ([True, False], "[true,false]"), (["hello"], '["hello"]'), (["hello", "world"], '["hello","world"]'), ([123], "[123]"), ([123, 456], "[123,456]"), ([123.45], "[123.45]"), ([123.45, 456.78], "[123.45,456.78]"), ([{"key": "value"}], '[{"key":"value"}]'), ( [{"key": "value"}, {"key2": "value2"}], '[{"key":"value"},{"key2":"value2"}]', ), ([SampleModel(x=1, y="hello")], '[{"x":1,"y":"hello"}]'), ( [SampleModel(x=1, y="hello"), SampleModel(x=2, y="world")], '[{"x":1,"y":"hello"},{"x":2,"y":"world"}]', ), ([1, "two", None, {"c": 3}, False], '[1,"two",null,{"c":3},false]'), ], ids=[ "none", "none_many", "boolean", "boolean_many", "string", "string_many", "integer", "integer_many", "float", "float_many", "object", "object_many", "basemodel", "basemodel_many", "mixed", ], ) def test_convert_list(self, result, expected_text): """Test that a list is converted to a TextContent.""" converted = _convert_to_content(result) assert converted == [TextContent(type="text", text=expected_text)] @pytest.mark.parametrize( argnames="content_block", argvalues=[ (TextContent(type="text", text="hello")), (ImageContent(type="image", data="fakeimagedata", mimeType="image/png")), (AudioContent(type="audio", data="fakeaudiodata", mimeType="audio/mpeg")), ( ResourceLink( type="resource_link", name="test resource", uri=AnyUrl("resource://test"), ) ), ( EmbeddedResource( type="resource", resource=TextResourceContents( uri=AnyUrl("resource://test"), mimeType="text/plain", text="resource content", ), ) ), ], ids=["text", "image", "audio", "resource link", "embedded resource"], ) def test_convert_content_block(self, content_block): converted = _convert_to_content(content_block) assert converted == [content_block] converted = _convert_to_content([content_block, content_block]) assert converted == [content_block, content_block] @pytest.mark.parametrize( argnames=("result", "expected"), argvalues=[ ( Image(data=b"fakeimagedata"), [ ImageContent( type="image", data="ZmFrZWltYWdlZGF0YQ==", mimeType="image/png" ) ], ), ( Audio(data=b"fakeaudiodata"), [ AudioContent( type="audio", data="ZmFrZWF1ZGlvZGF0YQ==", mimeType="audio/wav" ) ], ), ( File(data=b"filedata", format="octet-stream"), [ EmbeddedResource( type="resource", resource=BlobResourceContents( uri=AnyUrl("file:///resource.octet-stream"), blob="ZmlsZWRhdGE=", mimeType="application/octet-stream", ), ) ], ), ], ids=["image", "audio", "file"], ) def test_convert_helpers(self, result, expected): converted = _convert_to_content(result) assert converted == expected converted = _convert_to_content([result, result]) assert converted == expected * 2 def test_convert_mixed_content(self): result = [ "hello", 123, 123.45, {"key": "value"}, SampleModel(x=1, y="hello"), Image(data=b"fakeimagedata"), Audio(data=b"fakeaudiodata"), ResourceLink( type="resource_link", name="test resource", uri=AnyUrl("resource://test"), ), EmbeddedResource( type="resource", resource=TextResourceContents( uri=AnyUrl("resource://test"), mimeType="text/plain", text="resource content", ), ), ] converted = _convert_to_content(result) assert converted == snapshot( [ TextContent(type="text", text="hello"), TextContent(type="text", text="123"), TextContent(type="text", text="123.45"), TextContent(type="text", text='{"key":"value"}'), TextContent(type="text", text='{"x":1,"y":"hello"}'), ImageContent( type="image", data="ZmFrZWltYWdlZGF0YQ==", mimeType="image/png" ), AudioContent( type="audio", data="ZmFrZWF1ZGlvZGF0YQ==", mimeType="audio/wav" ), ResourceLink( name="test resource", uri=AnyUrl("resource://test"), type="resource_link", ), EmbeddedResource( type="resource", resource=TextResourceContents( uri=AnyUrl("resource://test"), mimeType="text/plain", text="resource content", ), ), ] ) def test_empty_list(self): """Test that an empty list results in an empty list.""" result = _convert_to_content([]) assert isinstance(result, list) assert len(result) == 0 def test_empty_dict(self): """Test that an empty dictionary is converted to TextContent.""" result = _convert_to_content({}) assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], TextContent) assert result[0].text == "{}" class TestAutomaticStructuredContent: """Tests for automatic structured content generation based on return types.""" async def test_dict_return_creates_structured_content_without_schema(self): """Test that dict returns automatically create structured content even without output schema.""" def get_user_data(user_id: str) -> dict: return {"name": "Alice", "age": 30, "active": True} # No explicit output schema provided tool = Tool.from_function(get_user_data) result = await tool.run({"user_id": "123"}) # Should have both content and structured content assert len(result.content) == 1 assert isinstance(result.content[0], TextContent) assert result.structured_content == {"name": "Alice", "age": 30, "active": True} async def test_dataclass_return_creates_structured_content_without_schema(self): """Test that dataclass returns automatically create structured content even without output schema.""" @dataclass class UserProfile: name: str age: int email: str def get_profile(user_id: str) -> UserProfile: 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=None) result = await tool.run({"user_id": "456"}) # Should have both content and structured content assert len(result.content) == 1 assert isinstance(result.content[0], TextContent) # Dataclass should serialize to dict assert result.structured_content == { "name": "Bob", "age": 25, "email": "bob@example.com", } async def test_pydantic_model_return_creates_structured_content_without_schema( self, ): """Test that Pydantic model returns automatically create structured content even without output schema.""" class UserData(BaseModel): username: str score: int verified: bool def get_user_stats(user_id: str) -> UserData: return UserData(username="charlie", score=100, verified=True) # 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"}) # Should have both content and structured content assert len(result.content) == 1 assert isinstance(result.content[0], TextContent) # Pydantic model should serialize to dict assert result.structured_content == { "username": "charlie", "score": 100, "verified": True, } async def test_self_referencing_dataclass_not_wrapped(self): """Test that self-referencing dataclasses are not wrapped in result field.""" @dataclass class ReturnThing: value: int stuff: list["ReturnThing"] def return_things() -> ReturnThing: return ReturnThing(value=123, stuff=[ReturnThing(value=456, stuff=[])]) tool = Tool.from_function(return_things) result = await tool.run({}) # Should have structured content without wrapping assert result.structured_content is not None # Should NOT be wrapped in "result" field assert "result" not in result.structured_content # Should have the actual data directly assert result.structured_content == { "value": 123, "stuff": [{"value": 456, "stuff": []}], } async def test_self_referencing_pydantic_model_has_type_object_at_root(self): """Test that self-referencing Pydantic models have type: object at root. MCP spec requires outputSchema to have "type": "object" at the root level. Pydantic generates schemas with $ref at root for self-referential models, which violates this requirement. FastMCP should resolve the $ref. Regression test for issue #2455. """ class Issue(BaseModel): id: str title: str dependencies: list["Issue"] = [] dependents: list["Issue"] = [] def get_issue(issue_id: str) -> Issue: return Issue(id=issue_id, title="Test") tool = Tool.from_function(get_issue) # The output schema should have "type": "object" at root, not $ref assert tool.output_schema is not None assert tool.output_schema.get("type") == "object" assert "properties" in tool.output_schema # Should still have $defs for nested references assert "$defs" in tool.output_schema # Should NOT have $ref at root level assert "$ref" not in tool.output_schema async def test_self_referencing_model_outputschema_mcp_compliant(self): """Test that self-referencing model schemas are MCP spec compliant. The MCP spec requires: - type: "object" at root level - properties field - required field (optional) This ensures clients can properly validate the schema. Regression test for issue #2455. """ class Node(BaseModel): id: str children: list["Node"] = [] def get_node() -> Node: return Node(id="1") tool = Tool.from_function(get_node) # Schema should be MCP-compliant assert tool.output_schema is not None assert tool.output_schema.get("type") == "object", ( "MCP spec requires 'type': 'object' at root" ) assert "properties" in tool.output_schema assert "id" in tool.output_schema["properties"] assert "children" in tool.output_schema["properties"] # Required should include 'id' assert "id" in tool.output_schema.get("required", []) async def test_int_return_no_structured_content_without_schema(self): """Test that int returns don't create structured content without output schema.""" def calculate_sum(a: int, b: int): """No return annotation.""" return a + b # No output schema tool = Tool.from_function(calculate_sum) result = await tool.run({"a": 5, "b": 3}) # Should only have content, no structured content assert len(result.content) == 1 assert isinstance(result.content[0], TextContent) assert result.content[0].text == "8" assert result.structured_content is None async def test_str_return_no_structured_content_without_schema(self): """Test that str returns don't create structured content without output schema.""" def get_greeting(name: str): """No return annotation.""" return f"Hello, {name}!" # No output schema tool = Tool.from_function(get_greeting) result = await tool.run({"name": "World"}) # Should only have content, no structured content assert len(result.content) == 1 assert isinstance(result.content[0], TextContent) assert result.content[0].text == "Hello, World!" assert result.structured_content is None async def test_list_return_no_structured_content_without_schema(self): """Test that list returns don't create structured content without output schema.""" def get_numbers(): """No return annotation.""" return [1, 2, 3, 4, 5] # No output schema tool = Tool.from_function(get_numbers) result = await tool.run({}) assert result.structured_content is None assert result.content == snapshot( [TextContent(type="text", text="[1,2,3,4,5]")] ) async def test_audio_return_creates_no_structured_content(self): """Test that audio returns don't create structured content.""" def get_audio() -> AudioContent: """No return annotation.""" return Audio(data=b"fakeaudiodata").to_audio_content() # No output schema tool = Tool.from_function(get_audio) result = await tool.run({}) assert result.content == snapshot( [ AudioContent( type="audio", data="ZmFrZWF1ZGlvZGF0YQ==", mimeType="audio/wav" ) ] ) assert result.structured_content is None async def test_int_return_with_schema_creates_structured_content(self): """Test that int returns DO create structured content when there's an output schema.""" def calculate_sum(a: int, b: int) -> int: """With return annotation.""" return a + b # Output schema should be auto-generated from annotation tool = Tool.from_function(calculate_sum) assert tool.output_schema is not None result = await tool.run({"a": 5, "b": 3}) # Should have both content and structured content assert len(result.content) == 1 assert isinstance(result.content[0], TextContent) assert result.content[0].text == "8" assert result.structured_content == {"result": 8} async def test_client_automatic_deserialization_with_dict_result(self): """Test that clients automatically deserialize dict results from structured content.""" from fastmcp import FastMCP from fastmcp.client import Client mcp = FastMCP() @mcp.tool def get_user_info(user_id: str) -> dict: return {"name": "Alice", "age": 30, "active": True} async with Client(mcp) as client: result = await client.call_tool("get_user_info", {"user_id": "123"}) # Client should provide the deserialized data assert result.data == {"name": "Alice", "age": 30, "active": True} assert result.structured_content == { "name": "Alice", "age": 30, "active": True, } assert len(result.content) == 1 async def test_client_automatic_deserialization_with_dataclass_result(self): """Test that clients automatically deserialize dataclass results from structured content.""" from fastmcp import FastMCP from fastmcp.client import Client mcp = FastMCP() @dataclass class UserProfile: name: str age: int verified: bool @mcp.tool def get_profile(user_id: str) -> UserProfile: return UserProfile(name="Bob", age=25, verified=True) async with Client(mcp) as client: result = await client.call_tool("get_profile", {"user_id": "456"}) # Client should deserialize back to a dataclass (but type name is lost with title pruning) assert result.data.__class__.__name__ == "Root" assert result.data.name == "Bob" assert result.data.age == 25 assert result.data.verified is True class TestToolResultCasting: @pytest.fixture async def client(self): from fastmcp import FastMCP from fastmcp.client import Client mcp = FastMCP() @mcp.tool def test_tool( unstructured: str | None = None, structured: dict[str, Any] | None = None, meta: dict[str, Any] | None = None, ): return ToolResult( content=unstructured, structured_content=structured, meta=meta, ) async with Client(mcp) as client: yield client async def test_only_unstructured_content(self, client): result = await client.call_tool("test_tool", {"unstructured": "test data"}) assert result.content[0].type == "text" assert result.content[0].text == "test data" assert result.structured_content is None assert result.meta is None async def test_neither_unstructured_or_structured_content(self, client): from fastmcp.exceptions import ToolError with pytest.raises(ToolError): await client.call_tool("test_tool", {}) async def test_structured_and_unstructured_content(self, client): result = await client.call_tool( "test_tool", {"unstructured": "test data", "structured": {"data_type": "test"}}, ) assert result.content[0].type == "text" assert result.content[0].text == "test data" assert result.structured_content == {"data_type": "test"} assert result.meta is None async def test_structured_unstructured_and_meta_content(self, client): result = await client.call_tool( "test_tool", { "unstructured": "test data", "structured": {"data_type": "test"}, "meta": {"some": "metadata"}, }, ) assert result.content[0].type == "text" assert result.content[0].text == "test data" assert result.structured_content == {"data_type": "test"} assert result.meta == {"some": "metadata"} class TestUnionReturnTypes: """Tests for tools with union return types.""" async def test_dataclass_union_string_works(self): """Test that union of dataclass and string works correctly.""" @dataclass class Data: value: int def get_data(return_error: bool) -> Data | str: if return_error: return "error occurred" return Data(value=42) tool = Tool.from_function(get_data) # Test returning dataclass result1 = await tool.run({"return_error": False}) assert result1.structured_content == {"result": {"value": 42}} # Test returning string result2 = await tool.run({"return_error": True}) assert result2.structured_content == {"result": "error occurred"} class TestSerializationAlias: """Tests for Pydantic field serialization alias support in tool output schemas.""" def test_output_schema_respects_serialization_alias(self): """Test that Tool.from_function generates output schema using serialization alias.""" from pydantic import AliasChoices, BaseModel, Field class Component(BaseModel): """Model with multiple validation aliases but specific serialization alias.""" component_id: str = Field( validation_alias=AliasChoices("id", "componentId"), serialization_alias="componentId", description="The ID of the component", ) async def get_component( component_id: str, ) -> Annotated[Component, Field(description="The component.")]: # API returns data with 'id' field api_data = {"id": component_id} return Component.model_validate(api_data) tool = Tool.from_function(get_component, name="get-component") # The output schema should use the serialization alias 'componentId' # not the first validation alias 'id' assert tool.output_schema is not None # Object schemas have properties directly at root (MCP spec compliance) # Root-level $refs are resolved to ensure type: object at root assert "properties" in tool.output_schema assert tool.output_schema.get("type") == "object" # Should have 'componentId' not 'id' in properties assert "componentId" in tool.output_schema["properties"] assert "id" not in tool.output_schema["properties"] # Should require 'componentId' not 'id' assert "componentId" in tool.output_schema.get("required", []) assert "id" not in tool.output_schema.get("required", []) async def test_tool_execution_with_serialization_alias(self): """Test that tool execution works correctly with serialization aliases.""" from pydantic import AliasChoices, BaseModel, Field from fastmcp import Client, FastMCP class Component(BaseModel): """Model with multiple validation aliases but specific serialization alias.""" component_id: str = Field( validation_alias=AliasChoices("id", "componentId"), serialization_alias="componentId", description="The ID of the component", ) mcp = FastMCP("TestServer") @mcp.tool async def get_component( component_id: str, ) -> Annotated[Component, Field(description="The component.")]: # API returns data with 'id' field api_data = {"id": component_id} return Component.model_validate(api_data) async with Client(mcp) as client: # Execute the tool - this should work without validation errors result = await client.call_tool( "get_component", {"component_id": "test123"} ) # The result should contain the serialized form with 'componentId' assert result.structured_content is not None # Object types may be wrapped in "result" or not, depending on schema structure if "result" in result.structured_content: component_data = result.structured_content["result"] else: component_data = result.structured_content assert component_data["componentId"] == "test123" assert "id" not in component_data class TestToolTitle: """Tests for tool title functionality.""" def test_tool_with_title(self): """Test that tools can have titles and they appear in MCP conversion.""" def calculate(x: int, y: int) -> int: """Calculate the sum of two numbers.""" return x + y tool = Tool.from_function( calculate, name="calc", title="Advanced Calculator Tool", description="Custom description", ) assert tool.name == "calc" assert tool.title == "Advanced Calculator Tool" assert tool.description == "Custom description" # Test MCP conversion includes title mcp_tool = tool.to_mcp_tool() assert mcp_tool.name == "calc" assert ( hasattr(mcp_tool, "title") and mcp_tool.title == "Advanced Calculator Tool" ) def test_tool_without_title(self): """Test that tools without titles use name as display name.""" def multiply(a: int, b: int) -> int: return a * b tool = Tool.from_function(multiply) assert tool.name == "multiply" assert tool.title is None # Test MCP conversion doesn't include title when None mcp_tool = tool.to_mcp_tool() assert mcp_tool.name == "multiply" assert not hasattr(mcp_tool, "title") or mcp_tool.title is None def test_tool_title_priority(self): """Test that explicit title takes priority over annotations.title.""" from mcp.types import ToolAnnotations def divide(x: int, y: int) -> float: """Divide two numbers.""" return x / y # Test with both explicit title and annotations.title annotations = ToolAnnotations(title="Annotation Title") tool = Tool.from_function( divide, name="div", title="Explicit Title", annotations=annotations, ) assert tool.title == "Explicit Title" assert tool.annotations is not None assert tool.annotations.title == "Annotation Title" # Explicit title should take priority mcp_tool = tool.to_mcp_tool() assert mcp_tool.title == "Explicit Title" def test_tool_annotations_title_fallback(self): """Test that annotations.title is used when no explicit title is provided.""" from mcp.types import ToolAnnotations def modulo(x: int, y: int) -> int: """Get modulo of two numbers.""" return x % y # Test with only annotations.title (no explicit title) annotations = ToolAnnotations(title="Annotation Title") tool = Tool.from_function( modulo, name="mod", annotations=annotations, ) assert tool.title is None assert tool.annotations is not None assert tool.annotations.title == "Annotation Title" # Should fall back to annotations.title mcp_tool = tool.to_mcp_tool() assert mcp_tool.title == "Annotation Title" class TestToolNameValidation: """Tests for tool name validation per MCP specification (SEP-986).""" @pytest.fixture def caplog_for_mcp_validation(self, caplog): """Capture logs from the MCP SDK's tool name validation logger.""" import logging caplog.set_level(logging.WARNING) logger = logging.getLogger("mcp.shared.tool_name_validation") original_level = logger.level logger.setLevel(logging.WARNING) logger.addHandler(caplog.handler) try: yield caplog finally: logger.removeHandler(caplog.handler) logger.setLevel(original_level) @pytest.mark.parametrize( "name", [ "valid_tool", "valid-tool", "valid.tool", "ValidTool", "tool123", "a", "a" * 128, ], ) def test_valid_tool_names_no_warnings(self, name, caplog_for_mcp_validation): """Valid tool names should not produce warnings.""" def fn() -> str: return "test" tool = Tool.from_function(fn, name=name) assert tool.name == name assert "Tool name validation warning" not in caplog_for_mcp_validation.text def test_tool_name_with_spaces_warns(self, caplog_for_mcp_validation): """Tool names with spaces should produce a warning.""" def fn() -> str: return "test" tool = Tool.from_function(fn, name="my tool") assert tool.name == "my tool" assert "Tool name validation warning" in caplog_for_mcp_validation.text assert "contains spaces" in caplog_for_mcp_validation.text def test_tool_name_with_invalid_chars_warns(self, caplog_for_mcp_validation): """Tool names with invalid characters should produce a warning.""" def fn() -> str: return "test" tool = Tool.from_function(fn, name="tool@name!") assert tool.name == "tool@name!" assert "Tool name validation warning" in caplog_for_mcp_validation.text assert "invalid characters" in caplog_for_mcp_validation.text def test_tool_name_too_long_warns(self, caplog_for_mcp_validation): """Tool names exceeding 128 characters should produce a warning.""" def fn() -> str: return "test" long_name = "a" * 129 tool = Tool.from_function(fn, name=long_name) assert tool.name == long_name assert "Tool name validation warning" in caplog_for_mcp_validation.text assert "exceeds maximum length" in caplog_for_mcp_validation.text def test_tool_name_with_leading_dash_warns(self, caplog_for_mcp_validation): """Tool names starting with dash should produce a warning.""" def fn() -> str: return "test" tool = Tool.from_function(fn, name="-tool") assert tool.name == "-tool" assert "Tool name validation warning" in caplog_for_mcp_validation.text assert "starts or ends with a dash" in caplog_for_mcp_validation.text def test_tool_still_created_despite_warnings(self, caplog_for_mcp_validation): """Tools with invalid names should still be created (SHOULD not MUST).""" def add(a: int, b: int) -> int: return a + b tool = Tool.from_function(add, name="invalid tool name!") assert tool.name == "invalid tool name!" assert tool.parameters is not None assert "a" in tool.parameters["properties"] assert "b" in tool.parameters["properties"] class TestToolExecutionField: """Tests for the execution field on the base Tool class.""" def test_tool_with_execution_field(self): """Test that Tool can store and return execution metadata.""" tool = Tool( name="my_tool", description="A tool with execution", parameters={"type": "object", "properties": {}}, execution=ToolExecution(taskSupport="optional"), ) mcp_tool = tool.to_mcp_tool() assert mcp_tool.execution is not None assert mcp_tool.execution.taskSupport == "optional" def test_tool_without_execution_field(self): """Test that Tool without execution returns None.""" tool = Tool( name="my_tool", description="A tool without execution", parameters={"type": "object", "properties": {}}, ) mcp_tool = tool.to_mcp_tool() assert mcp_tool.execution is None def test_execution_override_takes_precedence(self): """Test that explicit override takes precedence over field value.""" tool = Tool( name="my_tool", description="A tool", parameters={"type": "object", "properties": {}}, execution=ToolExecution(taskSupport="optional"), ) override_execution = ToolExecution(taskSupport="required") mcp_tool = tool.to_mcp_tool(execution=override_execution) assert mcp_tool.execution is not None assert mcp_tool.execution.taskSupport == "required" async def test_function_tool_task_config_still_works(self): """FunctionTool should still derive execution from task_config.""" async def my_fn() -> str: return "hello" tool = Tool.from_function(my_fn, task=True) mcp_tool = tool.to_mcp_tool() # FunctionTool sets execution from task_config assert mcp_tool.execution is not None assert mcp_tool.execution.taskSupport == "optional" def test_tool_execution_required_mode(self): """Test that Tool can store required execution mode.""" tool = Tool( name="my_tool", description="A tool with required execution", parameters={"type": "object", "properties": {}}, execution=ToolExecution(taskSupport="required"), ) mcp_tool = tool.to_mcp_tool() assert mcp_tool.execution is not None assert mcp_tool.execution.taskSupport == "required" def test_tool_execution_forbidden_mode(self): """Test that Tool can store forbidden execution mode.""" tool = Tool( name="my_tool", description="A tool with forbidden execution", parameters={"type": "object", "properties": {}}, execution=ToolExecution(taskSupport="forbidden"), ) mcp_tool = tool.to_mcp_tool() assert mcp_tool.execution is not None assert mcp_tool.execution.taskSupport == "forbidden" class TestToolCallable: """Test tools with callable objects.""" async def test_callable_object_sync(self): """Test that callable objects with sync __call__ work.""" class MyTool: def __init__(self, multiplier: int): self.multiplier = multiplier def __call__(self, x: int) -> int: return x * self.multiplier tool = Tool.from_function(MyTool(3)) result = await tool.run({"x": 5}) assert result.content == [TextContent(type="text", text="15")] async def test_callable_object_async(self): """Test that callable objects with async __call__ work.""" class AsyncTool: def __init__(self, multiplier: int): self.multiplier = multiplier async def __call__(self, x: int) -> int: return x * self.multiplier tool = Tool.from_function(AsyncTool(4)) result = await tool.run({"x": 5}) assert result.content == [TextContent(type="text", text="20")] class TestSyncToolConcurrency: """Tests for concurrent execution of sync tools without blocking the event loop.""" async def test_sync_tools_run_concurrently(self): """Test that sync tools run in threadpool and don't block each other. Uses a threading barrier to prove concurrent execution: all calls must reach the barrier simultaneously for any to proceed. If they ran sequentially, only one would reach the barrier and it would timeout. """ import asyncio import threading num_calls = 3 # Barrier requires all threads to arrive before any proceed # Short timeout since concurrent threads should arrive within milliseconds barrier = threading.Barrier(num_calls, timeout=0.5) def concurrent_tool(x: int) -> int: """Tool that proves concurrency via barrier synchronization.""" # If calls run sequentially, only 1 thread reaches barrier and times out # If calls run concurrently, all 3 reach barrier and proceed barrier.wait() return x * 2 tool = Tool.from_function(concurrent_tool) # Run concurrent calls - will raise BrokenBarrierError if not concurrent results = await asyncio.gather( tool.run({"x": 1}), tool.run({"x": 2}), tool.run({"x": 3}), ) # Verify results assert [r.content for r in results] == [ [TextContent(type="text", text="2")], [TextContent(type="text", text="4")], [TextContent(type="text", text="6")], ] async def test_sync_tool_with_context_runs_concurrently(self): """Test that sync tools with Context dependency also run concurrently.""" import asyncio import threading from fastmcp import Context, FastMCP num_calls = 3 barrier = threading.Barrier(num_calls, timeout=0.5) mcp = FastMCP("test") @mcp.tool def ctx_tool(x: int, ctx: Context) -> str: """A sync tool with context that uses barrier to prove concurrency.""" barrier.wait() return f"{ctx.fastmcp.name}:{x}" # Run concurrent calls through the server interface (which sets up Context) results = await asyncio.gather( mcp.call_tool("ctx_tool", {"x": 1}), mcp.call_tool("ctx_tool", {"x": 2}), mcp.call_tool("ctx_tool", {"x": 3}), ) # Verify results for i, result in enumerate(results, 1): assert result.content == [TextContent(type="text", text=f"test:{i}")]