import pytest from mcp.types import ( AudioContent, EmbeddedResource, ImageContent, TextContent, TextResourceContents, ) from pydantic import AnyUrl, BaseModel from fastmcp import FastMCP from fastmcp.client import Client from fastmcp.exceptions import ToolError from fastmcp.tools.tool import Tool, _convert_to_content from fastmcp.utilities.tests import temporary_settings 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.name == "add" assert tool.description == "Add two numbers." assert len(tool.parameters["properties"]) == 2 assert tool.parameters["properties"]["a"]["type"] == "integer" assert tool.parameters["properties"]["b"]["type"] == "integer" 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.name == "fetch_data" assert tool.description == "Fetch data from URL." assert tool.parameters["properties"]["url"]["type"] == "string" 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.name == "Adder" assert tool.description == "Adds two numbers." assert len(tool.parameters["properties"]) == 2 assert tool.parameters["properties"]["x"]["type"] == "integer" assert tool.parameters["properties"]["y"]["type"] == "integer" 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.name == "Adder" assert tool.description == "Adds two numbers." assert len(tool.parameters["properties"]) == 2 assert tool.parameters["properties"]["x"]["type"] == "integer" assert tool.parameters["properties"]["y"]["type"] == "integer" 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.name == "create_user" assert tool.description == "Create a new user." assert "name" in tool.parameters["$defs"]["UserInput"]["properties"] assert "age" in tool.parameters["$defs"]["UserInput"]["properties"] assert "flag" in tool.parameters["properties"] async def test_tool_with_image_return(self): def image_tool(data: bytes) -> Image: return Image(data=data) tool = Tool.from_function(image_tool) result = await tool.run({"data": "test.png"}) assert tool.parameters["properties"]["data"]["type"] == "string" assert isinstance(result[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) result = await tool.run({"data": "test.wav"}) assert tool.parameters["properties"]["data"]["type"] == "string" assert isinstance(result[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) result = await tool.run({"data": "test.bin"}) assert tool.parameters["properties"]["data"]["type"] == "string" assert len(result) == 1 assert isinstance(result[0], EmbeddedResource) assert result[0].type == "resource" assert hasattr(result[0], "resource") resource = result[0].resource assert resource.mimeType == "application/octet-stream" 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.name == "my_tool" 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.parameters["properties"]["_a"]["type"] == "integer" assert tool.parameters["properties"]["_b"]["type"] == "integer" 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 tool.name == "add" assert tool.description == "Add two numbers." assert "self" not in tool.parameters["properties"] 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"] async def test_tool_serializer(self): """Test that a tool's serializer is used to serialize the result.""" def custom_serializer(data) -> str: return f"Custom serializer: {data}" def process_list(items: list[int]) -> int: return sum(items) tool = Tool.from_function(process_list, serializer=custom_serializer) result = await tool.run(arguments={"items": [1, 2, 3, 4, 5]}) assert isinstance(result[0], TextContent) assert result[0].text == "Custom serializer: 15" class TestLegacyToolJsonParsing: """Tests for Tool's JSON pre-parsing functionality.""" @pytest.fixture(autouse=True) def enable_legacy_json_parsing(self): with temporary_settings(tool_attempt_parse_json_args=True): yield async def test_json_string_arguments(self): """Test that JSON string arguments are parsed and validated correctly""" def simple_func(x: int, y: list[str]) -> str: return f"{x}-{','.join(y)}" # Create a tool to use its JSON pre-parsing logic tool = Tool.from_function(simple_func) # Prepare arguments where some are JSON strings json_args = { "x": 1, "y": '["a", "b", "c"]', # JSON string } # Run the tool which will do JSON parsing result = await tool.run(json_args) assert result[0].text == "1-a,b,c" # type: ignore[attr-dict] async def test_str_vs_list_str(self): """Test handling of string vs list[str] type annotations.""" def func_with_str_types(str_or_list: str | list[str]) -> str | list[str]: return str_or_list tool = Tool.from_function(func_with_str_types) # Test regular string input (should remain a string) result = await tool.run({"str_or_list": "hello"}) assert result[0].text == "hello" # type: ignore[attr-dict] # Test JSON string input (should be parsed as a string) result = await tool.run({"str_or_list": '"hello"'}) assert result[0].text == "hello" # type: ignore[attr-dict] # Test JSON list input (should be parsed as a list) result = await tool.run({"str_or_list": '["hello", "world"]'}) # The exact formatting might vary, so we just check that it contains the key elements text_without_whitespace = result[0].text.replace(" ", "").replace("\n", "") # type: ignore[attr-dict] assert "hello" in text_without_whitespace assert "world" in text_without_whitespace assert "[" in text_without_whitespace assert "]" in text_without_whitespace async def test_keep_str_as_str(self): """Test that string arguments are kept as strings when they're not valid JSON""" def func_with_str_types(string: str) -> str: return string tool = Tool.from_function(func_with_str_types) # Invalid JSON should remain a string invalid_json = "{'nice to meet you': 'hello', 'goodbye': 5}" result = await tool.run({"string": invalid_json}) assert result[0].text == invalid_json # type: ignore[attr-dict] async def test_keep_str_union_as_str(self): """Test that string arguments are kept as strings when parsing would create an invalid value""" def func_with_str_types( string: str | dict[int, str] | None, ) -> str | dict[int, str] | None: return string tool = Tool.from_function(func_with_str_types) # Invalid JSON for the union type should remain a string invalid_json = "{'nice to meet you': 'hello', 'goodbye': 5}" result = await tool.run({"string": invalid_json}) assert result[0].text == invalid_json # type: ignore[attr-dict] async def test_complex_type_validation(self): """Test that parsed JSON is validated against complex types""" class SomeModel(BaseModel): x: int y: dict[int, str] def func_with_complex_type(data: SomeModel) -> SomeModel: return data tool = Tool.from_function(func_with_complex_type) # Valid JSON for the model valid_json = '{"x": 1, "y": {"1": "hello"}}' result = await tool.run({"data": valid_json}) assert '"x": 1' in result[0].text # type: ignore[attr-dict] assert '"y": {' in result[0].text # type: ignore[attr-dict] assert '"1": "hello"' in result[0].text # type: ignore[attr-dict] # Invalid JSON for the model (y has string keys, not int keys) # Should throw a validation error invalid_json = '{"x": 1, "y": {"invalid": "hello"}}' with pytest.raises(Exception): await tool.run({"data": invalid_json}) async def test_tool_list_coercion(self): """Test JSON string to collection type coercion.""" mcp = FastMCP() @mcp.tool def process_list(items: list[int]) -> int: return sum(items) async with Client(mcp) as client: # JSON array string should be coerced to list result = await client.call_tool( "process_list", {"items": "[1, 2, 3, 4, 5]"} ) assert result[0].text == "15" # type: ignore[attr-dict] async def test_tool_list_coercion_error(self): """Test that a list coercion error is raised if the input is not a valid list.""" mcp = FastMCP() @mcp.tool def process_list(items: list[int]) -> int: return sum(items) async with Client(mcp) as client: with pytest.raises( ToolError, match="Error calling tool 'process_list'", ): await client.call_tool("process_list", {"items": "['a', 'b', 3]"}) async def test_tool_dict_coercion(self): """Test JSON string to dict type coercion.""" mcp = FastMCP() @mcp.tool def process_dict(data: dict[str, int]) -> int: return sum(data.values()) async with Client(mcp) as client: # JSON object string should be coerced to dict result = await client.call_tool( "process_dict", {"data": '{"a": 1, "b": "2", "c": 3}'} ) assert result[0].text == "6" # type: ignore[attr-dict] async def test_tool_set_coercion(self): """Test JSON string to set type coercion.""" mcp = FastMCP() @mcp.tool def process_set(items: set[int]) -> int: assert isinstance(items, set) return sum(items) async with Client(mcp) as client: result = await client.call_tool("process_set", {"items": "[1, 2, 3, 4, 5]"}) assert result[0].text == "15" # type: ignore[attr-dict] async def test_tool_tuple_coercion(self): """Test JSON string to tuple type coercion.""" mcp = FastMCP() @mcp.tool def process_tuple(items: tuple[int, str]) -> int: assert isinstance(items, tuple) return items[0] + len(items[1]) async with Client(mcp) as client: result = await client.call_tool("process_tuple", {"items": '["1", "two"]'}) assert isinstance(result[0], TextContent) assert result[0].text == "4" # type: ignore[attr-dict] class TestConvertResultToContent: """Tests for the _convert_to_content helper function.""" def test_none_result(self): """Test that None results in an empty list.""" result = _convert_to_content(None) assert isinstance(result, list) assert len(result) == 0 def test_text_content_result(self): """Test that TextContent is returned as a list containing itself.""" content = TextContent(type="text", text="hello") result = _convert_to_content(content) assert isinstance(result, list) assert len(result) == 1 assert result[0] is content def test_image_content_result(self): """Test that ImageContent is returned as a list containing itself.""" content = ImageContent(type="image", data="fakeimagedata", mimeType="image/png") result = _convert_to_content(content) assert isinstance(result, list) assert len(result) == 1 assert result[0] is content def test_embedded_resource_result(self): """Test that EmbeddedResource is returned as a list containing itself.""" content = EmbeddedResource( type="resource", resource=TextResourceContents( uri=AnyUrl("resource://test"), mimeType="text/plain", text="resource content", ), ) result = _convert_to_content(content) assert isinstance(result, list) assert len(result) == 1 assert result[0] is content def test_image_object_result(self): """Test that an Image object is converted to ImageContent.""" image_obj = Image(data=b"fakeimagedata") result = _convert_to_content(image_obj) assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], ImageContent) assert result[0].data == "ZmFrZWltYWdlZGF0YQ==" def test_audio_object_result(self): """Test that an Audio object is converted to AudioContent.""" audio_obj = Audio(data=b"fakeaudiodata") result = _convert_to_content(audio_obj) assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], AudioContent) assert result[0].data == "ZmFrZWF1ZGlvZGF0YQ==" def test_file_object_result(self): """Test that a File object is converted to EmbeddedResource with BlobResourceContents.""" file_obj = File(data=b"filedata", format="octet-stream") result = _convert_to_content(file_obj) assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], EmbeddedResource) assert result[0].type == "resource" assert hasattr(result[0], "resource") resource = result[0].resource assert resource.mimeType == "application/octet-stream" # Check for blob attribute and its value assert hasattr(resource, "blob") assert getattr(resource, "blob") == "ZmlsZWRhdGE=" # base64 encoded "filedata" # Convert URI to string for startswith check assert str(resource.uri).startswith("file:///resource.octet-stream") def test_file_object_text_result(self): """Test that a File object with text data is converted to EmbeddedResource with TextResourceContents.""" file_obj = File(data=b"sometext", format="plain") result = _convert_to_content(file_obj) assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], EmbeddedResource) assert result[0].type == "resource" resource = result[0].resource assert isinstance(resource, TextResourceContents) assert resource.mimeType == "text/plain" assert resource.text == "sometext" def test_basic_type_result(self): """Test that a basic type is converted to TextContent.""" result = _convert_to_content(123) assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], TextContent) assert result[0].text == "123" result = _convert_to_content("hello") assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], TextContent) assert result[0].text == "hello" result = _convert_to_content({"a": 1, "b": 2}) assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], TextContent) assert result[0].text == '{\n "a": 1,\n "b": 2\n}' def test_list_of_basic_types(self): """Test that a list of basic types is converted to a single TextContent.""" result = _convert_to_content([1, "two", {"c": 3}]) assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], TextContent) assert result[0].text == '[\n 1,\n "two",\n {\n "c": 3\n }\n]' def test_list_of_mcp_types(self): """Test that a list of MCP types is returned as a list of those types.""" content1 = TextContent(type="text", text="hello") content2 = ImageContent( type="image", data="fakeimagedata2", mimeType="image/png" ) result = _convert_to_content([content1, content2]) assert isinstance(result, list) assert len(result) == 2 assert result[0] is content1 assert result[1] is content2 def test_list_of_mixed_types(self): """Test that a list of mixed types is converted correctly.""" content1 = TextContent(type="text", text="hello") image_obj = Image(data=b"fakeimagedata") basic_data = {"a": 1} result = _convert_to_content([content1, image_obj, basic_data]) assert isinstance(result, list) assert len(result) == 3 text_content_count = sum(isinstance(item, TextContent) for item in result) image_content_count = sum(isinstance(item, ImageContent) for item in result) assert text_content_count == 2 assert image_content_count == 1 text_item = next(item for item in result if isinstance(item, TextContent)) assert text_item.text == '{\n "a": 1\n}' image_item = next(item for item in result if isinstance(item, ImageContent)) assert image_item.data == "ZmFrZWltYWdlZGF0YQ==" def test_list_of_mixed_types_list(self): """Test that a list of mixed types, including a list as one of the elements, is converted correctly.""" content1 = TextContent(type="text", text="hello") image_obj = Image(data=b"fakeimagedata") basic_data = [{"a": 1}, {"b": 2}] result = _convert_to_content([content1, image_obj, basic_data]) assert isinstance(result, list) assert len(result) == 3 text_content_count = sum(isinstance(item, TextContent) for item in result) image_content_count = sum(isinstance(item, ImageContent) for item in result) assert text_content_count == 2 assert image_content_count == 1 text_item = next(item for item in result if isinstance(item, TextContent)) assert text_item.text == '[\n {\n "a": 1\n },\n {\n "b": 2\n }\n]' image_item = next(item for item in result if isinstance(item, ImageContent)) assert image_item.data == "ZmFrZWltYWdlZGF0YQ==" def test_list_of_mixed_types_with_audio(self): """Test that a list of mixed types including Audio is converted correctly.""" content1 = TextContent(type="text", text="hello") audio_obj = Audio(data=b"fakeaudiodata") basic_data = {"a": 1} result = _convert_to_content([content1, audio_obj, basic_data]) assert isinstance(result, list) assert len(result) == 3 text_content_count = sum(isinstance(item, TextContent) for item in result) audio_content_count = sum(isinstance(item, AudioContent) for item in result) assert text_content_count == 2 assert audio_content_count == 1 text_item = next(item for item in result if isinstance(item, TextContent)) assert text_item.text == '{\n "a": 1\n}' audio_item = next(item for item in result if isinstance(item, AudioContent)) assert audio_item.data == "ZmFrZWF1ZGlvZGF0YQ==" def test_list_of_mixed_types_with_file(self): """Test that a list of mixed types including File is converted correctly.""" content1 = TextContent(type="text", text="hello") file_obj = File(data=b"filedata", format="octet-stream") basic_data = {"a": 1} result = _convert_to_content([content1, file_obj, basic_data]) assert isinstance(result, list) assert len(result) == 3 text_content_count = sum(isinstance(item, TextContent) for item in result) embedded_content_count = sum( isinstance(item, EmbeddedResource) and item.type == "resource" for item in result ) assert text_content_count == 2 assert embedded_content_count == 1 text_item = next(item for item in result if isinstance(item, TextContent)) assert text_item.text == '{\n "a": 1\n}' embedded_item = next( item for item in result if isinstance(item, EmbeddedResource) and item.type == "resource" ) resource = embedded_item.resource assert resource.mimeType == "application/octet-stream" # Check for blob attribute and its value assert hasattr(resource, "blob") assert getattr(resource, "blob") == "ZmlsZWRhdGE=" 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 == "{}" def test_with_custom_serializer(self): """Test that a custom serializer is used for non-MCP types.""" def custom_serializer(data): return f"Serialized: {data}" result = _convert_to_content({"a": 1}, serializer=custom_serializer) assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], TextContent) assert result[0].text == "Serialized: {'a': 1}" def test_custom_serializer_error_fallback(self, caplog): """Test that if a custom serializer fails, it falls back to the default.""" import logging def custom_serializer_that_fails(data): raise ValueError("Serialization failed") with caplog.at_level(logging.WARNING): result = _convert_to_content( {"a": 1}, serializer=custom_serializer_that_fails ) assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], TextContent) # Should fall back to default serializer (pydantic_core.to_json) assert result[0].text == '{\n "a": 1\n}' assert "Error serializing tool result" in caplog.text def test_process_as_single_item_flag(self): """Test that _process_as_single_item forces list to be treated as one item.""" result = _convert_to_content([1, "two", {"c": 3}], _process_as_single_item=True) assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], TextContent) assert result[0].text == '[\n 1,\n "two",\n {\n "c": 3\n }\n]' content1 = TextContent(type="text", text="hello") result = _convert_to_content([1, content1], _process_as_single_item=True) assert isinstance(result, list) assert len(result) == 1 assert isinstance(result[0], TextContent) assert ( result[0].text == '[\n 1,\n {\n "type": "text",\n "text": "hello",\n "annotations": null\n }\n]' )