From cb34191128abbf7bc25cdba373a6c37ca4c1e3f1 Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Tue, 3 Mar 2026 14:49:36 -0500
Subject: [PATCH] Fix resource templates with query params on mounted servers

Closes #3366
---
 .../server/providers/fastmcp_provider.py      | 20 +++++-
 tests/server/mount/test_resources.py          | 62 +++++++++++++++++++
 2 files changed, 81 insertions(+), 1 deletion(-)

diff --git a/src/fastmcp/server/providers/fastmcp_provider.py b/src/fastmcp/server/providers/fastmcp_provider.py
index cab6bc541..92038c540 100644
--- a/src/fastmcp/server/providers/fastmcp_provider.py
+++ b/src/fastmcp/server/providers/fastmcp_provider.py
@@ -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
 
 
diff --git a/tests/server/mount/test_resources.py b/tests/server/mount/test_resources.py
index d0e18b7af..e6c4fb502 100644
--- a/tests/server/mount/test_resources.py
+++ b/tests/server/mount/test_resources.py
@@ -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"