diff --git a/src/fastmcp/resources/template.py b/src/fastmcp/resources/template.py index 265e88f91..53ea6a15e 100644 --- a/src/fastmcp/resources/template.py +++ b/src/fastmcp/resources/template.py @@ -7,7 +7,7 @@ import inspect import re from collections.abc import Callable from typing import TYPE_CHECKING, Any, ClassVar, overload -from urllib.parse import parse_qs, unquote +from urllib.parse import parse_qs, quote, unquote import mcp.types from mcp.types import Annotations, Icon @@ -109,6 +109,38 @@ def match_uri_template(uri: str, uri_template: str) -> dict[str, str] | None: return params +def expand_uri_template(uri_template: str, params: dict[str, Any]) -> str: + """Expand a URI template with parameters — inverse of `match_uri_template`. + + Supports the same RFC 6570 subset: + - Path params: `{var}`, `{var*}` + - Query params: `{?var1,var2}` + """ + result = uri_template + + # Replace {name} and {name*} path placeholders + for key, value in params.items(): + value_str = str(value) + result = result.replace(f"{{{key}}}", value_str) + result = result.replace(f"{{{key}*}}", value_str) + + # 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 = [ + f"{quote(name)}={quote(str(params[name]))}" + for name in names + if name in params + ] + if parts: + return "?" + "&".join(parts) + return "" + + result = re.sub(r"\{\?([^}]+)\}", _expand_query_block, result) + + return result + + class ResourceTemplate(FastMCPComponent): """A template for dynamically creating resources.""" diff --git a/src/fastmcp/server/providers/fastmcp_provider.py b/src/fastmcp/server/providers/fastmcp_provider.py index 6c6b2ad11..8ae2c18d3 100644 --- a/src/fastmcp/server/providers/fastmcp_provider.py +++ b/src/fastmcp/server/providers/fastmcp_provider.py @@ -10,18 +10,16 @@ executed. from __future__ import annotations -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 from fastmcp.prompts.base import Prompt, PromptResult from fastmcp.resources.base import Resource, ResourceResult -from fastmcp.resources.template import ResourceTemplate +from fastmcp.resources.template import ResourceTemplate, expand_uri_template from fastmcp.server.providers.base import Provider from fastmcp.server.tasks.config import TaskMeta from fastmcp.server.telemetry import delegate_span @@ -36,34 +34,6 @@ if TYPE_CHECKING: from fastmcp.server.server import FastMCP -def _expand_uri_template(template: str, params: dict[str, Any]) -> str: - """Expand a URI template with parameters. - - 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 - - # ----------------------------------------------------------------------------- # FastMCPProvider component classes # ----------------------------------------------------------------------------- @@ -403,7 +373,7 @@ class FastMCPProviderResourceTemplate(ResourceTemplate): URI that the nested server understands. """ # Expand the original template with params to get internal URI - original_uri = _expand_uri_template(self._original_uri_template or "", params) + original_uri = expand_uri_template(self._original_uri_template or "", params) return FastMCPProviderResource( server=self._server, original_uri=original_uri, @@ -433,7 +403,7 @@ class FastMCPProviderResourceTemplate(ResourceTemplate): server before calling this method. """ # Expand the original template with params to get internal URI - original_uri = _expand_uri_template(self._original_uri_template or "", params) + original_uri = expand_uri_template(self._original_uri_template or "", params) # Pass exact version so child reads the correct version version = VersionSpec(eq=self.version) if self.version else None @@ -455,9 +425,7 @@ class FastMCPProviderResourceTemplate(ResourceTemplate): This method is called by Docket during background task execution. """ # Expand the original template with arguments to get internal URI - original_uri = _expand_uri_template( - self._original_uri_template or "", arguments - ) + original_uri = expand_uri_template(self._original_uri_template or "", arguments) # Pass exact version so child reads the correct version version = VersionSpec(eq=self.version) if self.version else None diff --git a/tests/resources/test_resource_template.py b/tests/resources/test_resource_template.py index 3b88545c5..d15c8da63 100644 --- a/tests/resources/test_resource_template.py +++ b/tests/resources/test_resource_template.py @@ -7,7 +7,11 @@ from pydantic import BaseModel from fastmcp import Context, FastMCP from fastmcp.resources import ResourceTemplate from fastmcp.resources.function_resource import FunctionResource -from fastmcp.resources.template import build_regex, match_uri_template +from fastmcp.resources.template import ( + build_regex, + expand_uri_template, + match_uri_template, +) class TestResourceTemplate: @@ -806,3 +810,86 @@ class TestMalformedURITemplates: assert match is not None assert match.group("name") == "foo" assert match.group("id") == "123" + + +class TestExpandUriTemplate: + """Test expand_uri_template — the inverse of match_uri_template.""" + + @pytest.mark.parametrize( + "template, params, expected", + [ + ("test://{x}", {"x": "foo"}, "test://foo"), + ("test://{x}/{y}", {"x": "foo", "y": "bar"}, "test://foo/bar"), + ("test://a/{x}/b", {"x": "mid"}, "test://a/mid/b"), + ], + ) + def test_expand_simple_params( + self, template: str, params: dict[str, str], expected: str + ): + assert expand_uri_template(template, params) == expected + + @pytest.mark.parametrize( + "template, params, expected", + [ + ("test://{path*}", {"path": "a/b/c"}, "test://a/b/c"), + ("test://{path*}", {"path": "single"}, "test://single"), + ("test://pre/{rest*}", {"rest": "x/y"}, "test://pre/x/y"), + ( + "test://{a*}/mid/{b*}", + {"a": "x/y", "b": "p/q"}, + "test://x/y/mid/p/q", + ), + ("test://{x}/{path*}", {"x": "foo", "path": "a/b"}, "test://foo/a/b"), + ], + ) + def test_expand_wildcard_params( + self, template: str, params: dict[str, str], expected: str + ): + assert expand_uri_template(template, params) == expected + + def test_expand_query_params(self): + result = expand_uri_template( + "test://data{?format,verbose}", + {"format": "json", "verbose": "true"}, + ) + assert result in ( + "test://data?format=json&verbose=true", + "test://data?verbose=true&format=json", + ) + + def test_expand_query_params_partial(self): + result = expand_uri_template( + "test://data{?format,verbose}", + {"format": "json"}, + ) + assert result == "test://data?format=json" + + def test_expand_query_params_none(self): + result = expand_uri_template("test://data{?format,verbose}", {}) + assert result == "test://data" + + def test_expand_ignores_extra_params(self): + result = expand_uri_template("test://{x}", {"x": "foo", "unused": "bar"}) + assert result == "test://foo" + + +class TestMatchExpandRoundTrip: + """match_uri_template and expand_uri_template must agree on the template grammar.""" + + @pytest.mark.parametrize( + "template, uri", + [ + ("test://{x}", "test://foo"), + ("test://{x}/{y}", "test://foo/bar"), + ("test://a/{x}/b", "test://a/mid/b"), + ("test://{path*}", "test://a/b/c"), + ("test://{path*}", "test://single"), + ("test://pre/{rest*}", "test://pre/x/y/z"), + ("test://{x}/{path*}", "test://foo/a/b/c"), + ], + ) + def test_expand_then_match_is_identity(self, template: str, uri: str): + """Extracting params from a URI and expanding them back reproduces the URI.""" + params = match_uri_template(uri, template) + assert params is not None + assert expand_uri_template(template, params) == uri diff --git a/tests/server/mount/test_resources.py b/tests/server/mount/test_resources.py index e6c4fb502..06e2ce548 100644 --- a/tests/server/mount/test_resources.py +++ b/tests/server/mount/test_resources.py @@ -54,6 +54,23 @@ class TestResourcesAndTemplates: assert profile["id"] == "123" assert profile["name"] == "User 123" + async def test_mount_with_wildcard_resource_template(self): + """Wildcard `{name*}` params must survive round-trip through a namespaced mount.""" + main_app = FastMCP("MainApp") + sub_app = FastMCP("SubApp") + + @sub_app.resource("resource://multi/{extra*}") + def multi(extra: str) -> str: + return extra + + main_app.mount(sub_app, namespace="sub") + + result = await main_app.read_resource("resource://sub/multi/abc/def") + assert result.contents[0].content == "abc/def" + + result = await main_app.read_resource("resource://sub/multi/abc") + assert result.contents[0].content == "abc" + async def test_adding_resource_after_mounting(self): """Test adding a resource after mounting.""" main_app = FastMCP("MainApp")