Resolve annotations before requiring explode, and scope it to lists

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BAKNJthovhc5VSi4MTiZUe

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: "

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."""