Fix resource templates with query params on mounted servers

Closes #3366
This commit is contained in:
Jeremiah Lowin 2026-03-03 14:49:36 -05:00
commit cb34191128
2 changed files with 81 additions and 1 deletions

View file

@ -14,6 +14,7 @@ import re
from collections.abc import AsyncIterator, Sequence
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING, Any, overload
from urllib.parse import quote
import mcp.types
from mcp.types import AnyUrl
@ -38,11 +39,28 @@ if TYPE_CHECKING:
def _expand_uri_template(template: str, params: dict[str, Any]) -> str:
"""Expand a URI template with parameters.
Simple implementation that handles {name} style placeholders.
Handles both {name} path placeholders and RFC 6570 {?param1,param2}
query parameter syntax.
"""
result = template
# Replace {name} path placeholders
for key, value in params.items():
result = re.sub(rf"\{{{key}\}}", str(value), result)
# Expand {?param1,param2,...} query parameter blocks
def _expand_query_block(match: re.Match[str]) -> str:
names = [n.strip() for n in match.group(1).split(",")]
parts = []
for name in names:
if name in params:
parts.append(f"{quote(name)}={quote(str(params[name]))}")
if parts:
return "?" + "&".join(parts)
return ""
result = re.sub(r"\{\?([^}]+)\}", _expand_query_block, result)
return result

View file

@ -134,3 +134,65 @@ class TestResourceUriPrefixing:
t for t in templates if t.uri_template == "resource://prefix/user/{user_id}"
)
assert template.name == "user_template"
class TestMountedResourceTemplateQueryParams:
"""Test that resource templates with query params work on mounted servers."""
async def test_mounted_template_with_query_param(self):
"""Query params in resource templates should work through mount."""
sub = FastMCP("Sub")
@sub.resource("resource://greet{?name}")
def greet(name: str = "World") -> str:
return f"Hello, {name}!"
main = FastMCP("Main")
main.mount(sub, "sub")
result = await main.read_resource("resource://sub/greet?name=Alice")
assert result.contents[0].content == "Hello, Alice!"
async def test_mounted_template_with_query_param_default(self):
"""Missing query params should use defaults through mount."""
sub = FastMCP("Sub")
@sub.resource("resource://greet{?name}")
def greet(name: str = "World") -> str:
return f"Hello, {name}!"
main = FastMCP("Main")
main.mount(sub, "sub")
result = await main.read_resource("resource://sub/greet")
assert result.contents[0].content == "Hello, World!"
async def test_mounted_template_with_multiple_query_params(self):
"""Multiple query params should all pass through mount correctly."""
sub = FastMCP("Sub")
@sub.resource("resource://data/{id}{?format,verbose}")
def get_data(id: str, format: str = "json", verbose: bool = False) -> str:
return f"id={id} format={format} verbose={verbose}"
main = FastMCP("Main")
main.mount(sub, "api")
result = await main.read_resource(
"resource://api/data/42?format=xml&verbose=true"
)
assert result.contents[0].content == "id=42 format=xml verbose=True"
async def test_mounted_template_with_partial_query_params(self):
"""Providing only some query params should use defaults for the rest."""
sub = FastMCP("Sub")
@sub.resource("resource://data/{id}{?format,limit}")
def get_data(id: str, format: str = "json", limit: int = 10) -> str:
return f"id={id} format={format} limit={limit}"
main = FastMCP("Main")
main.mount(sub, "api")
result = await main.read_resource("resource://api/data/42?limit=5")
assert result.contents[0].content == "id=42 format=json limit=5"