modelcontextprotocol · codecracker007 · Apr 4, 2025 · Apr 4, 2025 · Apr 5, 2025 · Apr 5, 2025
@@ -288,6 +288,98 @@ def get_settings() -> str:
   "language": "en",
   "debug": false
 }"""
+
+
+# Form-style query expansion examples using RFC 6570 URI templates
+
+
+@mcp.resource("articles://{article_id}/view{?format,lang}")
+def view_article(article_id: str, format: str = "html", lang: str = "en") -> str:
+    """View an article with optional format and language selection.
+
+    Example URIs:
+    - articles://123/view (uses defaults: format=html, lang=en)
+    - articles://123/view?format=pdf (format=pdf, lang=en)
+    - articles://123/view?format=pdf&lang=fr (format=pdf, lang=fr)
+    """
+    if format == "pdf":
+        content = f"PDF content for article {article_id} in {lang}"
+    elif format == "json":
+        content = f'{{"article_id": "{article_id}", "content": "...", "lang": "{lang}"}}'
+    else:
+        content = f"<html><body>Article {article_id} in {lang}</body></html>"
+
+    return content
+
+
+@mcp.resource("search://query/{search_term}{?page,limit,category,sort}")
+def search_content(
+    search_term: str, page: int = 1, limit: int = 10, category: str = "all", sort: str = "relevance"
+) -> str:
+    """Search content with optional pagination and filtering.
+
+    Example URIs:
+    - search://query/python (basic search)
+    - search://query/python?page=2&limit=20 (pagination)
+    - search://query/python?category=tutorial&sort=date (filtering)
+    """
+    offset = (page - 1) * limit
+    results = f"Search results for '{search_term}' (category: {category}, sort: {sort})"
+    results += f"\nShowing {limit} results starting from {offset + 1}"
+
+    # Simulated search results
+    for i in range(limit):
+        result_num = offset + i + 1
+        results += f"\n{result_num}. Result about {search_term} in {category}"
+
+    return results
+
+
+@mcp.resource("users://{user_id}/profile{?include_private,format}")
+def get_user_profile(user_id: str, include_private: bool = False, format: str = "summary") -> str:
+    """Get user profile with optional private data and format selection.
+
+    Example URIs:
+    - users://123/profile (public data, summary format)
+    - users://123/profile?include_private=true (includes private data)
+    - users://123/profile?format=detailed&include_private=true (detailed with private)
+    """
+    from typing import Any
+
+    profile_data: dict[str, Any] = {"user_id": user_id, "name": "John Doe", "public_bio": "Software developer"}
+
+    if include_private:
+        profile_data.update({"email": "john@example.com", "phone": "+1234567890"})
+
+    if format == "detailed":
+        profile_data.update({"last_active": "2024-01-20", "preferences": {"notifications": True}})
+
+    return str(profile_data)
+
+
+@mcp.resource("api://weather/{location}{?units,lang,include_forecast,days}")
+def get_weather_data(
+    location: str, units: str = "metric", lang: str = "en", include_forecast: bool = False, days: int = 5
+) -> str:
+    """Get weather data with customizable options.
+
+    Example URIs:
+    - api://weather/london (basic weather)
+    - api://weather/london?units=imperial&lang=es (different units and language)
+    - api://weather/london?include_forecast=true&days=7 (with 7-day forecast)
+    """
+    temp_unit = "C" if units == "metric" else "F"
+    base_temp = 22 if units == "metric" else 72
+
+    weather_info = f"Weather for {location}: {base_temp}{temp_unit}"
+
+    if include_forecast:
+        weather_info += f"\n{days}-day forecast:"
+        for day in range(1, days + 1):
+            forecast_temp = base_temp + (day % 3)
+            weather_info += f"\nDay {day}: {forecast_temp}{temp_unit}"
+
+    return weather_info
 ```
 
 _Full example: [examples/snippets/servers/basic_resource.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/basic_resource.py)_

@@ -18,3 +18,95 @@ def get_settings() -> str:
   "language": "en",
   "debug": false
 }"""
