mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-20 04:24:17 +02:00
Merge pull request #1773 from jlowin/content_block_handling
Cleanup Tool Content Conversion
This commit is contained in:
William Easton
GitHub
commit
1e137641bb
5 changed files with 385 additions and 480 deletions
|
|
@ -10,6 +10,7 @@ from typing import (
|
|||
Any,
|
||||
Generic,
|
||||
Literal,
|
||||
TypeAlias,
|
||||
get_type_hints,
|
||||
)
|
||||
|
||||
|
|
@ -55,6 +56,9 @@ class _UnserializableType:
|
|||
pass
|
||||
|
||||
|
||||
ToolResultSerializerType: TypeAlias = Callable[[Any], str]
|
||||
|
||||
|
||||
def default_serializer(data: Any) -> str:
|
||||
return pydantic_core.to_json(data, fallback=str).decode()
|
||||
|
||||
|
|
@ -70,12 +74,12 @@ class ToolResult:
|
|||
elif content is None:
|
||||
content = structured_content
|
||||
|
||||
self.content = _convert_to_content(content)
|
||||
self.content: list[ContentBlock] = _convert_to_content(result=content)
|
||||
|
||||
if structured_content is not None:
|
||||
try:
|
||||
structured_content = pydantic_core.to_jsonable_python(
|
||||
structured_content
|
||||
value=structured_content
|
||||
)
|
||||
except pydantic_core.PydanticSerializationError as e:
|
||||
logger.error(
|
||||
|
|
@ -112,7 +116,7 @@ class Tool(FastMCPComponent):
|
|||
Field(description="Additional annotations about the tool"),
|
||||
] = None
|
||||
serializer: Annotated[
|
||||
Callable[[Any], str] | None,
|
||||
ToolResultSerializerType | None,
|
||||
Field(description="Optional custom serializer for tool results"),
|
||||
] = None
|
||||
|
||||
|
|
@ -168,7 +172,7 @@ class Tool(FastMCPComponent):
|
|||
annotations: ToolAnnotations | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
output_schema: dict[str, Any] | None | NotSetT | Literal[False] = NotSet,
|
||||
serializer: Callable[[Any], str] | None = None,
|
||||
serializer: ToolResultSerializerType | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> FunctionTool:
|
||||
|
|
@ -210,7 +214,7 @@ class Tool(FastMCPComponent):
|
|||
tags: set[str] | None = None,
|
||||
annotations: ToolAnnotations | None | NotSetT = NotSet,
|
||||
output_schema: dict[str, Any] | None | NotSetT | Literal[False] = NotSet,
|
||||
serializer: Callable[[Any], str] | None = None,
|
||||
serializer: ToolResultSerializerType | None = None,
|
||||
meta: dict[str, Any] | None | NotSetT = NotSet,
|
||||
transform_args: dict[str, ArgTransform] | None = None,
|
||||
enabled: bool | None = None,
|
||||
|
|
@ -248,7 +252,7 @@ class FunctionTool(Tool):
|
|||
annotations: ToolAnnotations | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
output_schema: dict[str, Any] | None | NotSetT | Literal[False] = NotSet,
|
||||
serializer: Callable[[Any], str] | None = None,
|
||||
serializer: ToolResultSerializerType | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> FunctionTool:
|
||||
|
|
@ -317,27 +321,33 @@ class FunctionTool(Tool):
|
|||
|
||||
unstructured_result = _convert_to_content(result, serializer=self.serializer)
|
||||
|
||||
structured_output = None
|
||||
# First handle structured content based on output schema, if any
|
||||
if self.output_schema is not None:
|
||||
if self.output_schema.get("x-fastmcp-wrap-result"):
|
||||
# Schema says wrap - always wrap in result key
|
||||
structured_output = {"result": result}
|
||||
else:
|
||||
structured_output = result
|
||||
# If no output schema, try to serialize the result. If it is a dict, use
|
||||
# it as structured content. If it is not a dict, ignore it.
|
||||
if structured_output is None:
|
||||
if self.output_schema is None:
|
||||
# Do not produce a structured output for MCP Content Types
|
||||
if isinstance(result, ContentBlock | Audio | Image | File) or (
|
||||
isinstance(result, list | tuple)
|
||||
and any(isinstance(item, ContentBlock) for item in result)
|
||||
):
|
||||
return ToolResult(content=unstructured_result)
|
||||
|
||||
# Otherwise, try to serialize the result as a dict
|
||||
try:
|
||||
structured_output = pydantic_core.to_jsonable_python(result)
|
||||
if not isinstance(structured_output, dict):
|
||||
structured_output = None
|
||||
except Exception:
|
||||
structured_content = pydantic_core.to_jsonable_python(result)
|
||||
if isinstance(structured_content, dict):
|
||||
return ToolResult(
|
||||
content=unstructured_result,
|
||||
structured_content=structured_content,
|
||||
)
|
||||
|
||||
except pydantic_core.PydanticSerializationError:
|
||||
pass
|
||||
|
||||
return ToolResult(content=unstructured_result)
|
||||
|
||||
wrap_result = self.output_schema.get("x-fastmcp-wrap-result")
|
||||
|
||||
return ToolResult(
|
||||
content=unstructured_result,
|
||||
structured_content=structured_output,
|
||||
structured_content={"result": result} if wrap_result else result,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -478,98 +488,69 @@ class ParsedFunction:
|
|||
)
|
||||
|
||||
|
||||
def _serialize_with_fallback(
|
||||
result: Any, serializer: ToolResultSerializerType | None = None
|
||||
) -> str:
|
||||
if serializer is not None:
|
||||
try:
|
||||
return serializer(result)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"Error serializing tool result: %s",
|
||||
e,
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
return default_serializer(result)
|
||||
|
||||
|
||||
def _convert_to_single_content_block(
|
||||
item: Any,
|
||||
serializer: ToolResultSerializerType | None = None,
|
||||
) -> ContentBlock:
|
||||
if isinstance(item, ContentBlock):
|
||||
return item
|
||||
|
||||
if isinstance(item, Image):
|
||||
return item.to_image_content()
|
||||
|
||||
if isinstance(item, Audio):
|
||||
return item.to_audio_content()
|
||||
|
||||
if isinstance(item, File):
|
||||
return item.to_resource_content()
|
||||
|
||||
if isinstance(item, str):
|
||||
return TextContent(type="text", text=item)
|
||||
|
||||
return TextContent(type="text", text=_serialize_with_fallback(item, serializer))
|
||||
|
||||
|
||||
def _convert_to_content(
|
||||
result: Any,
|
||||
serializer: Callable[[Any], str] | None = None,
|
||||
_process_as_single_item: bool = False,
|
||||
serializer: ToolResultSerializerType | None = None,
|
||||
) -> list[ContentBlock]:
|
||||
"""Convert a result to a sequence of content objects."""
|
||||
|
||||
if result is None:
|
||||
return []
|
||||
|
||||
if isinstance(result, ContentBlock):
|
||||
return [result]
|
||||
if not isinstance(result, (list | tuple)):
|
||||
return [_convert_to_single_content_block(result, serializer)]
|
||||
|
||||
if isinstance(result, Image):
|
||||
return [result.to_image_content()]
|
||||
# If all items are ContentBlocks, return them as is
|
||||
if all(isinstance(item, ContentBlock) for item in result):
|
||||
return result
|
||||
|
||||
elif isinstance(result, Audio):
|
||||
return [result.to_audio_content()]
|
||||
# If any item is a ContentBlock, convert non-ContentBlock items to TextContent
|
||||
# without aggregating them
|
||||
if any(isinstance(item, ContentBlock) for item in result):
|
||||
return [
|
||||
_convert_to_single_content_block(item, serializer)
|
||||
if not isinstance(item, ContentBlock)
|
||||
else item
|
||||
for item in result
|
||||
]
|
||||
|
||||
elif isinstance(result, File):
|
||||
return [result.to_resource_content()]
|
||||
|
||||
if isinstance(result, list | tuple) and not _process_as_single_item:
|
||||
# if the result is a list, then it could either be a list of MCP types,
|
||||
# or a "regular" list that the tool is returning, or a mix of both.
|
||||
#
|
||||
# Group adjacent non-MCP types together while preserving order
|
||||
|
||||
content_items = []
|
||||
non_mcp_batch = []
|
||||
|
||||
def flush_non_mcp_batch():
|
||||
"""Convert accumulated non-MCP items to a single TextContent block."""
|
||||
if non_mcp_batch:
|
||||
if len(non_mcp_batch) == 1:
|
||||
# Single item - convert directly to avoid combining when not needed
|
||||
content_items.extend(
|
||||
_convert_to_content(
|
||||
non_mcp_batch[0],
|
||||
serializer=serializer,
|
||||
_process_as_single_item=True,
|
||||
)
|
||||
)
|
||||
else:
|
||||
# Multiple items - combine into a single text block
|
||||
combined_text = ""
|
||||
for item in non_mcp_batch:
|
||||
if isinstance(item, str):
|
||||
combined_text += item
|
||||
else:
|
||||
if serializer is None:
|
||||
combined_text += default_serializer(item)
|
||||
else:
|
||||
try:
|
||||
combined_text += serializer(item)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"Error serializing tool result: %s",
|
||||
e,
|
||||
exc_info=True,
|
||||
)
|
||||
combined_text += default_serializer(item)
|
||||
content_items.append(TextContent(type="text", text=combined_text))
|
||||
non_mcp_batch.clear()
|
||||
|
||||
for item in result:
|
||||
if isinstance(item, ContentBlock | Image | Audio | File):
|
||||
# Flush any accumulated non-MCP items first
|
||||
flush_non_mcp_batch()
|
||||
# Add the MCP item
|
||||
content_items.extend(_convert_to_content(item))
|
||||
else:
|
||||
# Accumulate non-MCP items
|
||||
non_mcp_batch.append(item)
|
||||
|
||||
# Flush any remaining non-MCP items
|
||||
flush_non_mcp_batch()
|
||||
|
||||
return content_items
|
||||
|
||||
if not isinstance(result, str):
|
||||
if serializer is None:
|
||||
result = default_serializer(result)
|
||||
else:
|
||||
try:
|
||||
result = serializer(result)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"Error serializing tool result: %s",
|
||||
e,
|
||||
exc_info=True,
|
||||
)
|
||||
result = default_serializer(result)
|
||||
|
||||
return [TextContent(type="text", text=result)]
|
||||
# If none of the items are ContentBlocks, aggregate all items into a single TextContent
|
||||
return [TextContent(type="text", text=_serialize_with_fallback(result, serializer))]
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ from pathlib import Path
|
|||
from typing import Annotated, Any, Literal
|
||||
|
||||
import pytest
|
||||
from inline_snapshot import snapshot
|
||||
from mcp import McpError
|
||||
from mcp.types import (
|
||||
AudioContent,
|
||||
|
|
@ -21,6 +22,7 @@ from pydantic import AnyUrl, BaseModel, Field, TypeAdapter
|
|||
from typing_extensions import TypedDict
|
||||
|
||||
from fastmcp import Client, Context, FastMCP
|
||||
from fastmcp.client.client import CallToolResult
|
||||
from fastmcp.client.transports import FastMCPTransport
|
||||
from fastmcp.exceptions import ToolError
|
||||
from fastmcp.prompts.prompt import Prompt, PromptMessage
|
||||
|
|
@ -187,7 +189,7 @@ class TestTools:
|
|||
result = await client.call_tool("list_tool", {})
|
||||
# Adjacent non-MCP list items are combined into single content block
|
||||
assert len(result.content) == 1
|
||||
assert result.content[0].text == "x2" # type: ignore[attr-defined]
|
||||
assert result.content[0].text == '["x",2]' # type: ignore[attr-defined]
|
||||
assert result.data == ["x", 2]
|
||||
|
||||
async def test_file_text_tool(self, tool_server: FastMCP):
|
||||
|
|
@ -1157,20 +1159,37 @@ class TestToolOutputSchema:
|
|||
result = await client.call_tool("mixed_output", {})
|
||||
|
||||
# Should have multiple content blocks
|
||||
assert len(result.content) >= 2
|
||||
|
||||
# Should have structured output with wrapped result
|
||||
expected_data = [
|
||||
"text message",
|
||||
{"structured": "data"},
|
||||
{
|
||||
"type": "text",
|
||||
"text": "direct MCP content",
|
||||
"annotations": None,
|
||||
"_meta": None,
|
||||
},
|
||||
]
|
||||
assert result.structured_content == {"result": expected_data}
|
||||
assert result == snapshot(
|
||||
CallToolResult(
|
||||
content=[
|
||||
TextContent(type="text", text="text message"),
|
||||
TextContent(type="text", text='{"structured":"data"}'),
|
||||
TextContent(type="text", text="direct MCP content"),
|
||||
],
|
||||
structured_content={
|
||||
"result": [
|
||||
"text message",
|
||||
{"structured": "data"},
|
||||
{
|
||||
"type": "text",
|
||||
"text": "direct MCP content",
|
||||
"annotations": None,
|
||||
"_meta": None,
|
||||
},
|
||||
]
|
||||
},
|
||||
data=[
|
||||
"text message",
|
||||
{"structured": "data"},
|
||||
{
|
||||
"type": "text",
|
||||
"text": "direct MCP content",
|
||||
"annotations": None,
|
||||
"_meta": None,
|
||||
},
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
async def test_output_schema_serialization_edge_cases(self):
|
||||
"""Test edge cases in output schema serialization."""
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ async def test_simple_echo():
|
|||
async with Client(mcp) as client:
|
||||
result = await client.call_tool_mcp("echo", {"text": "hello"})
|
||||
assert len(result.content) == 1
|
||||
assert result.content[0].text == "hello" # type: ignore[attr-defined]
|
||||
assert result.content[0].text == "hello"
|
||||
|
||||
|
||||
async def test_complex_inputs():
|
||||
|
|
@ -26,7 +26,7 @@ async def test_complex_inputs():
|
|||
)
|
||||
# Adjacent non-MCP list items are combined into single content
|
||||
assert len(result.content) == 1
|
||||
assert result.content[0].text == "bobalicecharlie" # type: ignore[attr-defined]
|
||||
assert result.content[0].text == '["bob","alice","charlie"]'
|
||||
|
||||
|
||||
async def test_desktop(monkeypatch):
|
||||
|
|
@ -37,12 +37,12 @@ async def test_desktop(monkeypatch):
|
|||
# Test the add function
|
||||
result = await client.call_tool_mcp("add", {"a": 1, "b": 2})
|
||||
assert len(result.content) == 1
|
||||
assert result.content[0].text == "3" # type: ignore[attr-defined]
|
||||
assert result.content[0].text == "3"
|
||||
|
||||
async with Client(mcp) as client:
|
||||
result = await client.read_resource(AnyUrl("greeting://rooter12"))
|
||||
assert len(result) == 1
|
||||
assert result[0].text == "Hello, rooter12!" # type: ignore[attr-defined]
|
||||
assert result[0].text == "Hello, rooter12!"
|
||||
|
||||
|
||||
async def test_echo():
|
||||
|
|
@ -52,19 +52,19 @@ async def test_echo():
|
|||
async with Client(mcp) as client:
|
||||
result = await client.call_tool_mcp("echo_tool", {"text": "hello"})
|
||||
assert len(result.content) == 1
|
||||
assert result.content[0].text == "hello" # type: ignore[attr-defined]
|
||||
assert result.content[0].text == "hello"
|
||||
|
||||
async with Client(mcp) as client:
|
||||
result = await client.read_resource(AnyUrl("echo://static"))
|
||||
assert len(result) == 1
|
||||
assert result[0].text == "Echo!" # type: ignore[attr-defined]
|
||||
assert result[0].text == "Echo!"
|
||||
|
||||
async with Client(mcp) as client:
|
||||
result = await client.read_resource(AnyUrl("echo://server42"))
|
||||
assert len(result) == 1
|
||||
assert result[0].text == "Echo: server42" # type: ignore[attr-defined]
|
||||
assert result[0].text == "Echo: server42"
|
||||
|
||||
async with Client(mcp) as client:
|
||||
result = await client.get_prompt("echo", {"text": "hello"})
|
||||
assert len(result.messages) == 1
|
||||
assert result.messages[0].content.text == "hello" # type: ignore[attr-defined]
|
||||
assert result.messages[0].content.text == "hello"
|
||||
|
|
|
|||
|
|
@ -1,4 +1,3 @@
|
|||
import json
|
||||
from dataclasses import dataclass
|
||||
from typing import Annotated, Any
|
||||
|
||||
|
|
@ -7,8 +6,10 @@ from dirty_equals import HasName
|
|||
from inline_snapshot import snapshot
|
||||
from mcp.types import (
|
||||
AudioContent,
|
||||
BlobResourceContents,
|
||||
EmbeddedResource,
|
||||
ImageContent,
|
||||
ResourceLink,
|
||||
TextContent,
|
||||
TextResourceContents,
|
||||
)
|
||||
|
|
@ -941,294 +942,215 @@ class TestToolFromFunctionOutputSchema:
|
|||
Tool.from_function(func, output_schema=schema)
|
||||
|
||||
|
||||
class SampleModel(BaseModel):
|
||||
x: int
|
||||
y: str
|
||||
|
||||
|
||||
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",
|
||||
@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
|
||||
|
||||
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",
|
||||
),
|
||||
),
|
||||
]
|
||||
)
|
||||
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 == '{"a":1,"b":2}'
|
||||
|
||||
def test_list_of_basic_types(self):
|
||||
"""Test that a list of basic types is converted to a single combined TextContent item."""
|
||||
result = _convert_to_content([1, "two", {"c": 3}])
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1 # Adjacent non-MCP types are combined
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert result[0].text == '1two{"c":3}' # All adjacent basic types combined
|
||||
|
||||
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
|
||||
|
||||
# Verify order is preserved: text, image, text
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert result[0].text == "hello"
|
||||
assert isinstance(result[1], ImageContent)
|
||||
assert result[1].data == "ZmFrZWltYWdlZGF0YQ=="
|
||||
assert isinstance(result[2], TextContent)
|
||||
assert result[2].text == '{"a":1}'
|
||||
|
||||
def test_mixed_text_and_images_preserve_order(self):
|
||||
"""Test that mixed text and images preserve their original order (GitHub issue #1656)."""
|
||||
img_obj1 = Image(data=b"imagedata1")
|
||||
img_obj2 = Image(data=b"imagedata2")
|
||||
|
||||
# Test the exact pattern from the issue: [text1, img1, text2, img2]
|
||||
result = _convert_to_content(["text1", img_obj1, "text2", img_obj2])
|
||||
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 4
|
||||
|
||||
# Verify exact order is preserved
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert result[0].text == "text1"
|
||||
|
||||
assert isinstance(result[1], ImageContent)
|
||||
assert result[1].data == "aW1hZ2VkYXRhMQ==" # base64 of "imagedata1"
|
||||
|
||||
assert isinstance(result[2], TextContent)
|
||||
assert result[2].text == "text2"
|
||||
|
||||
assert isinstance(result[3], ImageContent)
|
||||
assert result[3].data == "aW1hZ2VkYXRhMg==" # base64 of "imagedata2"
|
||||
|
||||
def test_adjacent_non_mcp_types_combined(self):
|
||||
"""Test strawgate's example: image, 'x', 'y', image should be 2 image blocks and 1 content block."""
|
||||
img_obj1 = Image(data=b"imagedata1")
|
||||
img_obj2 = Image(data=b"imagedata2")
|
||||
|
||||
# Test the exact pattern from strawgate's request: [image, 'x', 'y', image]
|
||||
result = _convert_to_content([img_obj1, "x", "y", img_obj2])
|
||||
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 3 # 2 image blocks + 1 combined text block
|
||||
|
||||
# First image
|
||||
assert isinstance(result[0], ImageContent)
|
||||
assert result[0].data == "aW1hZ2VkYXRhMQ==" # base64 of "imagedata1"
|
||||
|
||||
# Combined text content for adjacent 'x' and 'y'
|
||||
assert isinstance(result[1], TextContent)
|
||||
assert result[1].text == "xy" # Adjacent strings combined
|
||||
|
||||
# Second image
|
||||
assert isinstance(result[2], ImageContent)
|
||||
assert result[2].data == "aW1hZ2VkYXRhMg==" # base64 of "imagedata2"
|
||||
|
||||
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
|
||||
) # Adjacent non-MCP types are combined: hello (TextContent) + image + basic_data (single TextContent)
|
||||
|
||||
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 # hello + serialized basic_data
|
||||
assert image_content_count == 1
|
||||
|
||||
# Verify order: hello, image, serialized basic_data
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert result[0].text == "hello"
|
||||
assert isinstance(result[1], ImageContent)
|
||||
assert result[1].data == "ZmFrZWltYWdlZGF0YQ=="
|
||||
assert isinstance(result[2], TextContent)
|
||||
assert result[2].text == '[{"a":1},{"b":2}]' # basic_data serialized as JSON
|
||||
|
||||
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
|
||||
|
||||
# Verify order is preserved: text, audio, text
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert result[0].text == "hello"
|
||||
assert isinstance(result[1], AudioContent)
|
||||
assert result[1].data == "ZmFrZWF1ZGlvZGF0YQ=="
|
||||
assert isinstance(result[2], TextContent)
|
||||
assert result[2].text == '{"a":1}'
|
||||
|
||||
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
|
||||
|
||||
# Verify order is preserved: text, file, text
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert result[0].text == "hello"
|
||||
assert isinstance(result[1], EmbeddedResource)
|
||||
assert result[1].type == "resource"
|
||||
assert isinstance(result[2], TextContent)
|
||||
assert result[2].text == '{"a":1}'
|
||||
|
||||
embedded_item = result[1]
|
||||
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."""
|
||||
|
|
@ -1244,17 +1166,17 @@ class TestConvertResultToContent:
|
|||
assert isinstance(result[0], TextContent)
|
||||
assert result[0].text == "{}"
|
||||
|
||||
def test_with_custom_serializer(self):
|
||||
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 isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert result[0].text == "Serialized: {'a': 1}"
|
||||
|
||||
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."""
|
||||
|
|
@ -1268,56 +1190,10 @@ class TestConvertResultToContent:
|
|||
)
|
||||
|
||||
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 result == snapshot([TextContent(type="text", 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 == '[1,"two",{"c":3}]'
|
||||
|
||||
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},
|
||||
]
|
||||
|
||||
def test_single_element_list_preserves_structure(self):
|
||||
"""Test that single-element lists are converted to individual TextContent items."""
|
||||
|
||||
# Test with a single integer - now returns the integer as individual content
|
||||
result = _convert_to_content([1])
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert result[0].text == "1" # Individual item, not wrapped in list
|
||||
|
||||
# Test with a single string - now returns the string as individual content
|
||||
result = _convert_to_content(["hello"])
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert result[0].text == "hello" # Individual string, not wrapped in list
|
||||
|
||||
# Test with a single dict - now returns the dict as individual content
|
||||
result = _convert_to_content([{"a": 1}])
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert result[0].text == '{"a":1}' # Individual dict, not wrapped in list
|
||||
|
||||
|
||||
class TestAutomaticStructuredContent:
|
||||
"""Tests for automatic structured content generation based on return types."""
|
||||
|
|
@ -1441,10 +1317,30 @@ class TestAutomaticStructuredContent:
|
|||
|
||||
result = await tool.run({})
|
||||
|
||||
# Adjacent non-MCP types should be combined into single content block, no structured content
|
||||
assert len(result.content) == 1
|
||||
assert all(isinstance(item, TextContent) for item in result.content)
|
||||
assert result.content[0].text == "12345" # All numbers combined
|
||||
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):
|
||||
|
|
|
|||
|
|
@ -5,7 +5,8 @@ from typing import Annotated, Any
|
|||
|
||||
import pydantic_core
|
||||
import pytest
|
||||
from mcp.types import ImageContent
|
||||
from inline_snapshot import snapshot
|
||||
from mcp.types import ImageContent, TextContent
|
||||
from pydantic import BaseModel
|
||||
|
||||
from fastmcp import Context, FastMCP
|
||||
|
|
@ -572,8 +573,10 @@ class TestCallTools:
|
|||
|
||||
# Adjacent non-MCP list items are combined into single content block
|
||||
assert len(result.content) == 1
|
||||
assert result.content[0].text == "rexgertrude" # type: ignore[attr-defined]
|
||||
assert result.structured_content == {"result": ["rex", "gertrude"]}
|
||||
assert result.content == snapshot(
|
||||
[TextContent(type="text", text='["rex","gertrude"]')]
|
||||
)
|
||||
assert result.structured_content == snapshot({"result": ["rex", "gertrude"]})
|
||||
|
||||
async def test_call_tool_with_custom_serializer(self):
|
||||
"""Test that a custom serializer provided to FastMCP is used by tools."""
|
||||
|
|
@ -616,16 +619,22 @@ class TestCallTools:
|
|||
result = await manager.call_tool("get_data", {})
|
||||
# Adjacent non-MCP list items get combined with custom serializer applied to each
|
||||
assert len(result.content) == 1
|
||||
assert (
|
||||
result.content[0].text # type: ignore[attr-defined]
|
||||
== '{"key": "value", "number": 123}{"key": "value2", "number": 456}' # Adjacent items combined after individual serialization
|
||||
)
|
||||
assert result.structured_content == {
|
||||
"result": [
|
||||
{"key": "value", "number": 123},
|
||||
{"key": "value2", "number": 456},
|
||||
assert result.content == snapshot(
|
||||
[
|
||||
TextContent(
|
||||
type="text",
|
||||
text='CUSTOM:[{"key": "value", "number": 123}, {"key": "value2", "number": 456}]',
|
||||
)
|
||||
]
|
||||
}
|
||||
)
|
||||
assert result.structured_content == snapshot(
|
||||
{
|
||||
"result": [
|
||||
{"key": "value", "number": 123},
|
||||
{"key": "value2", "number": 456},
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
async def test_custom_serializer_fallback_on_error(self):
|
||||
"""Test that a broken custom serializer gracefully falls back."""
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue