mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 23:29:10 +02:00
1607 lines
55 KiB
Python
1607 lines
55 KiB
Python
from dataclasses import dataclass
|
|
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,
|
|
)
|
|
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.tests import caplog_for_fastmcp
|
|
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(),
|
|
"enabled": True,
|
|
"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"),
|
|
}
|
|
)
|
|
|
|
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(),
|
|
"enabled": True,
|
|
"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"),
|
|
}
|
|
)
|
|
|
|
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(),
|
|
"enabled": True,
|
|
"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,
|
|
},
|
|
}
|
|
)
|
|
|
|
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(),
|
|
"enabled": True,
|
|
"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,
|
|
},
|
|
}
|
|
)
|
|
|
|
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(),
|
|
"enabled": True,
|
|
"parameters": {
|
|
"$defs": {
|
|
"UserInput": {
|
|
"properties": {
|
|
"name": {"type": "string"},
|
|
"age": {"type": "integer"},
|
|
},
|
|
"required": ["name", "age"],
|
|
"type": "object",
|
|
}
|
|
},
|
|
"properties": {
|
|
"user": {"$ref": "#/$defs/UserInput"},
|
|
"flag": {"type": "boolean"},
|
|
},
|
|
"required": ["user", "flag"],
|
|
"type": "object",
|
|
},
|
|
"output_schema": {"additionalProperties": True, "type": "object"},
|
|
"fn": HasName("create_user"),
|
|
}
|
|
)
|
|
|
|
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(),
|
|
"enabled": True,
|
|
"parameters": {
|
|
"properties": {"x": {"title": "X"}},
|
|
"required": ["x"],
|
|
"type": "object",
|
|
},
|
|
}
|
|
)
|
|
|
|
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(),
|
|
"enabled": True,
|
|
"parameters": {
|
|
"properties": {
|
|
"_a": {"type": "integer"},
|
|
"_b": {"type": "integer"},
|
|
},
|
|
"required": ["_a", "_b"],
|
|
"type": "object",
|
|
},
|
|
}
|
|
)
|
|
|
|
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(),
|
|
"enabled": True,
|
|
"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,
|
|
},
|
|
}
|
|
)
|
|
|
|
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]})
|
|
# Custom serializer affects unstructured content
|
|
assert isinstance(result.content[0], TextContent)
|
|
assert result.content[0].text == "Custom serializer: 15"
|
|
# Structured output should have the raw value
|
|
assert result.structured_content == {"result": 15}
|
|
|
|
|
|
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: # type: ignore
|
|
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: # type: ignore
|
|
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: # type: ignore
|
|
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 result.content[0].text == '{"message":"Hello, world!"}' # type: ignore[attr-defined]
|
|
|
|
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 result.content[0].text == "42" # type: ignore[attr-defined]
|
|
|
|
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 result.content[0].text == '{"value":42}' # type: ignore[attr-defined]
|
|
|
|
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 result.content[0].text == "hello" # type: ignore[attr-defined]
|
|
# 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 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."""
|
|
|
|
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 have "type" set to "object"'
|
|
):
|
|
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 == "{}"
|
|
|
|
def test_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 result == snapshot(
|
|
[TextContent(type="text", 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."""
|
|
|
|
def custom_serializer_that_fails(data):
|
|
raise ValueError("Serialization failed")
|
|
|
|
with caplog_for_fastmcp(caplog):
|
|
result = _convert_to_content(
|
|
{"a": 1}, serializer=custom_serializer_that_fails
|
|
)
|
|
|
|
assert isinstance(result, list)
|
|
assert result == snapshot([TextContent(type="text", text='{"a":1}')])
|
|
|
|
assert "Error serializing tool result" in caplog.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_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 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
|
|
|
|
# Check the wrapped result schema
|
|
assert "properties" in tool.output_schema
|
|
assert "result" in tool.output_schema["properties"]
|
|
assert "$defs" in tool.output_schema
|
|
|
|
# Find the Component definition
|
|
component_def = list(tool.output_schema["$defs"].values())[0]
|
|
|
|
# Should have 'componentId' not 'id' in properties
|
|
assert "componentId" in component_def["properties"]
|
|
assert "id" not in component_def["properties"]
|
|
|
|
# Should require 'componentId' not 'id'
|
|
assert "componentId" in component_def["required"]
|
|
assert "id" not in component_def.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
|
|
assert result.structured_content["result"]["componentId"] == "test123"
|
|
assert "id" not in result.structured_content["result"]
|
|
|
|
|
|
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"
|