Added RFC 6570 complaint form style query expansion as optional parameters to resources

Author: Agasthya Kasturi

pyproject.toml

@@ -30,6 +30,7 @@ dependencies = [
     "sse-starlette>=1.6.1",
     "pydantic-settings>=2.5.2",
     "uvicorn>=0.23.1",
+    "pyright>=1.1.391",
 ]
 
 [project.optional-dependencies]

src/mcp/server/fastmcp/resources/templates.py

@@ -4,6 +4,7 @@
 
 import inspect
 import re
+import urllib.parse
 from collections.abc import Callable
 from typing import Any
 
@@ -16,7 +17,7 @@ class ResourceTemplate(BaseModel):
     """A template for dynamically creating resources."""
 
     uri_template: str = Field(
-        description="URI template with parameters (e.g. weather://{city}/current)"
+        description="URI template with parameters (e.g. weather://{city}/current{?units,format})"
     )
     name: str = Field(description="Name of the resource")
     description: str | None = Field(description="Description of what the resource does")
@@ -27,6 +28,14 @@ class ResourceTemplate(BaseModel):
     parameters: dict[str, Any] = Field(
         description="JSON schema for function parameters"
     )
+    required_params: set[str] = Field(
+        default_factory=set,
+        description="Set of required parameters from the path component",
+    )
+    optional_params: set[str] = Field(
+        default_factory=set,
+        description="Set of optional parameters specified in the query component",
+    )
 
     @classmethod
     def from_function(
@@ -48,29 +57,113 @@ def from_function(
         # ensure the arguments are properly cast
         fn = validate_call(fn)
 
+        # Extract required and optional parameters from function signature
+        required_params, optional_params = cls._analyze_function_params(fn)
+
+        # Extract path parameters from URI template
+        path_params: set[str] = set(
+            re.findall(r"{(\w+)}", re.sub(r"{(\?.+?)}", "", uri_template))
+        )
+
+        # Extract query parameters from the URI template if present
+        query_param_match = re.search(r"{(\?(?:\w+,)*\w+)}", uri_template)
+        query_params: set[str] = set()
+        if query_param_match:
+            # Extract query parameters from {?param1,param2,...} syntax
+            query_str = query_param_match.group(1)
+            query_params = set(
+                query_str[1:].split(",")
+            )  # Remove the leading '?' and split
+
+        # Validate path parameters match required function parameters
+        print(f"path_params: {path_params}")
+        print(f"required_params: {required_params}")
+        if path_params != required_params:
+            raise ValueError(
+                f"Mismatch between URI path parameters {path_params} "
+                f"and required function parameters {required_params}"
+            )
+
+        # Validate query parameters are a subset of optional function parameters
+        if not query_params.issubset(optional_params):
+            invalid_params: set[str] = query_params - optional_params
+            raise ValueError(
+                f"Query parameters {invalid_params} do not match optional "
+                f"function parameters {optional_params}"
+            )
+
         return cls(
             uri_template=uri_template,
             name=func_name,
             description=description or fn.__doc__ or "",
             mime_type=mime_type or "text/plain",
             fn=fn,
             parameters=parameters,
+            required_params=required_params,
+            optional_params=optional_params,
         )
 
+    @staticmethod
+    def _analyze_function_params(fn: Callable[..., Any]) -> tuple[set[str], set[str]]:
+        """Analyze function signature to extract required and optional parameters."""
+        required_params: set[str] = set()
+        optional_params: set[str] = set()
+
+        signature = inspect.signature(fn)
+        for name, param in signature.parameters.items():
+            # Parameters with default values are optional
+            if param.default is param.empty:
+                required_params.add(name)
+            else:
+                optional_params.add(name)
+
+        return required_params, optional_params
+
     def matches(self, uri: str) -> dict[str, Any] | None:
         """Check if URI matches template and extract parameters."""
-        # Convert template to regex pattern
-        pattern = self.uri_template.replace("{", "(?P<").replace("}", ">[^/]+)")
-        match = re.match(f"^{pattern}$", uri)
-        if match:
-            return match.groupdict()
-        return None
+        # Split URI into path and query parts
+        if "?" in uri:
+            path, query = uri.split("?", 1)
+        else:
+            path, query = uri, ""
+
+        # Remove the query parameter part from the template for matching
+        path_template = re.sub(r"{(\?.+?)}", "", self.uri_template)
+
+        # Convert template to regex pattern for path part
+        pattern = path_template.replace("{", "(?P<").replace("}", ">[^/]+)")
+        match = re.match(f"^{pattern}$", path)
+
+        if not match:
+            return None
+
+        # Extract path parameters
+        params = match.groupdict()
+
+        # Parse and add query parameters if present
+        if query:
+            query_params = urllib.parse.parse_qs(query)
+            for key, value in query_params.items():
+                if key in self.optional_params:
+                    # Use the first value if multiple are provided
+                    params[key] = value[0] if value else None
+
+        return params
 
     async def create_resource(self, uri: str, params: dict[str, Any]) -> Resource:
         """Create a resource from the template with the given parameters."""
         try:
+            # Prepare parameters for function call
+            # For optional parameters not in URL, use their default values
+            fn_params = {}
+
+            # First add extracted parameters
+            for name, value in params.items():
+                if name in self.required_params or name in self.optional_params:
+                    fn_params[name] = value
+
             # Call function and check if result is a coroutine
-            result = self.fn(**params)
+            result = self.fn(**fn_params)
             if inspect.iscoroutine(result):
                 result = await result
 

src/mcp/server/fastmcp/server.py

@@ -4,7 +4,6 @@
 
 import inspect
 import json
-import re
 from collections.abc import AsyncIterator, Callable, Iterable, Sequence
 from contextlib import (
     AbstractAsyncContextManager,
@@ -327,6 +326,15 @@ def resource(
         If the URI contains parameters (e.g. "resource://{param}") or the function
         has parameters, it will be registered as a template resource.
 
+        Function parameters in the path are required,
+        while parameters with default values
+        can be optionally provided as query parameters using RFC 6570 form-style query
+        expansion syntax: {?param1,param2,...}
+
+        Examples:
+        - resource://{category}/{id}{?filter,sort,limit}
+        - resource://{user_id}/profile{?format,fields}
+
         Args:
             uri: URI for the resource (e.g. "resource://my-resource" or "resource://{param}")
             name: Optional name for the resource
@@ -347,6 +355,19 @@ def get_data() -> str:
             def get_weather(city: str) -> str:
                 return f"Weather for {city}"
 
+            @server.resource("resource://{city}/weather{?units}")
+            def get_weather_with_options(city: str, units: str = "metric") -> str:
+                # Can be called with resource://paris/weather?units=imperial
+                return f"Weather for {city} in {units} units"
+
+            @server.resource("resource://{category}/{id}
+            {?filter,sort,limit}")
+            def get_item(category: str, id: str, filter: str = "all", sort: str = "name"
+            , limit: int = 10) -> str:
+                # Can be called with resource://electronics/1234?filter=new&sort=price&limit=20
+                return f"Item {id} in {category}, filtered by {filter}, sorted by {sort}
+                , limited to {limit}"
+
             @server.resource("resource://{city}/weather")
             async def get_weather(city: str) -> str:
                 data = await fetch_weather(city)
@@ -365,16 +386,6 @@ def decorator(fn: AnyFunction) -> AnyFunction:
             has_func_params = bool(inspect.signature(fn).parameters)
 
             if has_uri_params or has_func_params:
-                # Validate that URI params match function params
-                uri_params = set(re.findall(r"{(\w+)}", uri))
-                func_params = set(inspect.signature(fn).parameters.keys())
-
-                if uri_params != func_params:
-                    raise ValueError(
-                        f"Mismatch between URI parameters {uri_params} "
-                        f"and function parameters {func_params}"
-                    )
-
                 # Register as template
                 self._resource_manager.add_template(
                     fn=fn,

tests/issues/test_141_resource_templates.py

@@ -21,34 +21,85 @@ async def test_resource_template_edge_cases():
     def get_user_post(user_id: str, post_id: str) -> str:
         return f"Post {post_id} by user {user_id}"
 
-    # Test case 2: Template with optional parameter (should fail)
-    with pytest.raises(ValueError, match="Mismatch between URI parameters"):
-
-        @mcp.resource("resource://users/{user_id}/profile")
-        def get_user_profile(user_id: str, optional_param: str | None = None) -> str:
-            return f"Profile for user {user_id}"
+    # Test case 2: Template with valid optional parameters
+    # using form-style query expansion
+    @mcp.resource("resource://users/{user_id}/profile{?format,fields}")
+    def get_user_profile(
+        user_id: str, format: str = "json", fields: str = "basic"
+    ) -> str:
+        return f"Profile for user {user_id} in {format} format with fields: {fields}"
 
     # Test case 3: Template with mismatched parameters
-    with pytest.raises(ValueError, match="Mismatch between URI parameters"):
+    with pytest.raises(
+        ValueError,
+        match="Mismatch between URI path parameters .* and "
+        "required function parameters .*",
+    ):
 
         @mcp.resource("resource://users/{user_id}/profile")
         def get_user_profile_mismatch(different_param: str) -> str:
             return f"Profile for user {different_param}"
 
-    # Test case 4: Template with extra function parameters
-    with pytest.raises(ValueError, match="Mismatch between URI parameters"):
+    # Test case 4: Template with extra required function parameters
+    with pytest.raises(
+        ValueError,
+        match="Mismatch between URI path parameters .* and "
+        "required function parameters .*",
+    ):
 
         @mcp.resource("resource://users/{user_id}/profile")
         def get_user_profile_extra(user_id: str, extra_param: str) -> str:
             return f"Profile for user {user_id}"
 
     # Test case 5: Template with missing function parameters
-    with pytest.raises(ValueError, match="Mismatch between URI parameters"):
+    with pytest.raises(
+        ValueError,
+        match="Mismatch between URI path parameters .* and "
+        "required function parameters .*",
+    ):
 
         @mcp.resource("resource://users/{user_id}/profile/{section}")
         def get_user_profile_missing(user_id: str) -> str:
             return f"Profile for user {user_id}"
 
+    # Test case 6: Invalid query parameter in template (not optional in function)
+    with pytest.raises(
+        ValueError,
+        match="Mismatch between URI path parameters .* and "
+        "required function parameters .*",
+    ):
+
+        @mcp.resource("resource://users/{user_id}/profile{?required_param}")
+        def get_user_profile_invalid_query(user_id: str, required_param: str) -> str:
+            return f"Profile for user {user_id}"
+
+    # Test case 7: Make sure the resource with form-style query parameters works
+    async with client_session(mcp._mcp_server) as client:
+        result = await client.read_resource(AnyUrl("resource://users/123/profile"))
+        assert isinstance(result.contents[0], TextResourceContents)
+        assert (
+            result.contents[0].text
+            == "Profile for user 123 in json format with fields: basic"
+        )
+
+        result = await client.read_resource(
+            AnyUrl("resource://users/123/profile?format=xml")
+        )
+        assert isinstance(result.contents[0], TextResourceContents)
+        assert (
+            result.contents[0].text
+            == "Profile for user 123 in xml format with fields: basic"
+        )
+
+        result = await client.read_resource(
+            AnyUrl("resource://users/123/profile?format=xml&fields=detailed")
+        )
+        assert isinstance(result.contents[0], TextResourceContents)
+        assert (
+            result.contents[0].text
+            == "Profile for user 123 in xml format with fields: detailed"
+        )
+
     # Verify valid template works
     result = await mcp.read_resource("resource://users/123/posts/456")
     result_list = list(result)