diff --git a/fastmcp_slim/fastmcp/resources/template.py b/fastmcp_slim/fastmcp/resources/template.py index de1858c60..14332ba74 100644 --- a/fastmcp_slim/fastmcp/resources/template.py +++ b/fastmcp_slim/fastmcp/resources/template.py @@ -7,7 +7,7 @@ import inspect import re from collections.abc import Callable from types import UnionType -from typing import Any, ClassVar, Union, get_args, get_origin +from typing import Annotated, Any, ClassVar, Union, get_args, get_origin from urllib.parse import parse_qs, quote, unquote from mcp_types import Annotations, Icon @@ -48,15 +48,21 @@ def extract_query_params(uri_template: str) -> set[str]: return set() -def _is_collection_annotation(annotation: Any) -> bool: - """Whether an annotation accepts a sequence — including `list[str] | None`.""" - if annotation in (list, set, tuple, frozenset): +def _is_list_annotation(annotation: Any) -> bool: + """Whether an annotation accepts a list — including `Annotated[list[str] | None, ...]`. + + Only `list` counts: it is the only collection `expand_uri_template` emits as + repeated keys, so it is the only one that round-trips. + """ + if annotation is list: return True origin = get_origin(annotation) - if origin in (list, set, tuple, frozenset): + if origin is list: return True + if origin is Annotated: + return _is_list_annotation(get_args(annotation)[0]) if origin in (Union, UnionType): - return any(_is_collection_annotation(arg) for arg in get_args(annotation)) + return any(_is_list_annotation(arg) for arg in get_args(annotation)) return False @@ -565,12 +571,27 @@ class FunctionResourceTemplate(ResourceTemplate): # A list-typed query parameter needs the RFC 6570 explode modifier; # without it the parameter is a scalar and every value but the first # is silently dropped, which then fails validation at read time. + # + # Resolve the hints rather than reading the raw signature: under + # `from __future__ import annotations` every annotation is a string, + # and `Annotated[...]` wrappers hide the underlying type. + from fastmcp.tools.function_tool import _resolve_param_hints + + try: + hints = _resolve_param_hints(fn) + except NameError: + # An annotation naming something we can't import here is not + # worth failing registration over; fall back to the raw form. + hints = {} + exploded = { p.replace("-", "_") for p in extract_exploded_query_params(uri_template) } for param_name in sorted(query_params - exploded): - annotation = user_sig.parameters[param_name].annotation - if _is_collection_annotation(annotation): + annotation = hints.get( + param_name, user_sig.parameters[param_name].annotation + ) + if _is_list_annotation(annotation): raise ValueError( f"Query parameter '{param_name}' is a collection type, so it " f"must be declared with the RFC 6570 explode modifier: " diff --git a/tests/resources/test_resource_template.py b/tests/resources/test_resource_template.py index a266c0117..9f8d01d68 100644 --- a/tests/resources/test_resource_template.py +++ b/tests/resources/test_resource_template.py @@ -1,8 +1,9 @@ import functools +from typing import Annotated from urllib.parse import quote import pytest -from pydantic import BaseModel +from pydantic import BaseModel, Field from fastmcp import Client, Context, FastMCP from fastmcp.resources import ResourceTemplate @@ -918,6 +919,76 @@ class TestMalformedURITemplates: fn=search, uri_template="items://{category}{?tags}" ) + @pytest.fixture + def postponed_search(self): + """A function whose annotations are strings, as under PEP 563. + + Defined by exec so the future import applies to it and not to this + whole test module. + """ + namespace: dict[str, object] = {} + exec( + "from __future__ import annotations\n" + "def search(category: str, tags: list[str] | None = None) -> dict:\n" + " return {'category': category, 'tags': tags}\n", + namespace, + ) + return namespace["search"] + + def test_postponed_annotations_still_require_explode(self, postponed_search): + """Regression: a string annotation must not slip past the explode check. + + Under `from __future__ import annotations` the raw signature gives the + string `"list[str] | None"`, so the check has to resolve hints or the + template registers and fails later at read time instead. + """ + with pytest.raises(ValueError, match="explode modifier"): + ResourceTemplate.from_function( + fn=postponed_search, uri_template="items://{category}{?tags}" + ) + + async def test_postponed_annotations_accept_explode(self, postponed_search): + """The explode form registers and collects one or many values.""" + template = ResourceTemplate.from_function( + fn=postponed_search, uri_template="items://{category}{?tags*}" + ) + + one = match_uri_template("items://books?tags=a", "items://{category}{?tags*}") + assert one is not None + assert one == {"category": "books", "tags": ["a"]} + assert await template.read(one) == {"category": "books", "tags": ["a"]} + + many = match_uri_template( + "items://books?tags=a&tags=b", "items://{category}{?tags*}" + ) + assert many is not None + assert many == {"category": "books", "tags": ["a", "b"]} + assert await template.read(many) == {"category": "books", "tags": ["a", "b"]} + + def test_annotated_list_query_param_requires_explode(self): + """`Annotated[...]` must be unwrapped before the collection check.""" + + def search( + category: str, + tags: Annotated[list[str] | None, Field(description="tags")] = None, + ) -> dict: + return {"category": category, "tags": tags} + + with pytest.raises(ValueError, match="explode modifier"): + ResourceTemplate.from_function( + fn=search, uri_template="items://{category}{?tags}" + ) + + def test_non_list_collection_query_params_are_unrestricted(self): + """Only `list` round-trips through expansion, so only `list` is checked.""" + + def search(category: str, tags: set[str] | None = None) -> dict: + return {"category": category, "tags": tags} + + ResourceTemplate.from_function( + fn=search, uri_template="items://{category}{?tags}" + ) + def test_from_function_rejects_hyphen_underscore_collision(self): """Two raw param names that normalize to the same key are rejected."""