modelcontextprotocol · codecracker007 · Apr 4, 2025 · Apr 4, 2025 · Apr 5, 2025 · Apr 5, 2025
diff --git a/README.md b/README.md
@@ -219,6 +219,15 @@ def get_config() -> str:
 def get_user_profile(user_id: str) -> str:
     """Dynamic user data"""
     return f"Profile data for user {user_id}"
+
+
+# Example with form-style query expansion (RFC 6570) using multiple parameters
+@mcp.resource("articles://{article_id}/view{?format,lang}")
+def view_article(article_id: str, format: str = "html", lang: str = "english") -> str:
+    """View an article, with optional format and language selection.
+    Example URI: articles://123/view?format=pdf&lang=english"""
+    content = f"Content for article {article_id} in {format} format Viewing in {lang}."
+    return content
 ```
 
 ### Tools
@@ -555,6 +564,23 @@ def echo_resource(message: str) -> str:
     return f"Resource echo: {message}"
 
 
+# Example with form-style query expansion for customizing echo output
+@mcp.resource("echo://custom/{message}{?case,reverse}")
+def custom_echo_resource(
+    message: str, case: str = "lower", reverse: bool = False
+) -> str:
+    """Echo a message with optional case transformation and reversal.
+    Example URI: echo://custom/Hello?case=upper&reverse=true"""
+    processed_message = message
+    if case == "upper":
+        processed_message = processed_message.upper()
+    elif case == "lower":
+        processed_message = processed_message.lower()
+    if reverse:
+        processed_message = processed_message[::-1]
+    return f"Custom resource echo: {processed_message}"
+
+
 @mcp.tool()
 def echo_tool(message: str) -> str:
     """Echo a message as a tool"""
@@ -587,6 +613,34 @@ def get_schema() -> str:
     return "\n".join(sql[0] for sql in schema if sql[0])
 
 
+# Example with form-style query expansion for table-specific schema
+@mcp.resource("schema://{table_name}{?include_indexes}")
+def get_table_schema(table_name: str, include_indexes: bool = False) -> str:
+    """Provide the schema for a specific table, optionally including indexes.
+    Example URI: schema://users?include_indexes=true"""
+    conn = sqlite3.connect("database.db")
+    cursor = conn.cursor()
+    try:
+        base_query = "SELECT sql FROM sqlite_master WHERE type='table' AND name=?"
+        params: list[str] = [table_name]
+        if include_indexes:
+            cursor.execute(base_query, params)
+            schema_parts = cursor.fetchall()
+
+            index_query = (
+                "SELECT sql FROM sqlite_master WHERE type='index' AND tbl_name=?"
+            )
+            cursor.execute(index_query, params)
+            schema_parts.extend(cursor.fetchall())
+        else:
+            cursor.execute(base_query, params)
+            schema_parts = cursor.fetchall()
+
+        return "\n".join(sql[0] for sql in schema_parts if sql and sql[0])
+    finally:
+        conn.close()
+
+
 @mcp.tool()
 def query_data(sql: str) -> str:
     """Execute SQL queries safely"""
@@ -885,4 +939,4 @@ We are passionate about supporting contributors of all levels of experience and
 
 ## License
 
-This project is licensed under the MIT License - see the LICENSE file for details.
+This project is licensed under the MIT License - see the LICENSE file for details.
diff --git a/pyproject.toml b/pyproject.toml
@@ -123,4 +123,4 @@ filterwarnings = [
     "ignore::DeprecationWarning:websockets",
     "ignore:websockets.server.WebSocketServerProtocol is deprecated:DeprecationWarning",
     "ignore:Returning str or bytes.*:DeprecationWarning:mcp.server.lowlevel"
-]
+]
diff --git a/src/mcp/server/fastmcp/resources/templates.py b/src/mcp/server/fastmcp/resources/templates.py
@@ -4,19 +4,23 @@
 
 import inspect
 import re
+import urllib.parse
 from collections.abc import Callable
 from typing import Any
 
 from pydantic import BaseModel, Field, TypeAdapter, validate_call
 
 from mcp.server.fastmcp.resources.types import FunctionResource, Resource
+from mcp.server.fastmcp.utilities.func_metadata import (
+    use_defaults_on_optional_validation_error,
+)
 
 
 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 +31,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(
@@ -38,39 +50,132 @@ def from_function(
         mime_type: str | None = None,
     ) -> ResourceTemplate:
         """Create a template from a function."""
-        func_name = name or fn.__name__
+        original_fn = fn
+        func_name = name or original_fn.__name__
         if func_name == "<lambda>":
             raise ValueError("You must provide a name for lambda functions")
 
-        # Get schema from TypeAdapter - will fail if function isn't properly typed
-        parameters = TypeAdapter(fn).json_schema()
+        # Get schema from TypeAdapter using the original function for correct schema
+        parameters = TypeAdapter(original_fn).json_schema()
+
+        # First, apply pydantic's validation and coercion
+        validated_fn = validate_call(original_fn)
 
-        # ensure the arguments are properly cast
-        fn = validate_call(fn)
+        # Then, apply our decorator to handle default fallback for optional params
+        final_fn = use_defaults_on_optional_validation_error(validated_fn)
+
+        # Extract required and optional params from the original function's signature
+        required_params, optional_params = cls._analyze_function_params(original_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
+        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 "",
+            description=description or original_fn.__doc__ or "",
             mime_type=mime_type or "text/plain",
-            fn=fn,
+            fn=final_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.
+        This should operate on the original, unwrapped function.
+        """
+        # Ensure we are looking at the original function if it was wrapped elsewhere
+        original_fn_for_analysis = inspect.unwrap(fn)
+        required_params: set[str] = set()
+        optional_params: set[str] = set()
+
+        signature = inspect.signature(original_fn_for_analysis)
+        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:
-            # Call function and check if result is a coroutine
-            result = self.fn(**params)
+            # Prepare parameters for function call
+            # For optional parameters not in URL, use their default values
+
+            # First add extracted parameters
+            fn_params = {
+                name: value
+                for name, value in params.items()
+                if name in self.required_params or name in self.optional_params
+            }
+
+            # self.fn is now multiply-decorated:
+            # 1. validate_call for coercion/validation
+            # 2. our new decorator for default fallback on optional param validation err
+            result = self.fn(**fn_params)
             if inspect.iscoroutine(result):
                 result = await result
 
@@ -82,4 +187,6 @@ async def create_resource(self, uri: str, params: dict[str, Any]) -> Resource:
                 fn=lambda: result,  # Capture result in closure
             )
         except Exception as e:
+            # This will catch errors from validate_call (e.g., for required params)
+            # or from our decorator if retry also fails, or any other errors.
             raise ValueError(f"Error creating resource from template: {e}")
diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py
@@ -3,7 +3,6 @@
 from __future__ import annotations as _annotations
 
 import inspect
-import re
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence
 from contextlib import (
     AbstractAsyncContextManager,
@@ -417,6 +416,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
@@ -437,6 +445,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)
@@ -455,16 +476,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,