mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-20 20:44:17 +02:00
953 lines
34 KiB
Python
953 lines
34 KiB
Python
import json
|
|
from dataclasses import dataclass
|
|
from typing import Annotated, Any, TypedDict
|
|
|
|
import pytest
|
|
from mcp.types import (
|
|
AudioContent,
|
|
EmbeddedResource,
|
|
ImageContent,
|
|
TextContent,
|
|
TextResourceContents,
|
|
)
|
|
from pydantic import AnyUrl, BaseModel, Field, TypeAdapter
|
|
|
|
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"
|
|
assert tool.output_schema == {"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 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",
|
|
[
|
|
None,
|
|
int,
|
|
float,
|
|
bool,
|
|
str,
|
|
int | float,
|
|
list[int],
|
|
list[int | float],
|
|
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)
|
|
assert tool.output_schema == TypeAdapter(annotation).json_schema()
|
|
|
|
@pytest.mark.parametrize(
|
|
"annotation",
|
|
[
|
|
Any,
|
|
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)
|
|
assert tool.output_schema == TypeAdapter(annotation).json_schema()
|
|
|
|
@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)
|
|
assert tool.output_schema == TypeAdapter(expected).json_schema()
|
|
|
|
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)
|
|
assert tool.output_schema == TypeAdapter(Person).json_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 == TypeAdapter(Person).json_schema()
|
|
|
|
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 == TypeAdapter(Person).json_schema()
|
|
|
|
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": "string", "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": "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": "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": "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": "array", "items": {"type": "number"}}
|
|
|
|
tool = Tool.from_function(func, output_schema=custom_schema)
|
|
assert tool.output_schema == custom_schema
|
|
|
|
|
|
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 json.loads(result[0].text) == {"a": 1}
|
|
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 json.loads(result[0].text) == [
|
|
1,
|
|
{"type": "text", "text": "hello", "annotations": None, "_meta": None},
|
|
]
|