From f950b8ca95a27566722d1ef17da2b23811308218 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 20 Jul 2025 21:05:13 -0400 Subject: [PATCH] Support from __future__ import annotations (#1199) --- src/fastmcp/tools/tool.py | 25 ++- src/fastmcp/utilities/types.py | 75 ++++++++- tests/server/test_server_interactions.py | 21 ++- tests/tools/test_tool_future_annotations.py | 170 ++++++++++++++++++++ 4 files changed, 283 insertions(+), 8 deletions(-) create mode 100644 tests/tools/test_tool_future_annotations.py diff --git a/src/fastmcp/tools/tool.py b/src/fastmcp/tools/tool.py index ae07b6216..0cb31523f 100644 --- a/src/fastmcp/tools/tool.py +++ b/src/fastmcp/tools/tool.py @@ -3,7 +3,15 @@ from __future__ import annotations import inspect from collections.abc import Callable from dataclasses import dataclass -from typing import TYPE_CHECKING, Annotated, Any, Generic, Literal, TypeVar +from typing import ( + TYPE_CHECKING, + Annotated, + Any, + Generic, + Literal, + TypeVar, + get_type_hints, +) import mcp.types import pydantic_core @@ -371,7 +379,20 @@ class ParsedFunction: input_schema = compress_schema(input_schema, prune_params=prune_params) output_schema = None - output_type = inspect.signature(fn).return_annotation + # Get the return annotation from the signature + sig = inspect.signature(fn) + output_type = sig.return_annotation + + # If the annotation is a string (from __future__ annotations), resolve it + if isinstance(output_type, str): + try: + # Use get_type_hints to resolve the return type + # include_extras=True preserves Annotated metadata + type_hints = get_type_hints(fn, include_extras=True) + output_type = type_hints.get("return", output_type) + except Exception: + # If resolution fails, keep the string annotation + pass if output_type not in (inspect._empty, None, Any, ...): # there are a variety of types that we don't want to attempt to diff --git a/src/fastmcp/utilities/types.py b/src/fastmcp/utilities/types.py index add9fdc0f..d309f2af9 100644 --- a/src/fastmcp/utilities/types.py +++ b/src/fastmcp/utilities/types.py @@ -8,7 +8,15 @@ from collections.abc import Callable from functools import lru_cache from pathlib import Path from types import EllipsisType, UnionType -from typing import Annotated, TypeAlias, TypeVar, Union, get_args, get_origin +from typing import ( + Annotated, + TypeAlias, + TypeVar, + Union, + get_args, + get_origin, + get_type_hints, +) import mcp.types from mcp.types import Annotations @@ -35,6 +43,54 @@ def get_cached_typeadapter(cls: T) -> TypeAdapter[T]: However, this isn't feasible for user-generated functions. Instead, we use a cache to minimize the cost of creating them as much as possible. """ + # For functions, we need to ensure TypeAdapter can resolve forward + # references + # Normally this could be done by setting e.g. parent_depth=3 to reflect the + # globals in the parent stack, but this utility function can't make that assumption. + if inspect.isfunction(cls) or inspect.ismethod(cls): + # Only try to resolve annotations if the function has them + if hasattr(cls, "__annotations__") and cls.__annotations__: + try: + # Use include_extras=True to preserve Annotated metadata + resolved_hints = get_type_hints(cls, include_extras=True) + # Check if we need to create a new function with resolved annotations + if resolved_hints != cls.__annotations__: + # Create a new function object with resolved annotations + import types + + # Handle both functions and methods + if inspect.ismethod(cls): + actual_func = cls.__func__ + code = actual_func.__code__ + globals_dict = actual_func.__globals__ + name = actual_func.__name__ + defaults = actual_func.__defaults__ + closure = actual_func.__closure__ + else: + code = cls.__code__ + globals_dict = cls.__globals__ + name = cls.__name__ + defaults = cls.__defaults__ + closure = cls.__closure__ + + new_func = types.FunctionType( + code, + globals_dict, + name, + defaults, + closure, + ) + new_func.__dict__.update(cls.__dict__) + new_func.__module__ = cls.__module__ + new_func.__qualname__ = getattr(cls, "__qualname__", cls.__name__) + new_func.__annotations__ = resolved_hints + return TypeAdapter(new_func) + except Exception: + # If resolution fails, this might be due to closure-scoped types + # that aren't available in the function's globals. In this case, + # we'll let TypeAdapter handle the string annotations directly. + pass + return TypeAdapter(cls) @@ -77,12 +133,21 @@ def find_kwarg_by_type(fn: Callable, kwarg_type: type) -> str | None: Includes union types that contain the kwarg_type, as well as Annotated types. """ if inspect.ismethod(fn) and hasattr(fn, "__func__"): - sig = inspect.signature(fn.__func__) - else: - sig = inspect.signature(fn) + fn = fn.__func__ + # Try to get resolved type hints + try: + # Use include_extras=True to preserve Annotated metadata + type_hints = get_type_hints(fn, include_extras=True) + except Exception: + # If resolution fails, use raw annotations if they exist + type_hints = getattr(fn, "__annotations__", {}) + + sig = inspect.signature(fn) for name, param in sig.parameters.items(): - if is_class_member_of_type(param.annotation, kwarg_type): + # Use resolved hint if available, otherwise raw annotation + annotation = type_hints.get(name, param.annotation) + if is_class_member_of_type(annotation, kwarg_type): return name return None diff --git a/tests/server/test_server_interactions.py b/tests/server/test_server_interactions.py index 986fb275a..a04fa2515 100644 --- a/tests/server/test_server_interactions.py +++ b/tests/server/test_server_interactions.py @@ -31,6 +31,20 @@ from fastmcp.utilities.json_schema import compress_schema from fastmcp.utilities.types import Audio, File, Image +def _normalize_anyof_order(schema): + """Normalize the order of items in anyOf arrays for consistent comparison.""" + if isinstance(schema, dict): + if "anyOf" in schema: + # Sort anyOf items by their string representation for consistent ordering + schema = schema.copy() + schema["anyOf"] = sorted(schema["anyOf"], key=str) + # Recursively normalize nested objects + return {k: _normalize_anyof_order(v) for k, v in schema.items()} + elif isinstance(schema, list): + return [_normalize_anyof_order(item) for item in schema] + return schema + + class PersonTypedDict(TypedDict): name: str age: int @@ -917,7 +931,12 @@ class TestToolOutputSchema: type_schema = compress_schema(TypeAdapter(annotation).json_schema()) assert len(tools) == 1 - assert tools[0].outputSchema == type_schema + + # Normalize anyOf ordering for comparison since union type order + # can vary between environments when using annotation resolution + actual_schema = _normalize_anyof_order(tools[0].outputSchema) + expected_schema = _normalize_anyof_order(type_schema) + assert actual_schema == expected_schema async def test_disabled_output_schema_no_structured_content(self): mcp = FastMCP() diff --git a/tests/tools/test_tool_future_annotations.py b/tests/tools/test_tool_future_annotations.py new file mode 100644 index 000000000..a5d9d8c0d --- /dev/null +++ b/tests/tools/test_tool_future_annotations.py @@ -0,0 +1,170 @@ +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()