Spaces:
Running
Running
File size: 6,379 Bytes
3ab4db9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | from __future__ import annotations
from typing import Any, cast
import mcp.types
import pytest
from fastmcp import Context, FastMCP
from fastmcp.client import Client
from fastmcp.tools.tool import ToolResult
from fastmcp.utilities.types import Image
fastmcp_server = FastMCP()
@fastmcp_server.tool
def simple_with_context(ctx: Context) -> str:
"""Simple tool with context parameter."""
return f"Request ID: {ctx.request_id}"
@fastmcp_server.tool
def complex_types(
data: dict[str, Any], items: list[int], ctx: Context
) -> dict[str, str | int]:
"""Tool with complex type annotations."""
return {"count": len(items), "request_id": ctx.request_id}
@fastmcp_server.tool
def optional_context(name: str, ctx: Context | None = None) -> str:
"""Tool with optional context."""
if ctx:
return f"Hello {name} from request {ctx.request_id}"
return f"Hello {name}"
@fastmcp_server.tool
def union_with_context(value: int | str, ctx: Context) -> ToolResult:
"""Tool returning ToolResult with context."""
return ToolResult(content=f"Value: {value}, Request: {ctx.request_id}")
@fastmcp_server.tool
def returns_image(ctx: Context) -> Image:
"""Tool that returns an Image."""
# Create a simple 1x1 white pixel PNG
png_data = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDATx\x9cc\xf8\x0f\x00\x00\x01\x01\x00\x05\x18\xd4c\x00\x00\x00\x00IEND\xaeB`\x82"
return Image(data=png_data, format="png")
@fastmcp_server.tool
async def async_with_context(ctx: Context) -> str:
"""Async tool with context."""
return f"Async request: {ctx.request_id}"
class TestFutureAnnotations:
async def test_simple_with_context(self):
async with Client(fastmcp_server) as client:
result = await client.call_tool("simple_with_context", {})
assert "Request ID:" in cast(mcp.types.TextContent, result.content[0]).text
async def test_complex_types(self):
async with Client(fastmcp_server) as client:
result = await client.call_tool(
"complex_types", {"data": {"key": "value"}, "items": [1, 2, 3]}
)
# Check the result is valid JSON with expected values
import json
data = json.loads(cast(mcp.types.TextContent, result.content[0]).text)
assert data["count"] == 3
assert "request_id" in data
async def test_optional_context(self):
async with Client(fastmcp_server) as client:
result = await client.call_tool("optional_context", {"name": "World"})
assert (
"Hello World from request"
in cast(mcp.types.TextContent, result.content[0]).text
)
async def test_union_with_context(self):
async with Client(fastmcp_server) as client:
result = await client.call_tool("union_with_context", {"value": 42})
assert (
"Value: 42, Request:"
in cast(mcp.types.TextContent, result.content[0]).text
)
async def test_returns_image(self):
async with Client(fastmcp_server) as client:
result = await client.call_tool("returns_image", {})
assert result.content[0].type == "image"
assert result.content[0].mimeType == "image/png"
async def test_async_with_context(self):
async with Client(fastmcp_server) as client:
result = await client.call_tool("async_with_context", {})
assert (
"Async request:" in cast(mcp.types.TextContent, result.content[0]).text
)
async def test_modern_union_syntax_works(self):
"""Test that modern | union syntax works with future annotations."""
# This demonstrates that our solution works with | syntax when types
# are available in module globals
# Define a tool with modern union syntax
@fastmcp_server.tool
def modern_union_tool(value: str | int | None) -> str | None:
"""Tool using modern | union syntax throughout."""
if value is None:
return None
return f"processed: {value}"
async with Client(fastmcp_server) as client:
# Test with string
result = await client.call_tool("modern_union_tool", {"value": "hello"})
assert (
"processed: hello"
in cast(mcp.types.TextContent, result.content[0]).text
)
# Test with int
result = await client.call_tool("modern_union_tool", {"value": 42})
assert (
"processed: 42" in cast(mcp.types.TextContent, result.content[0]).text
)
# Test with None
result = await client.call_tool("modern_union_tool", {"value": None})
# When function returns None, FastMCP returns empty content
assert (
len(result.content) == 0
or cast(mcp.types.TextContent, result.content[0]).text == "null"
)
@pytest.mark.xfail(
reason="Closure-scoped types cannot be resolved with 'from __future__ import annotations'. "
"When using future annotations, all type annotations become strings that need to be evaluated "
"using eval() in the function's global namespace. Types defined only in closure scope "
"(like local imports or type aliases) are not available in the function's __globals__ "
"and therefore cannot be resolved by get_type_hints()."
)
def test_closure_scoped_types_limitation():
"""
This test demonstrates that closure-scoped types don't work with future annotations.
The fundamental issue is that 'from __future__ import annotations' converts all
annotations to strings, and those strings can only be resolved using the function's
global namespace, not local variables from closures.
"""
def create_failing_closure():
# This import is only available in the closure scope
mcp = FastMCP()
@mcp.tool
def closure_tool(value: str | None) -> str:
"""This will fail because Optional can't be resolved from closure import."""
return str(value)
return mcp
# This should raise an error during tool registration
create_failing_closure()
|