mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-23 14:04:18 +02:00
Update all tests
This commit is contained in:
Jeremiah Lowin
parent
a7eb2da273
commit
db24c85359
12 changed files with 912 additions and 101 deletions
|
|
@ -450,8 +450,10 @@ class OpenAPITool(Tool):
|
|||
# Try to parse as JSON first
|
||||
try:
|
||||
result = response.json()
|
||||
if not isinstance(result, dict):
|
||||
result = {"result": result}
|
||||
return ToolResult(structured_content=result)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
except json.JSONDecodeError:
|
||||
return ToolResult(content=response.text)
|
||||
|
||||
except httpx.HTTPStatusError as e:
|
||||
|
|
|
|||
|
|
@ -240,6 +240,13 @@ class FunctionTool(Tool):
|
|||
output_schema = None
|
||||
# Note: explicit schemas (dict) are used as-is without auto-wrapping
|
||||
|
||||
# Validate that explicit schemas are object type for structured content
|
||||
if output_schema is not None and isinstance(output_schema, dict):
|
||||
if output_schema.get("type") != "object":
|
||||
raise ValueError(
|
||||
f'Output schemas must have "type" set to "object" due to MCP spec limitations. Received: {output_schema!r}'
|
||||
)
|
||||
|
||||
return cls(
|
||||
fn=parsed_fn.fn,
|
||||
name=name or parsed_fn.name,
|
||||
|
|
@ -264,6 +271,7 @@ class FunctionTool(Tool):
|
|||
|
||||
type_adapter = get_cached_typeadapter(self.fn)
|
||||
result = type_adapter.validate_python(arguments)
|
||||
|
||||
if inspect.isawaitable(result):
|
||||
result = await result
|
||||
|
||||
|
|
@ -275,6 +283,7 @@ class FunctionTool(Tool):
|
|||
# Handle structured content based on output schema
|
||||
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
|
||||
|
|
@ -354,26 +363,43 @@ class ParsedFunction:
|
|||
output_schema = None
|
||||
output_type = inspect.signature(fn).return_annotation
|
||||
|
||||
# there are a variety of types that we don't want to attempt to
|
||||
# serialize because they are either used by FastMCP internally,
|
||||
# or are MCP content types that explicitly don't form structured
|
||||
# content. By replacing them with an explicitly unserializable type,
|
||||
# we ensure that no output schema is automatically generated.
|
||||
if output_type not in (inspect._empty, None, Any, ...):
|
||||
# there are a variety of types that we don't want to attempt to
|
||||
# serialize because they are either used by FastMCP internally,
|
||||
# or are MCP content types that explicitly don't form structured
|
||||
# content. By replacing them with an explicitly unserializable type,
|
||||
# we ensure that no output schema is automatically generated.
|
||||
output_type = replace_type(
|
||||
output_type,
|
||||
{
|
||||
t: _UnserializableType
|
||||
for t in (
|
||||
Image,
|
||||
Audio,
|
||||
File,
|
||||
ToolResult,
|
||||
mcp.types.TextContent,
|
||||
mcp.types.ImageContent,
|
||||
mcp.types.AudioContent,
|
||||
mcp.types.ResourceLink,
|
||||
mcp.types.EmbeddedResource,
|
||||
)
|
||||
},
|
||||
)
|
||||
|
||||
output_type = replace_type(
|
||||
output_type,
|
||||
{
|
||||
inspect._empty: _UnserializableType,
|
||||
Image: _UnserializableType,
|
||||
Audio: _UnserializableType,
|
||||
File: _UnserializableType,
|
||||
ToolResult: _UnserializableType,
|
||||
mcp.types.TextContent: _UnserializableType,
|
||||
mcp.types.ImageContent: _UnserializableType,
|
||||
mcp.types.AudioContent: _UnserializableType,
|
||||
mcp.types.ResourceLink: _UnserializableType,
|
||||
mcp.types.EmbeddedResource: _UnserializableType,
|
||||
},
|
||||
try:
|
||||
output_type_adapter = get_cached_typeadapter(output_type)
|
||||
output_schema = output_type_adapter.json_schema()
|
||||
except PydanticSchemaGenerationError as e:
|
||||
if "_UnserializableType" not in str(e):
|
||||
logger.debug(f"Unable to generate schema for type {output_type!r}")
|
||||
|
||||
return cls(
|
||||
fn=fn,
|
||||
name=fn_name,
|
||||
description=fn_doc,
|
||||
input_schema=input_schema,
|
||||
output_schema=output_schema or None,
|
||||
)
|
||||
|
||||
try:
|
||||
|
|
@ -388,7 +414,7 @@ class ParsedFunction:
|
|||
name=fn_name,
|
||||
description=fn_doc,
|
||||
input_schema=input_schema,
|
||||
output_schema=output_schema,
|
||||
output_schema=output_schema or None,
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -269,9 +269,32 @@ class TransformedTool(Tool):
|
|||
try:
|
||||
result = await self.fn(**arguments)
|
||||
|
||||
# If transform function returns ToolResult, use it directly
|
||||
# If transform function returns ToolResult, respect our output_schema setting
|
||||
if isinstance(result, ToolResult):
|
||||
return result
|
||||
if self.output_schema is None:
|
||||
# Check if this is from a custom function that returns ToolResult
|
||||
import inspect
|
||||
|
||||
return_annotation = inspect.signature(self.fn).return_annotation
|
||||
if return_annotation is ToolResult:
|
||||
# Custom function returns ToolResult - preserve its content
|
||||
return result
|
||||
else:
|
||||
# Forwarded call with disabled schema - strip structured content
|
||||
return ToolResult(
|
||||
content=result.content,
|
||||
structured_content=None,
|
||||
)
|
||||
elif self.output_schema.get(
|
||||
"type"
|
||||
) != "object" and not self.output_schema.get("x-fastmcp-wrap-result"):
|
||||
# Non-object explicit schemas disable structured content
|
||||
return ToolResult(
|
||||
content=result.content,
|
||||
structured_content=None,
|
||||
)
|
||||
else:
|
||||
return result
|
||||
|
||||
# Otherwise convert to content and create ToolResult with proper structured content
|
||||
from fastmcp.tools.tool import _convert_to_content
|
||||
|
|
@ -283,8 +306,11 @@ class TransformedTool(Tool):
|
|||
# Handle structured content based on output schema
|
||||
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:
|
||||
# Object schemas - use result directly
|
||||
# User is responsible for returning dict-compatible data
|
||||
structured_output = result
|
||||
else:
|
||||
structured_output = None
|
||||
|
|
@ -381,7 +407,16 @@ class TransformedTool(Tool):
|
|||
parsed_fn = ParsedFunction.from_function(transform_fn, validate=False)
|
||||
final_output_schema = _wrap_schema_if_needed(parsed_fn.output_schema)
|
||||
if final_output_schema is None:
|
||||
final_output_schema = tool.output_schema
|
||||
# Check if function returns ToolResult - if so, don't fall back to parent
|
||||
import inspect
|
||||
|
||||
return_annotation = inspect.signature(
|
||||
transform_fn
|
||||
).return_annotation
|
||||
if return_annotation is ToolResult:
|
||||
final_output_schema = None
|
||||
else:
|
||||
final_output_schema = tool.output_schema
|
||||
else:
|
||||
final_output_schema = tool.output_schema
|
||||
|
||||
|
|
|
|||
|
|
@ -169,8 +169,17 @@ def json_schema_to_type(
|
|||
"""
|
||||
# Always use the top-level schema for references
|
||||
if schema.get("type") == "object":
|
||||
# If no properties defined but additionalProperties is True, return dict[str, Any]
|
||||
if not schema.get("properties") and schema.get("additionalProperties") is True:
|
||||
# If no properties defined but has additionalProperties, return typed dict
|
||||
if not schema.get("properties") and schema.get("additionalProperties"):
|
||||
additional_props = schema["additionalProperties"]
|
||||
if additional_props is True:
|
||||
return dict[str, Any] # type: ignore - additionalProperties: true means dict[str, Any]
|
||||
else:
|
||||
# Handle typed dictionaries like dict[str, str]
|
||||
value_type = _schema_to_type(additional_props, schemas=schema)
|
||||
return dict[str, value_type] # type: ignore
|
||||
# If no properties and no additionalProperties, default to dict[str, Any] for safety
|
||||
elif not schema.get("properties") and not schema.get("additionalProperties"):
|
||||
return dict[str, Any] # type: ignore
|
||||
# If has properties AND additionalProperties is True, use Pydantic BaseModel
|
||||
elif schema.get("properties") and schema.get("additionalProperties") is True:
|
||||
|
|
@ -328,6 +337,43 @@ def _schema_to_type(
|
|||
if "enum" in schema:
|
||||
return _create_enum(f"Enum_{len(_classes)}", schema["enum"])
|
||||
|
||||
# Handle anyOf unions
|
||||
if "anyOf" in schema:
|
||||
types: list[type | Any] = []
|
||||
for subschema in schema["anyOf"]:
|
||||
# Special handling for dict-like objects in unions
|
||||
if (
|
||||
subschema.get("type") == "object"
|
||||
and not subschema.get("properties")
|
||||
and subschema.get("additionalProperties")
|
||||
):
|
||||
# This is a dict type, handle it directly
|
||||
additional_props = subschema["additionalProperties"]
|
||||
if additional_props is True:
|
||||
types.append(dict[str, Any]) # type: ignore
|
||||
else:
|
||||
value_type = _schema_to_type(additional_props, schemas)
|
||||
types.append(dict[str, value_type]) # type: ignore
|
||||
else:
|
||||
types.append(_schema_to_type(subschema, schemas))
|
||||
|
||||
# Check if one of the types is None (null)
|
||||
has_null = type(None) in types
|
||||
types = [t for t in types if t is not type(None)]
|
||||
|
||||
if len(types) == 0:
|
||||
return type(None)
|
||||
elif len(types) == 1:
|
||||
if has_null:
|
||||
return Optional[types[0]] # type: ignore # noqa: UP007
|
||||
else:
|
||||
return types[0]
|
||||
else:
|
||||
if has_null:
|
||||
return Union[tuple(types + [type(None)])] # type: ignore # noqa: UP007
|
||||
else:
|
||||
return Union[tuple(types)] # type: ignore # noqa: UP007
|
||||
|
||||
schema_type = schema.get("type")
|
||||
if not schema_type:
|
||||
return Any
|
||||
|
|
|
|||
|
|
@ -45,13 +45,8 @@ async def echo_tool(arg1: str) -> str:
|
|||
def echo_tool_result_factory(arg1: str) -> CallToolRequestResult:
|
||||
"""A tool that returns a result based on the input arguments."""
|
||||
return CallToolRequestResult(
|
||||
isError=True,
|
||||
content=[
|
||||
TextContent(
|
||||
text="Output validation error: outputSchema defined but no structured output returned",
|
||||
type="text",
|
||||
)
|
||||
],
|
||||
isError=False,
|
||||
content=[TextContent(text=f"{arg1}", type="text")],
|
||||
tool="echo_tool",
|
||||
arguments={"arg1": arg1},
|
||||
)
|
||||
|
|
@ -64,13 +59,8 @@ async def no_return_tool(arg1: str) -> None:
|
|||
def no_return_tool_result_factory(arg1: str) -> CallToolRequestResult:
|
||||
"""A tool that returns a result based on the input arguments."""
|
||||
return CallToolRequestResult(
|
||||
isError=True,
|
||||
content=[
|
||||
TextContent(
|
||||
text="Output validation error: outputSchema defined but no structured output returned",
|
||||
type="text",
|
||||
)
|
||||
],
|
||||
isError=False,
|
||||
content=[],
|
||||
tool="no_return_tool",
|
||||
arguments={"arg1": arg1},
|
||||
)
|
||||
|
|
|
|||
|
|
@ -86,9 +86,8 @@ async def test_http_headers_tool_shttp(shttp_server: str):
|
|||
)
|
||||
) as client:
|
||||
result = await client.call_tool("get_headers_tool")
|
||||
json_result = json.loads(result[0].text) # type: ignore[attr-defined]
|
||||
assert "x-demo-header" in json_result
|
||||
assert json_result["x-demo-header"] == "ABC"
|
||||
assert "x-demo-header" in result.data
|
||||
assert result.data["x-demo-header"] == "ABC"
|
||||
|
||||
|
||||
async def test_http_headers_tool_sse(sse_server: str):
|
||||
|
|
@ -96,9 +95,8 @@ async def test_http_headers_tool_sse(sse_server: str):
|
|||
transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
|
||||
) as client:
|
||||
result = await client.call_tool("get_headers_tool")
|
||||
json_result = json.loads(result[0].text) # type: ignore[attr-defined]
|
||||
assert "x-demo-header" in json_result
|
||||
assert json_result["x-demo-header"] == "ABC"
|
||||
assert "x-demo-header" in result.data
|
||||
assert result.data["x-demo-header"] == "ABC"
|
||||
|
||||
|
||||
async def test_http_headers_prompt_shttp(shttp_server: str):
|
||||
|
|
|
|||
|
|
@ -278,7 +278,7 @@ async def test_call_nested_imported_tool():
|
|||
|
||||
async with Client(main_app) as client:
|
||||
result = await client.call_tool("service_provider_compute", {"input": 21})
|
||||
assert result.data == "42"
|
||||
assert result.data == 42
|
||||
|
||||
|
||||
async def test_import_with_proxy_tools():
|
||||
|
|
|
|||
|
|
@ -127,8 +127,9 @@ class TestToolDecorator:
|
|||
def add(x: int, y: int) -> int:
|
||||
return x + y
|
||||
|
||||
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
|
||||
assert result[0].text == "3" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("add", {"x": 1, "y": 2})
|
||||
assert result.data == 3
|
||||
|
||||
async def test_tool_decorator_without_parentheses(self):
|
||||
"""Test that @tool decorator works without parentheses."""
|
||||
|
|
@ -144,8 +145,9 @@ class TestToolDecorator:
|
|||
assert "add" in tools
|
||||
|
||||
# Verify it can be called
|
||||
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
|
||||
assert result[0].text == "3" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("add", {"x": 1, "y": 2})
|
||||
assert result.data == 3
|
||||
|
||||
async def test_tool_decorator_with_name(self):
|
||||
mcp = FastMCP()
|
||||
|
|
@ -154,8 +156,9 @@ class TestToolDecorator:
|
|||
def add(x: int, y: int) -> int:
|
||||
return x + y
|
||||
|
||||
result = await mcp._mcp_call_tool("custom-add", {"x": 1, "y": 2})
|
||||
assert result[0].text == "3" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("custom-add", {"x": 1, "y": 2})
|
||||
assert result.data == 3
|
||||
|
||||
async def test_tool_decorator_with_description(self):
|
||||
mcp = FastMCP()
|
||||
|
|
@ -181,8 +184,9 @@ class TestToolDecorator:
|
|||
|
||||
obj = MyClass(10)
|
||||
mcp.add_tool(Tool.from_function(obj.add))
|
||||
result = await mcp._mcp_call_tool("add", {"y": 2})
|
||||
assert result[0].text == "12" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("add", {"y": 2})
|
||||
assert result.data == 12
|
||||
|
||||
async def test_tool_decorator_classmethod(self):
|
||||
mcp = FastMCP()
|
||||
|
|
@ -195,8 +199,9 @@ class TestToolDecorator:
|
|||
return cls.x + y
|
||||
|
||||
mcp.add_tool(Tool.from_function(MyClass.add))
|
||||
result = await mcp._mcp_call_tool("add", {"y": 2})
|
||||
assert result[0].text == "12" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("add", {"y": 2})
|
||||
assert result.data == 12
|
||||
|
||||
async def test_tool_decorator_staticmethod(self):
|
||||
mcp = FastMCP()
|
||||
|
|
@ -207,8 +212,9 @@ class TestToolDecorator:
|
|||
def add(x: int, y: int) -> int:
|
||||
return x + y
|
||||
|
||||
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
|
||||
assert result[0].text == "3" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("add", {"x": 1, "y": 2})
|
||||
assert result.data == 3
|
||||
|
||||
async def test_tool_decorator_async_function(self):
|
||||
mcp = FastMCP()
|
||||
|
|
@ -217,8 +223,9 @@ class TestToolDecorator:
|
|||
async def add(x: int, y: int) -> int:
|
||||
return x + y
|
||||
|
||||
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
|
||||
assert result[0].text == "3" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("add", {"x": 1, "y": 2})
|
||||
assert result.data == 3
|
||||
|
||||
async def test_tool_decorator_classmethod_error(self):
|
||||
mcp = FastMCP()
|
||||
|
|
@ -242,8 +249,9 @@ class TestToolDecorator:
|
|||
return cls.x + y
|
||||
|
||||
mcp.add_tool(Tool.from_function(MyClass.add))
|
||||
result = await mcp._mcp_call_tool("add", {"y": 2})
|
||||
assert result[0].text == "12" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("add", {"y": 2})
|
||||
assert result.data == 12
|
||||
|
||||
async def test_tool_decorator_staticmethod_async_function(self):
|
||||
mcp = FastMCP()
|
||||
|
|
@ -254,8 +262,9 @@ class TestToolDecorator:
|
|||
return x + y
|
||||
|
||||
mcp.add_tool(Tool.from_function(MyClass.add))
|
||||
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
|
||||
assert result[0].text == "3" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("add", {"x": 1, "y": 2})
|
||||
assert result.data == 3
|
||||
|
||||
async def test_tool_decorator_staticmethod_order(self):
|
||||
"""Test that the recommended decorator order works for static methods"""
|
||||
|
|
@ -268,8 +277,9 @@ class TestToolDecorator:
|
|||
return x + y
|
||||
|
||||
# Test that the recommended order works
|
||||
result = await mcp._mcp_call_tool("add_v1", {"x": 1, "y": 2})
|
||||
assert result[0].text == "3" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("add_v1", {"x": 1, "y": 2})
|
||||
assert result.data == 3
|
||||
|
||||
async def test_tool_decorator_with_tags(self):
|
||||
"""Test that the tool decorator properly sets tags."""
|
||||
|
|
@ -299,8 +309,9 @@ class TestToolDecorator:
|
|||
assert "custom_multiply" in tools
|
||||
|
||||
# Call the tool by its custom name
|
||||
result = await mcp._mcp_call_tool("custom_multiply", {"a": 5, "b": 3})
|
||||
assert result[0].text == "15" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("custom_multiply", {"a": 5, "b": 3})
|
||||
assert result.data == 15
|
||||
|
||||
# Original name should not be registered
|
||||
assert "multiply" not in tools
|
||||
|
|
@ -354,8 +365,9 @@ class TestToolDecorator:
|
|||
assert tools["direct_call_tool"] is result_fn
|
||||
|
||||
# Verify it can be called
|
||||
result = await mcp._mcp_call_tool("direct_call_tool", {"x": 5, "y": 3})
|
||||
assert result[0].text == "8" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("direct_call_tool", {"x": 5, "y": 3})
|
||||
assert result.data == 8
|
||||
|
||||
async def test_tool_decorator_with_string_name(self):
|
||||
"""Test that @tool("custom_name") syntax works correctly."""
|
||||
|
|
@ -372,8 +384,9 @@ class TestToolDecorator:
|
|||
assert "my_function" not in tools # Original name should not be registered
|
||||
|
||||
# Verify it can be called
|
||||
result = await mcp._mcp_call_tool("string_named_tool", {"x": 42})
|
||||
assert result[0].text == "Result: 42" # type: ignore[attr-defined]
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("string_named_tool", {"x": 42})
|
||||
assert result.data == "Result: 42"
|
||||
|
||||
async def test_tool_decorator_conflicting_names_error(self):
|
||||
"""Test that providing both positional and keyword name raises an error."""
|
||||
|
|
@ -391,11 +404,13 @@ class TestToolDecorator:
|
|||
async def test_tool_decorator_with_output_schema(self):
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(output_schema={"type": "integer"})
|
||||
def my_function(x: int) -> str:
|
||||
return f"Result: {x}"
|
||||
with pytest.raises(
|
||||
ValueError, match='Output schemas must have "type" set to "object"'
|
||||
):
|
||||
|
||||
assert my_function.output_schema == {"type": "integer"}
|
||||
@mcp.tool(output_schema={"type": "integer"})
|
||||
def my_function(x: int) -> str:
|
||||
return f"Result: {x}"
|
||||
|
||||
|
||||
class TestResourceDecorator:
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import uuid
|
|||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
from typing import Annotated, Literal
|
||||
from typing import Annotated, Any, Literal
|
||||
|
||||
import pytest
|
||||
from mcp import McpError
|
||||
|
|
@ -946,6 +946,198 @@ class TestToolOutputSchema:
|
|||
assert result.structured_content == {"message": "Hello, world!"}
|
||||
assert result.data == {"message": "Hello, world!"}
|
||||
|
||||
async def test_output_schema_false_full_handshake(self):
|
||||
"""Test that output_schema=False works through full client/server handshake."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(output_schema=False)
|
||||
def simple_tool() -> dict[str, str]:
|
||||
return {"message": "Hello from disabled schema"}
|
||||
|
||||
async with Client(mcp) as client:
|
||||
# List tools and verify output schema is None
|
||||
tools = await client.list_tools()
|
||||
tool = next(t for t in tools if t.name == "simple_tool")
|
||||
assert tool.outputSchema is None
|
||||
|
||||
# Call tool and verify no structured content
|
||||
result = await client.call_tool("simple_tool", {})
|
||||
assert result.structured_content is None
|
||||
assert result.data is None
|
||||
assert json.loads(result.content[0].text) == {
|
||||
"message": "Hello from disabled schema"
|
||||
} # type: ignore[attr-defined]
|
||||
|
||||
async def test_output_schema_explicit_object_full_handshake(self):
|
||||
"""Test explicit object output schema through full client/server handshake."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(
|
||||
output_schema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"greeting": {"type": "string"},
|
||||
"count": {"type": "integer"},
|
||||
},
|
||||
"required": ["greeting"],
|
||||
}
|
||||
)
|
||||
def explicit_tool() -> dict[str, Any]:
|
||||
return {"greeting": "Hello", "count": 42}
|
||||
|
||||
async with Client(mcp) as client:
|
||||
# List tools and verify exact schema is preserved
|
||||
tools = await client.list_tools()
|
||||
tool = next(t for t in tools if t.name == "explicit_tool")
|
||||
expected_schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"greeting": {"type": "string"},
|
||||
"count": {"type": "integer"},
|
||||
},
|
||||
"required": ["greeting"],
|
||||
}
|
||||
assert tool.outputSchema == expected_schema
|
||||
|
||||
# Call tool and verify structured content matches return value directly
|
||||
result = await client.call_tool("explicit_tool", {})
|
||||
assert result.structured_content == {"greeting": "Hello", "count": 42}
|
||||
# Client deserializes according to schema, so check fields
|
||||
assert result.data.greeting == "Hello" # type: ignore[attr-defined]
|
||||
assert result.data.count == 42 # type: ignore[attr-defined]
|
||||
|
||||
async def test_output_schema_wrapped_primitive_full_handshake(self):
|
||||
"""Test wrapped primitive output schema through full client/server handshake."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool
|
||||
def primitive_tool() -> str:
|
||||
return "Hello, primitives!"
|
||||
|
||||
async with Client(mcp) as client:
|
||||
# List tools and verify schema shows wrapped structure
|
||||
tools = await client.list_tools()
|
||||
tool = next(t for t in tools if t.name == "primitive_tool")
|
||||
expected_schema = {
|
||||
"type": "object",
|
||||
"properties": {"result": {"type": "string"}},
|
||||
"x-fastmcp-wrap-result": True,
|
||||
}
|
||||
assert tool.outputSchema == expected_schema
|
||||
|
||||
# Call tool and verify structured content is wrapped
|
||||
result = await client.call_tool("primitive_tool", {})
|
||||
assert result.structured_content == {"result": "Hello, primitives!"}
|
||||
assert result.data == "Hello, primitives!" # Client unwraps for convenience
|
||||
|
||||
async def test_output_schema_complex_type_full_handshake(self):
|
||||
"""Test complex type output schema through full client/server handshake."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool
|
||||
def complex_tool() -> list[dict[str, int]]:
|
||||
return [{"a": 1, "b": 2}, {"c": 3, "d": 4}]
|
||||
|
||||
async with Client(mcp) as client:
|
||||
# List tools and verify schema shows wrapped array
|
||||
tools = await client.list_tools()
|
||||
tool = next(t for t in tools if t.name == "complex_tool")
|
||||
expected_inner_schema = TypeAdapter(list[dict[str, int]]).json_schema()
|
||||
expected_schema = {
|
||||
"type": "object",
|
||||
"properties": {"result": expected_inner_schema},
|
||||
"x-fastmcp-wrap-result": True,
|
||||
}
|
||||
assert tool.outputSchema == expected_schema
|
||||
|
||||
# Call tool and verify structured content is wrapped
|
||||
result = await client.call_tool("complex_tool", {})
|
||||
expected_data = [{"a": 1, "b": 2}, {"c": 3, "d": 4}]
|
||||
assert result.structured_content == {"result": expected_data}
|
||||
# Client deserializes - just verify we got data back
|
||||
assert result.data is not None
|
||||
|
||||
async def test_output_schema_dataclass_full_handshake(self):
|
||||
"""Test dataclass output schema through full client/server handshake."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@dataclass
|
||||
class User:
|
||||
name: str
|
||||
age: int
|
||||
|
||||
@mcp.tool
|
||||
def dataclass_tool() -> User:
|
||||
return User(name="Alice", age=30)
|
||||
|
||||
async with Client(mcp) as client:
|
||||
# List tools and verify schema is object type (not wrapped)
|
||||
tools = await client.list_tools()
|
||||
tool = next(t for t in tools if t.name == "dataclass_tool")
|
||||
expected_schema = TypeAdapter(User).json_schema()
|
||||
assert tool.outputSchema == expected_schema
|
||||
assert "x-fastmcp-wrap-result" not in tool.outputSchema
|
||||
|
||||
# Call tool and verify structured content is direct
|
||||
result = await client.call_tool("dataclass_tool", {})
|
||||
assert result.structured_content == {"name": "Alice", "age": 30}
|
||||
# Client deserializes according to schema
|
||||
assert result.data.name == "Alice" # type: ignore[attr-defined]
|
||||
assert result.data.age == 30 # type: ignore[attr-defined]
|
||||
|
||||
async def test_output_schema_mixed_content_types(self):
|
||||
"""Test tools with mixed content and output schemas."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool
|
||||
def mixed_output() -> list[Any]:
|
||||
# Return mixed content that includes MCP types and regular data
|
||||
return [
|
||||
"text message",
|
||||
{"structured": "data"},
|
||||
TextContent(type="text", text="direct MCP content"),
|
||||
]
|
||||
|
||||
async with Client(mcp) as client:
|
||||
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}
|
||||
|
||||
async def test_output_schema_serialization_edge_cases(self):
|
||||
"""Test edge cases in output schema serialization."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool
|
||||
def edge_case_tool() -> tuple[int, str]:
|
||||
return (42, "hello")
|
||||
|
||||
async with Client(mcp) as client:
|
||||
# Verify tuple gets proper schema
|
||||
tools = await client.list_tools()
|
||||
tool = next(t for t in tools if t.name == "edge_case_tool")
|
||||
|
||||
# Tuples should be wrapped since they're not object type
|
||||
assert "x-fastmcp-wrap-result" in tool.outputSchema
|
||||
|
||||
result = await client.call_tool("edge_case_tool", {})
|
||||
# Should be wrapped with result key
|
||||
assert result.structured_content == {"result": [42, "hello"]}
|
||||
assert result.data == [42, "hello"]
|
||||
|
||||
|
||||
class TestToolContextInjection:
|
||||
"""Test context injection in tools."""
|
||||
|
|
|
|||
|
|
@ -263,14 +263,16 @@ class TestToolFromFunctionOutputSchema:
|
|||
@pytest.mark.parametrize(
|
||||
"annotation",
|
||||
[
|
||||
None,
|
||||
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],
|
||||
|
|
@ -304,7 +306,6 @@ class TestToolFromFunctionOutputSchema:
|
|||
@pytest.mark.parametrize(
|
||||
"annotation",
|
||||
[
|
||||
Any,
|
||||
AnyUrl,
|
||||
Annotated[int, Field(ge=1)],
|
||||
Annotated[int, Field(ge=1)],
|
||||
|
|
@ -317,17 +318,26 @@ class TestToolFromFunctionOutputSchema:
|
|||
tool = Tool.from_function(func)
|
||||
base_schema = TypeAdapter(annotation).json_schema()
|
||||
|
||||
# Special case for Any type - it generates an empty schema and doesn't get wrapped
|
||||
if annotation is Any:
|
||||
assert tool.output_schema == base_schema # Should be {}
|
||||
else:
|
||||
# All other non-object types get wrapped, including complex constrained types
|
||||
expected_schema = {
|
||||
"type": "object",
|
||||
"properties": {"result": base_schema},
|
||||
"x-fastmcp-wrap-result": True,
|
||||
}
|
||||
assert tool.output_schema == expected_schema
|
||||
expected_schema = {
|
||||
"type": "object",
|
||||
"properties": {"result": base_schema},
|
||||
"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",
|
||||
|
|
@ -413,7 +423,7 @@ class TestToolFromFunctionOutputSchema:
|
|||
return {"a": 1, "b": 2}
|
||||
|
||||
# Provide a custom output schema that differs from the inferred one
|
||||
custom_schema = {"type": "string", "description": "Custom schema"}
|
||||
custom_schema = {"type": "object", "description": "Custom schema"}
|
||||
|
||||
tool = Tool.from_function(func, output_schema=custom_schema)
|
||||
assert tool.output_schema == custom_schema
|
||||
|
|
@ -445,7 +455,10 @@ class TestToolFromFunctionOutputSchema:
|
|||
return Unserializable(data="test")
|
||||
|
||||
# Provide a custom output schema even though the annotation is unserializable
|
||||
custom_schema = {"type": "array", "items": {"type": "string"}}
|
||||
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
|
||||
|
|
@ -457,7 +470,10 @@ class TestToolFromFunctionOutputSchema:
|
|||
return "hello"
|
||||
|
||||
# Provide a custom output schema even though there's no return annotation
|
||||
custom_schema = {"type": "number", "minimum": 0}
|
||||
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
|
||||
|
|
@ -486,7 +502,7 @@ class TestToolFromFunctionOutputSchema:
|
|||
return "hello"
|
||||
|
||||
# Provide a custom output schema that differs from the inferred union schema
|
||||
custom_schema = {"type": "boolean"}
|
||||
custom_schema = {"type": "object", "properties": {"flag": {"type": "boolean"}}}
|
||||
|
||||
tool = Tool.from_function(func, output_schema=custom_schema)
|
||||
assert tool.output_schema == custom_schema
|
||||
|
|
@ -504,11 +520,215 @@ class TestToolFromFunctionOutputSchema:
|
|||
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"}}
|
||||
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_disables_structured_content(self):
|
||||
"""Test that output_schema=False disables structured content generation."""
|
||||
|
||||
def func() -> dict[str, str]:
|
||||
return {"message": "Hello, world!"}
|
||||
|
||||
tool = Tool.from_function(func, output_schema=False)
|
||||
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 == '{\n "message": "Hello, world!"\n}'
|
||||
|
||||
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"
|
||||
|
||||
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)
|
||||
expected_schema = {
|
||||
"type": "object",
|
||||
"properties": {"result": {"type": "integer"}},
|
||||
"x-fastmcp-wrap-result": True,
|
||||
}
|
||||
assert tool.output_schema == expected_schema
|
||||
|
||||
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 "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 == '{\n "value": 42\n}'
|
||||
|
||||
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 "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)
|
||||
expected_schema = {
|
||||
"type": "object",
|
||||
"properties": {"result": {"type": "string"}},
|
||||
"x-fastmcp-wrap-result": True,
|
||||
}
|
||||
assert tool.output_schema == expected_schema
|
||||
|
||||
result = await tool.run({})
|
||||
# Unstructured content
|
||||
assert len(result.content) == 1
|
||||
assert result.content[0].text == "hello"
|
||||
# Structured content should be wrapped
|
||||
assert result.structured_content == {"result": "hello"}
|
||||
|
||||
async def test_structured_content_with_explicit_object_schema(self):
|
||||
"""Test structured content with explicit object schema."""
|
||||
|
||||
def func() -> dict[str, str]:
|
||||
return {"greeting": "hello"}
|
||||
|
||||
# Provide explicit object schema
|
||||
explicit_schema = {
|
||||
"type": "object",
|
||||
"properties": {"greeting": {"type": "string"}},
|
||||
"required": ["greeting"],
|
||||
}
|
||||
tool = Tool.from_function(func, output_schema=explicit_schema)
|
||||
assert tool.output_schema == explicit_schema
|
||||
|
||||
result = await tool.run({})
|
||||
# Should use direct value since explicit schema doesn't have wrap marker
|
||||
assert result.structured_content == {"greeting": "hello"}
|
||||
|
||||
async def test_structured_content_with_custom_wrapper_schema(self):
|
||||
"""Test structured content with custom schema that includes wrap marker."""
|
||||
|
||||
def func() -> str:
|
||||
return "world"
|
||||
|
||||
# Custom schema with wrap marker
|
||||
custom_schema = {
|
||||
"type": "object",
|
||||
"properties": {"message": {"type": "string"}},
|
||||
"x-fastmcp-wrap-result": True,
|
||||
}
|
||||
tool = Tool.from_function(func, output_schema=custom_schema)
|
||||
assert tool.output_schema == custom_schema
|
||||
|
||||
result = await tool.run({})
|
||||
# Should wrap with "result" key due to wrap marker
|
||||
assert result.structured_content == {"result": "world"}
|
||||
|
||||
async def test_none_vs_false_output_schema_behavior(self):
|
||||
"""Test the difference between None and False for output_schema."""
|
||||
|
||||
def func() -> int:
|
||||
return 123
|
||||
|
||||
# None should disable
|
||||
tool_none = Tool.from_function(func, output_schema=None)
|
||||
assert tool_none.output_schema is None
|
||||
|
||||
# False should also disable
|
||||
tool_false = Tool.from_function(func, output_schema=False)
|
||||
assert tool_false.output_schema is None
|
||||
|
||||
# Both should have same behavior
|
||||
result_none = await tool_none.run({})
|
||||
result_false = await tool_false.run({})
|
||||
|
||||
assert result_none.structured_content is None
|
||||
assert result_false.structured_content is None
|
||||
assert result_none.content[0].text == result_false.content[0].text == "123"
|
||||
|
||||
async def test_non_object_output_schema_raises_error(self):
|
||||
"""Test that providing a non-object output schema raises a ValueError."""
|
||||
|
||||
def func() -> int:
|
||||
return 42
|
||||
|
||||
# Test various non-object schemas that should raise errors
|
||||
non_object_schemas = [
|
||||
{"type": "string"},
|
||||
{"type": "integer", "minimum": 0},
|
||||
{"type": "number"},
|
||||
{"type": "boolean"},
|
||||
{"type": "array", "items": {"type": "string"}},
|
||||
]
|
||||
|
||||
for schema in non_object_schemas:
|
||||
with pytest.raises(
|
||||
ValueError, match='Output schemas must have "type" set to "object"'
|
||||
):
|
||||
Tool.from_function(func, output_schema=schema)
|
||||
|
||||
|
||||
class TestConvertResultToContent:
|
||||
"""Tests for the _convert_to_content helper function."""
|
||||
|
|
|
|||
|
|
@ -4,14 +4,15 @@ from typing import Annotated, Any
|
|||
|
||||
import pytest
|
||||
from dirty_equals import IsList
|
||||
from pydantic import BaseModel, Field
|
||||
from mcp.types import TextContent
|
||||
from pydantic import BaseModel, Field, TypeAdapter
|
||||
from typing_extensions import TypedDict
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.client.client import Client
|
||||
from fastmcp.exceptions import ToolError
|
||||
from fastmcp.tools import Tool, forward, forward_raw
|
||||
from fastmcp.tools.tool import FunctionTool
|
||||
from fastmcp.tools.tool import FunctionTool, ToolResult
|
||||
from fastmcp.tools.tool_transform import ArgTransform, TransformedTool
|
||||
|
||||
|
||||
|
|
@ -691,7 +692,7 @@ async def test_arg_transform_type_precedence_runtime():
|
|||
# Convert string back to int for the original function
|
||||
result = await forward_raw(x=int(x), y=y)
|
||||
# Extract the text from the result
|
||||
result_text = result.content[0].text
|
||||
result_text = result.content[0].text # type: ignore[attr-defined]
|
||||
return f"String input '{x}' converted to result: {result_text}"
|
||||
|
||||
tool = Tool.from_tool(
|
||||
|
|
@ -1030,3 +1031,266 @@ def test_arg_transform_examples_in_schema(add_tool):
|
|||
)
|
||||
prop3 = get_property(new_tool3, "old_x")
|
||||
assert "examples" not in prop3
|
||||
|
||||
|
||||
class TestTransformToolOutputSchema:
|
||||
"""Test output schema handling in transformed tools."""
|
||||
|
||||
@pytest.fixture
|
||||
def base_string_tool(self) -> FunctionTool:
|
||||
"""Tool that returns a string (gets wrapped)."""
|
||||
|
||||
def string_tool(x: int) -> str:
|
||||
return f"Result: {x}"
|
||||
|
||||
return Tool.from_function(string_tool)
|
||||
|
||||
@pytest.fixture
|
||||
def base_dict_tool(self) -> FunctionTool:
|
||||
"""Tool that returns a dict (object type, not wrapped)."""
|
||||
|
||||
def dict_tool(x: int) -> dict[str, int]:
|
||||
return {"value": x}
|
||||
|
||||
return Tool.from_function(dict_tool)
|
||||
|
||||
def test_transform_inherits_parent_output_schema(self, base_string_tool):
|
||||
"""Test that transformed tool inherits parent's output schema by default."""
|
||||
new_tool = Tool.from_tool(base_string_tool)
|
||||
|
||||
# Should inherit parent's wrapped string schema
|
||||
expected_schema = {
|
||||
"type": "object",
|
||||
"properties": {"result": {"type": "string"}},
|
||||
"x-fastmcp-wrap-result": True,
|
||||
}
|
||||
assert new_tool.output_schema == expected_schema
|
||||
assert new_tool.output_schema == base_string_tool.output_schema
|
||||
|
||||
def test_transform_with_explicit_output_schema_false(self, base_string_tool):
|
||||
"""Test that output_schema=False disables structured output."""
|
||||
new_tool = Tool.from_tool(base_string_tool, output_schema=False)
|
||||
|
||||
assert new_tool.output_schema is None
|
||||
|
||||
async def test_transform_output_schema_false_runtime(self, base_string_tool):
|
||||
"""Test runtime behavior with output_schema=False."""
|
||||
new_tool = Tool.from_tool(base_string_tool, output_schema=False)
|
||||
|
||||
# Debug: check that output_schema is actually None
|
||||
assert new_tool.output_schema is None, (
|
||||
f"Expected None, got {new_tool.output_schema}"
|
||||
)
|
||||
|
||||
result = await new_tool.run({"x": 5})
|
||||
assert result.structured_content is None
|
||||
assert result.content[0].text == "Result: 5" # type: ignore[attr-defined]
|
||||
|
||||
def test_transform_with_explicit_output_schema_dict(self, base_string_tool):
|
||||
"""Test that explicit output schema overrides parent."""
|
||||
custom_schema = {
|
||||
"type": "object",
|
||||
"properties": {"message": {"type": "string"}},
|
||||
}
|
||||
new_tool = Tool.from_tool(base_string_tool, output_schema=custom_schema)
|
||||
|
||||
assert new_tool.output_schema == custom_schema
|
||||
assert new_tool.output_schema != base_string_tool.output_schema
|
||||
|
||||
async def test_transform_explicit_schema_runtime(self, base_string_tool):
|
||||
"""Test runtime behavior with explicit output schema."""
|
||||
custom_schema = {"type": "string", "minLength": 1}
|
||||
new_tool = Tool.from_tool(base_string_tool, output_schema=custom_schema)
|
||||
|
||||
result = await new_tool.run({"x": 10})
|
||||
# Non-object explicit schemas disable structured content
|
||||
assert result.structured_content is None
|
||||
assert result.content[0].text == "Result: 10" # type: ignore[attr-defined]
|
||||
|
||||
def test_transform_with_custom_function_inferred_schema(self, base_dict_tool):
|
||||
"""Test that custom function's output schema is inferred."""
|
||||
|
||||
async def custom_fn(x: int) -> str:
|
||||
result = await forward(x=x)
|
||||
return f"Custom: {result.content[0].text}" # type: ignore[attr-defined]
|
||||
|
||||
new_tool = Tool.from_tool(base_dict_tool, transform_fn=custom_fn)
|
||||
|
||||
# Should infer string schema from custom function and wrap it
|
||||
expected_schema = {
|
||||
"type": "object",
|
||||
"properties": {"result": {"type": "string"}},
|
||||
"x-fastmcp-wrap-result": True,
|
||||
}
|
||||
assert new_tool.output_schema == expected_schema
|
||||
|
||||
async def test_transform_custom_function_runtime(self, base_dict_tool):
|
||||
"""Test runtime behavior with custom function that has inferred schema."""
|
||||
|
||||
async def custom_fn(x: int) -> str:
|
||||
result = await forward(x=x)
|
||||
return f"Custom: {result.content[0].text}" # type: ignore[attr-defined]
|
||||
|
||||
new_tool = Tool.from_tool(base_dict_tool, transform_fn=custom_fn)
|
||||
|
||||
result = await new_tool.run({"x": 3})
|
||||
# Should wrap string result
|
||||
assert result.structured_content == {"result": 'Custom: {\n "value": 3\n}'}
|
||||
|
||||
def test_transform_custom_function_fallback_to_parent(self, base_string_tool):
|
||||
"""Test that custom function without output annotation falls back to parent."""
|
||||
|
||||
async def custom_fn(x: int):
|
||||
# No return annotation - should fallback to parent schema
|
||||
result = await forward(x=x)
|
||||
return result
|
||||
|
||||
new_tool = Tool.from_tool(base_string_tool, transform_fn=custom_fn)
|
||||
|
||||
# Should use parent's schema since custom function has no annotation
|
||||
assert new_tool.output_schema == base_string_tool.output_schema
|
||||
|
||||
def test_transform_custom_function_explicit_overrides(self, base_string_tool):
|
||||
"""Test that explicit output_schema overrides both custom function and parent."""
|
||||
|
||||
async def custom_fn(x: int) -> dict[str, str]:
|
||||
return {"custom": "value"}
|
||||
|
||||
explicit_schema = {"type": "array", "items": {"type": "number"}}
|
||||
new_tool = Tool.from_tool(
|
||||
base_string_tool, transform_fn=custom_fn, output_schema=explicit_schema
|
||||
)
|
||||
|
||||
# Explicit schema should win
|
||||
assert new_tool.output_schema == explicit_schema
|
||||
|
||||
async def test_transform_custom_function_object_return(self, base_string_tool):
|
||||
"""Test custom function returning object type."""
|
||||
|
||||
async def custom_fn(x: int) -> dict[str, int]:
|
||||
result = await forward(x=x)
|
||||
return {"original": x, "transformed": x * 2}
|
||||
|
||||
new_tool = Tool.from_tool(base_string_tool, transform_fn=custom_fn)
|
||||
|
||||
# Object types should not be wrapped
|
||||
expected_schema = TypeAdapter(dict[str, int]).json_schema()
|
||||
assert new_tool.output_schema == expected_schema
|
||||
assert "x-fastmcp-wrap-result" not in new_tool.output_schema
|
||||
|
||||
result = await new_tool.run({"x": 4})
|
||||
# Direct value, not wrapped
|
||||
assert result.structured_content == {"original": 4, "transformed": 8}
|
||||
|
||||
async def test_transform_preserves_wrap_marker_behavior(self, base_string_tool):
|
||||
"""Test that wrap marker behavior is preserved through transformation."""
|
||||
new_tool = Tool.from_tool(base_string_tool)
|
||||
|
||||
result = await new_tool.run({"x": 7})
|
||||
# Should wrap because parent schema has wrap marker
|
||||
assert result.structured_content == {"result": "Result: 7"}
|
||||
assert "x-fastmcp-wrap-result" in new_tool.output_schema
|
||||
|
||||
def test_transform_chained_output_schema_inheritance(self, base_string_tool):
|
||||
"""Test output schema inheritance through multiple transformations."""
|
||||
# First transformation keeps parent schema
|
||||
tool1 = Tool.from_tool(base_string_tool)
|
||||
assert tool1.output_schema == base_string_tool.output_schema
|
||||
|
||||
# Second transformation also inherits
|
||||
tool2 = Tool.from_tool(tool1)
|
||||
assert (
|
||||
tool2.output_schema == tool1.output_schema == base_string_tool.output_schema
|
||||
)
|
||||
|
||||
# Third transformation with explicit override
|
||||
custom_schema = {"type": "number"}
|
||||
tool3 = Tool.from_tool(tool2, output_schema=custom_schema)
|
||||
assert tool3.output_schema == custom_schema
|
||||
assert tool3.output_schema != tool2.output_schema
|
||||
|
||||
async def test_transform_mixed_structured_unstructured_content(
|
||||
self, base_string_tool
|
||||
):
|
||||
"""Test transformation handling of mixed content types."""
|
||||
|
||||
async def custom_fn(x: int) -> list:
|
||||
# Return mixed content including ToolResult
|
||||
if x == 1:
|
||||
return ["text", {"data": x}]
|
||||
else:
|
||||
# Return ToolResult directly
|
||||
return ToolResult(
|
||||
content=[TextContent(type="text", text=f"Custom: {x}")],
|
||||
structured_content={"custom_value": x},
|
||||
)
|
||||
|
||||
new_tool = Tool.from_tool(base_string_tool, transform_fn=custom_fn)
|
||||
|
||||
# Test mixed content return
|
||||
result1 = await new_tool.run({"x": 1})
|
||||
assert result1.structured_content == {"result": ["text", {"data": 1}]}
|
||||
|
||||
# Test ToolResult return
|
||||
result2 = await new_tool.run({"x": 2})
|
||||
assert result2.structured_content == {"custom_value": 2}
|
||||
assert result2.content[0].text == "Custom: 2" # type: ignore[attr-defined]
|
||||
|
||||
def test_transform_output_schema_with_arg_transforms(self, base_string_tool):
|
||||
"""Test that output schema works correctly with argument transformations."""
|
||||
|
||||
async def custom_fn(new_x: int) -> dict[str, str]:
|
||||
result = await forward(new_x=new_x)
|
||||
return {"transformed": result.content[0].text} # type: ignore[attr-defined]
|
||||
|
||||
new_tool = Tool.from_tool(
|
||||
base_string_tool,
|
||||
transform_fn=custom_fn,
|
||||
transform_args={"x": ArgTransform(name="new_x")},
|
||||
)
|
||||
|
||||
# Should infer object schema from custom function
|
||||
expected_schema = TypeAdapter(dict[str, str]).json_schema()
|
||||
assert new_tool.output_schema == expected_schema
|
||||
|
||||
async def test_transform_output_schema_none_vs_false(self, base_string_tool):
|
||||
"""Test None vs False behavior for output_schema in transforms."""
|
||||
# None (default) should use smart fallback (inherit from parent)
|
||||
tool_none = Tool.from_tool(base_string_tool) # default output_schema=None
|
||||
assert tool_none.output_schema == base_string_tool.output_schema # Inherits
|
||||
|
||||
# False should explicitly disable
|
||||
tool_false = Tool.from_tool(base_string_tool, output_schema=False)
|
||||
assert tool_false.output_schema is None
|
||||
|
||||
# Different behavior at runtime
|
||||
result_none = await tool_none.run({"x": 5})
|
||||
result_false = await tool_false.run({"x": 5})
|
||||
|
||||
assert result_none.structured_content == {
|
||||
"result": "Result: 5"
|
||||
} # Inherits wrapping
|
||||
assert result_false.structured_content is None # Disabled
|
||||
assert result_none.content[0].text == result_false.content[0].text
|
||||
|
||||
async def test_transform_output_schema_with_tool_result_return(
|
||||
self, base_string_tool
|
||||
):
|
||||
"""Test transform when custom function returns ToolResult directly."""
|
||||
|
||||
async def custom_fn(x: int) -> ToolResult:
|
||||
# Custom function returns ToolResult - should bypass schema handling
|
||||
return ToolResult(
|
||||
content=[TextContent(type="text", text=f"Direct: {x}")],
|
||||
structured_content={"direct_value": x, "doubled": x * 2},
|
||||
)
|
||||
|
||||
new_tool = Tool.from_tool(base_string_tool, transform_fn=custom_fn)
|
||||
|
||||
# ToolResult return type should result in None output schema
|
||||
assert new_tool.output_schema is None
|
||||
|
||||
result = await new_tool.run({"x": 6})
|
||||
# Should use ToolResult content directly
|
||||
assert result.content[0].text == "Direct: 6" # type: ignore[attr-defined]
|
||||
assert result.structured_content == {"direct_value": 6, "doubled": 12}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
from datetime import datetime
|
||||
from typing import Union
|
||||
from typing import Any, Union
|
||||
|
||||
import pytest
|
||||
from pydantic import AnyUrl, BaseModel, TypeAdapter, ValidationError
|
||||
|
|
@ -326,6 +326,29 @@ class TestObjectTypes:
|
|||
}
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"input_type, expected_type",
|
||||
[
|
||||
# Plain dict becomes dict[str, Any] (JSON Schema accurate)
|
||||
(dict, dict[str, Any]),
|
||||
# dict[str, Any] stays the same
|
||||
(dict[str, Any], dict[str, Any]),
|
||||
# Simple typed dicts work correctly
|
||||
(dict[str, str], dict[str, str]),
|
||||
(dict[str, int], dict[str, int]),
|
||||
# Union value types work
|
||||
(dict[str, str | int], dict[str, str | int]),
|
||||
# Key types are constrained to str in JSON Schema
|
||||
(dict[int, list[str]], dict[str, list[str]]),
|
||||
# Union key types become str (JSON Schema limitation)
|
||||
(dict[str | int, str | None], dict[str, str | None]),
|
||||
],
|
||||
)
|
||||
def test_dict_types_are_generated_correctly(self, input_type, expected_type):
|
||||
schema = TypeAdapter(input_type).json_schema()
|
||||
generated_type = json_schema_to_type(schema)
|
||||
assert generated_type == expected_type
|
||||
|
||||
def test_object_accepts_valid(self, simple_object):
|
||||
validator = TypeAdapter(simple_object)
|
||||
result = validator.validate_python({"name": "test", "age": 30})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue