Support RFC 6570 explode modifier for list-typed query params

🤖 Generated with Claude Code
This commit is contained in:
Sai Mouli 2026-07-29 16:39:26 +05:30
commit 49fe614331
3 changed files with 138 additions and 9 deletions

View file

@ -673,6 +673,22 @@ def read_file(path: str, encoding: str = "utf-8", lines: int = 100) -> str:
FastMCP automatically coerces query parameter string values to the correct types based on your function's type hints (`int`, `float`, `bool`, `str`).
**Repeatable query parameters:**
A plain `{?param}` is a single value — if a client repeats the key, only the first value is used. To accept a list, add the RFC 6570 explode modifier `*`:
```python
@mcp.resource("items://{category}{?tags*}")
def search(category: str, tags: list[str] | None = None) -> dict:
return {"category": category, "tags": tags or []}
```
- `items://books` → `tags=[]`
- `items://books?tags=alpha` → `tags=["alpha"]`
- `items://books?tags=alpha&tags=beta` → `tags=["alpha", "beta"]`
A single value still arrives as a one-element list, so the parameter's type is stable. Declaring a list-typed parameter without the `*` raises an error when the template is created.
**Query parameters vs. hidden defaults:**
Query parameters expose optional configuration to clients. To hide optional parameters from clients entirely (always use defaults), simply omit them from the URI template:
@ -698,6 +714,7 @@ FastMCP enforces these validation rules when creating resource templates:
1. **Required function parameters** (no default values) must appear in the URI path template
2. **Query parameters** (specified with `{?param}` syntax) must be optional function parameters with default values
3. **All URI template parameters** (path and query) must exist as function parameters
4. **List-typed query parameters** must use the explode modifier (`{?param*}`)
Optional function parameters (those with default values) can be:
- Included as query parameters (`{?param}`) - clients can override via query string

View file