+
+
+# Form-style query expansion examples using RFC 6570 URI templates
+
+
+@mcp.resource("articles://{article_id}/view{?format,lang}")
+def view_article(article_id: str, format: str = "html", lang: str = "en") -> str:
+    """View an article with optional format and language selection.
+
+    Example URIs:
+    - articles://123/view (uses defaults: format=html, lang=en)
+    - articles://123/view?format=pdf (format=pdf, lang=en)
+    - articles://123/view?format=pdf&lang=fr (format=pdf, lang=fr)
+    """
+    if format == "pdf":
+        content = f"PDF content for article {article_id} in {lang}"
+    elif format == "json":
+        content = f'{{"article_id": "{article_id}", "content": "...", "lang": "{lang}"}}'
+    else:
+        content = f"<html><body>Article {article_id} in {lang}</body></html>"
+
+    return content
+
+
+@mcp.resource("search://query/{search_term}{?page,limit,category,sort}")
+def search_content(
+    search_term: str, page: int = 1, limit: int = 10, category: str = "all", sort: str = "relevance"
+) -> str:
+    """Search content with optional pagination and filtering.
+
+    Example URIs:
+    - search://query/python (basic search)
+    - search://query/python?page=2&limit=20 (pagination)
+    - search://query/python?category=tutorial&sort=date (filtering)
+    """
+    offset = (page - 1) * limit
+    results = f"Search results for '{search_term}' (category: {category}, sort: {sort})"
+    results += f"\nShowing {limit} results starting from {offset + 1}"
+
+    # Simulated search results
+    for i in range(limit):
+        result_num = offset + i + 1
+        results += f"\n{result_num}. Result about {search_term} in {category}"
+
+    return results
+
+
+@mcp.resource("users://{user_id}/profile{?include_private,format}")
+def get_user_profile(user_id: str, include_private: bool = False, format: str = "summary") -> str:
+    """Get user profile with optional private data and format selection.
+
+    Example URIs:
+    - users://123/profile (public data, summary format)
+    - users://123/profile?include_private=true (includes private data)
+    - users://123/profile?format=detailed&include_private=true (detailed with private)
+    """
+    from typing import Any
+
+    profile_data: dict[str, Any] = {"user_id": user_id, "name": "John Doe", "public_bio": "Software developer"}
+
+    if include_private:
+        profile_data.update({"email": "john@example.com", "phone": "+1234567890"})
+
+    if format == "detailed":
+        profile_data.update({"last_active": "2024-01-20", "preferences": {"notifications": True}})
+
+    return str(profile_data)
+
+
+@mcp.resource("api://weather/{location}{?units,lang,include_forecast,days}")
+def get_weather_data(
+    location: str, units: str = "metric", lang: str = "en", include_forecast: bool = False, days: int = 5
+) -> str:
+    """Get weather data with customizable options.
+
+    Example URIs:
+    - api://weather/london (basic weather)
+    - api://weather/london?units=imperial&lang=es (different units and language)
+    - api://weather/london?include_forecast=true&days=7 (with 7-day forecast)
+    """
+    temp_unit = "C" if units == "metric" else "F"
+    base_temp = 22 if units == "metric" else 72
+
+    weather_info = f"Weather for {location}: {base_temp}{temp_unit}"
+
+    if include_forecast:
+        weather_info += f"\n{days}-day forecast:"
+        for day in range(1, days + 1):
+            forecast_temp = base_temp + (day % 3)
+            weather_info += f"\nDay {day}: {forecast_temp}{temp_unit}"
+
+    return weather_info
@@ -4,24 +4,36 @@
 
 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)")
+    uri_template: str = Field(description="URI template with parameters (e.g. weather://{city}/current{?units,format})")
     name: str = Field(description="Name of the resource")
     title: str | None = Field(description="Human-readable title of the resource", default=None)
     description: str | None = Field(description="Description of what the resource does")
     mime_type: str = Field(default="text/plain", description="MIME type of the resource content")
     fn: Callable[..., Any] = Field(exclude=True)
     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(
@@ -34,40 +46,127 @@ 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()
 
-        # ensure the arguments are properly cast
-        fn = validate_call(fn)
+        # First, apply pydantic's validation and coercion
+        validated_fn = validate_call(original_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} 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 function parameters {optional_params}"
+            )
 
         return cls(
             uri_template=uri_template,
             name=func_name,
             title=title,
-            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
 
@@ -80,4 +179,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}")