@ -6,7 +6,8 @@ import functools
import inspect
import re
from collections.abc import Callable
from typing import Any, ClassVar
from types import UnionType
from typing import Any, ClassVar, Union, get_args, get_origin
from urllib.parse import parse_qs, quote, unquote
from mcp_types import Annotations, Icon
@ -36,10 +37,40 @@ from fastmcp.utilities.types import get_cached_typeadapter
def extract_query_params(uri_template: str) -> set[str]:
"""Extract query parameter names from RFC 6570 `{?param1,param2}` syntax."""
"""Extract query parameter names from RFC 6570 `{?param1,param2}` syntax.
The explode modifier is stripped, so `{?tags*}` yields `{"tags"}`. Use
`extract_exploded_query_params` to find which names carried it.
"""
match = re.search(r"\{\?([^}]+)\}", uri_template)
if match:
return {p.strip() for p in match.group(1).split(",")}
return {p.strip().removesuffix("*") for p in match.group(1).split(",")}
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):
return True
origin = get_origin(annotation)
if origin in (list, set, tuple, frozenset):
return True
if origin in (Union, UnionType):
return any(_is_collection_annotation(arg) for arg in get_args(annotation))
return False
def extract_exploded_query_params(uri_template: str) -> set[str]:
"""Extract query parameter names declared with the RFC 6570 explode modifier.
`{?tags*}` marks `tags` as repeatable `?tags=a&tags=b` collects into a
list rather than collapsing to the first value.
"""
match = re.search(r"\{\?([^}]+)\}", uri_template)
if match:
return {
p.strip()[:-1] for p in match.group(1).split(",") if p.strip().endswith("*")
}
return set()
@ -80,7 +111,7 @@ def build_regex(template: str) -> re.Pattern[str] | None:
return None
def match_uri_template(uri: str, uri_template: str) -> dict[str, str] | None:
def match_uri_template(uri: str, uri_template: str) -> dict[str, Any] | None:
"""Match URI against template and extract both path and query parameters.
Supports RFC 6570 URI templates:
@ -98,7 +129,7 @@ def match_uri_template(uri: str, uri_template: str) -> dict[str, str] | None:
if not match:
return None
params = {k: unquote(v) for k, v in match.groupdict().items()}
params: dict[str, Any] = {k: unquote(v) for k, v in match.groupdict().items()}
# Extract query parameters if present in URI and template
if query_string:
@ -107,14 +138,21 @@ def match_uri_template(uri: str, uri_template: str) -> dict[str, str] | None:
# so callers can distinguish "explicitly empty" from "missing".
parsed_query = parse_qs(query_string, keep_blank_values=True)
exploded = extract_exploded_query_params(uri_template)
for name in query_param_names:
if name in parsed_query:
# Take first value if multiple provided.
# Normalize hyphens to underscores to match Python param names.
# Don't overwrite path params that were already extracted.
key = name.replace("-", "_")
if key not in params:
params[key] = parsed_query[name][0]
# An exploded `{?name*}` param keeps every repetition;
# a plain `{?name}` param is a scalar, so take the first.
params[key] = (
parsed_query[name]
if name in exploded
else parsed_query[name][0]
)
return params
@ -153,11 +191,18 @@ def expand_uri_template(uri_template: str, params: dict[str, Any]) -> str:
names = [n.strip() for n in match.group(1).split(",")]
parts = []
for name in names:
# `{?tags*}` is the explode form; the value is a list, emitted as a
# repeated key so it round-trips back through match_uri_template.
name = name.removesuffix("*")
underscored = name.replace("-", "_")
if name in params:
parts.append(f"{quote(name)}={quote(str(params[name]))}")
value = params[name]
elif underscored in params:
parts.append(f"{quote(name)}={quote(str(params[underscored]))}")
value = params[underscored]
else:
continue
values = value if isinstance(value, (list, tuple)) else [value]
parts.extend(f"{quote(name)}={quote(str(v))}" for v in values)
if parts:
return "?" + "&".join(parts)
return ""
@ -517,6 +562,22 @@ class FunctionResourceTemplate(ResourceTemplate):
f"Query parameters {invalid_query_params} must be optional function parameters with default values"
)
# 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.
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):
raise ValueError(
f"Query parameter '{param_name}' is a collection type, so it "
f"must be declared with the RFC 6570 explode modifier: "
f"use '{{?{param_name}*}}' instead of '{{?{param_name}}}' "
f"so repeated values (?{param_name}=a&{param_name}=b) are collected."
)
# Check if required parameters are a subset of the path parameters
if not required_params.issubset(path_params):
raise ValueError(

View file

@ -869,6 +869,55 @@ class TestMalformedURITemplates:
assert result is not None
assert result == {"id": "42", "format": "", "verbose": "true"}
def test_exploded_query_param_collects_repeated_values(self):
"""Regression for #4378: `{?tags*}` collects every repetition."""
result = match_uri_template(
"items://books?tags=alpha&tags=beta", "items://{category}{?tags*}"
)
assert result == {"category": "books", "tags": ["alpha", "beta"]}
def test_exploded_query_param_with_single_value_is_a_list(self):
"""A single value still arrives as a list, so the type is stable."""
result = match_uri_template(
"items://books?tags=alpha", "items://{category}{?tags*}"
)
assert result == {"category": "books", "tags": ["alpha"]}
def test_non_exploded_query_param_stays_scalar(self):
"""Without the explode modifier the param is a scalar — first value wins."""
result = match_uri_template(
"items://books?tag=alpha&tag=beta", "items://{category}{?tag}"
)
assert result == {"category": "books", "tag": "alpha"}
async def test_exploded_query_param_reaches_the_function(self):
"""End-to-end: repeated values arrive as a list[str] argument."""
def search(category: str, tags: list[str] | None = None) -> dict:
return {"category": category, "tags": tags}
template = ResourceTemplate.from_function(
fn=search, uri_template="items://{category}{?tags*}"
)
assert await template.read({"category": "books", "tags": ["a", "b"]}) == {
"category": "books",
"tags": ["a", "b"],
}
def test_from_function_rejects_collection_query_param_without_explode(self):
"""Regression for #4378: `{?tags}` on a list param silently dropped every
value but the first, then failed validation at read time. Reject it up
front and point at the explode form."""
def search(category: str, tags: list[str] | None = 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_from_function_rejects_hyphen_underscore_collision(self):
"""Two raw param names that normalize to the same key are rejected."""
@ -989,6 +1038,8 @@ class TestMatchExpandRoundTrip:
("test://{path*}", "test://single"),
("test://pre/{rest*}", "test://pre/x/y/z"),
("test://{x}/{path*}", "test://foo/a/b/c"),
("test://{x}{?tags*}", "test://foo?tags=a"),
("test://{x}{?tags*}", "test://foo?tags=a&tags=b"),
],
)
def test_expand_then_match_is_identity(self, template: str, uri: str):