diff --git a/README.md b/README.md index d2fb9194a..8eaca9943 100644 --- a/README.md +++ b/README.md @@ -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"Article {article_id} in {lang}" + + 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)_ diff --git a/examples/snippets/servers/basic_resource.py b/examples/snippets/servers/basic_resource.py index 5c1973059..360d0de2f 100644 --- a/examples/snippets/servers/basic_resource.py +++ b/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"Article {article_id} in {lang}" + + 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 diff --git a/src/mcp/server/fastmcp/resources/templates.py b/src/mcp/server/fastmcp/resources/templates.py index b1c7b2711..5393ee12c 100644 --- a/src/mcp/server/fastmcp/resources/templates.py +++ b/src/mcp/server/fastmcp/resources/templates.py @@ -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 == "": 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}") diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index 8724deca7..e20e32d9c 100644 --- 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, Collection, Iterable, Sequence from contextlib import AbstractAsyncContextManager, asynccontextmanager from typing import Any, Generic, Literal @@ -475,6 +474,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 @@ -496,6 +504,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) @@ -514,15 +535,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} and function parameters {func_params}" - ) - # Register as template self._resource_manager.add_template( fn=fn, diff --git a/src/mcp/server/fastmcp/utilities/func_metadata.py b/src/mcp/server/fastmcp/utilities/func_metadata.py index 6d07cf4dc..8b6976a9b 100644 --- a/src/mcp/server/fastmcp/utilities/func_metadata.py +++ b/src/mcp/server/fastmcp/utilities/func_metadata.py @@ -1,3 +1,4 @@ +import functools import inspect import json from collections.abc import Awaitable, Callable, Sequence @@ -11,6 +12,7 @@ ConfigDict, Field, RootModel, + ValidationError, WithJsonSchema, create_model, ) @@ -521,3 +523,110 @@ def _convert_to_content( result = pydantic_core.to_json(result, fallback=str, indent=2).decode() return [TextContent(type="text", text=result)] + + +def use_defaults_on_optional_validation_error( + decorated_fn: Callable[..., Any], +) -> Callable[..., Any]: + """ + Decorator for a function already wrapped by pydantic.validate_call. + If the wrapped function call fails due to a ValidationError, this decorator + checks if the error was caused by an optional parameter. If so, it retries + the call, explicitly omitting the failing optional parameter(s) to allow + Pydantic/the function to use their default values. + + If the error is for a required parameter, or if the retry fails, the original + error is re-raised. + """ + # Get the original function's signature (before validate_call) to inspect defaults + original_fn = inspect.unwrap(decorated_fn) + original_sig = inspect.signature(original_fn) + optional_params_with_defaults = { + name: param.default + for name, param in original_sig.parameters.items() + if param.default is not inspect.Parameter.empty + } + + @functools.wraps(decorated_fn) + async def async_wrapper(*args: Any, **kwargs: Any) -> Any: + try: + return await decorated_fn(*args, **kwargs) + except ValidationError as e: + # Check if the validation error is solely for optional parameters + failing_optional_params_to_retry: dict[str, bool] = {} + failing_required_params: list[str] = [] # Explicitly typed + + for error in e.errors(): + # error['loc'] is a tuple, e.g., ('param_name',) + # Pydantic error locations are tuples of strings or ints. + # For field errors, the first element is the field name (str). + if error["loc"] and isinstance(error["loc"][0], str): + param_name: str = error["loc"][0] + if param_name in optional_params_with_defaults: + # It's an optional param that failed. Mark for retry by exclude. + failing_optional_params_to_retry[param_name] = True + else: + # It's a required parameter or a non-parameter error + failing_required_params.append(param_name) + else: # Non-parameter specific error or unexpected error structure + raise e + + if failing_required_params or not failing_optional_params_to_retry: + # re-raise if any req params failed, or if no opt params were identified + logger.debug( + f"Validation failed for required params or no optional params " + f"identified. Re-raising original error for {original_fn.__name__}." + ) + raise e + + # At this point, only optional parameters caused the ValidationError. + # Retry the call, removing the failing optional params from kwargs. + # This allows validate_call/the function to use their defaults. + new_kwargs = {k: v for k, v in kwargs.items() if k not in failing_optional_params_to_retry} + + # Preserve positional arguments + # failing_optional_params_to_retry.keys() is a KeysView[str] + # list(KeysView[str]) is list[str] + logger.info( + f"Retrying {original_fn.__name__} with default values" + f"for optional params: {list(failing_optional_params_to_retry.keys())}" + ) + return await decorated_fn(*args, **new_kwargs) + + @functools.wraps(decorated_fn) + def sync_wrapper(*args: Any, **kwargs: Any) -> Any: + try: + return decorated_fn(*args, **kwargs) + except ValidationError as e: + failing_optional_params_to_retry: dict[str, bool] = {} + failing_required_params: list[str] = [] # Explicitly typed + + for error in e.errors(): + if error["loc"] and isinstance(error["loc"][0], str): + param_name: str = error["loc"][0] + if param_name in optional_params_with_defaults: + failing_optional_params_to_retry[param_name] = True + else: + failing_required_params.append(param_name) + else: + raise e + + if failing_required_params or not failing_optional_params_to_retry: + logger.debug( + f"Validation failed for required params or no optional params " + f"identified. Re-raising original error for {original_fn.__name__}." + ) + raise e + + new_kwargs = {k: v for k, v in kwargs.items() if k not in failing_optional_params_to_retry} + logger.info( + f"Retrying {original_fn.__name__} with default values" + f"for optional params: {list(failing_optional_params_to_retry.keys())}" + ) + return decorated_fn(*args, **new_kwargs) + + if inspect.iscoroutinefunction( + original_fn + ): # Check original_fn because decorated_fn might be a partial or already wrapped + return async_wrapper + return sync_wrapper diff --git a/tests/issues/test_141_resource_templates.py b/tests/issues/test_141_resource_templates.py index 3145f65e8..1af2277fd 100644 --- a/tests/issues/test_141_resource_templates.py +++ b/tests/issues/test_141_resource_templates.py @@ -1,3 +1,5 @@ +import json + import pytest from pydantic import AnyUrl @@ -21,34 +23,66 @@ 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) @@ -112,3 +146,113 @@ def get_user_profile(user_id: str) -> str: with pytest.raises(Exception): # Specific exception type may vary await session.read_resource(AnyUrl("resource://users/123/invalid")) # Invalid template + + +@pytest.mark.anyio +async def test_resource_template_optional_param_default_fallback_e2e(): + """Test end-to-end that optional params fallback to defaults on validation error.""" + mcp = FastMCP("FallbackDemo") + + @mcp.resource("resource://config/{section}{?theme,timeout,is_feature_enabled}") + def get_config( + section: str, + theme: str = "dark", + timeout: int = 30, + is_feature_enabled: bool = False, + ) -> dict[str, str | int | bool]: + return { + "section": section, + "theme": theme, + "timeout": timeout, + "is_feature_enabled": is_feature_enabled, + } + + async with client_session(mcp._mcp_server) as client: + await client.initialize() + + # 1. All defaults for optional params + uri1 = "resource://config/network" + res1 = await client.read_resource(AnyUrl(uri1)) + assert res1.contents and isinstance(res1.contents[0], TextResourceContents) + data1 = json.loads(res1.contents[0].text) + assert data1 == { + "section": "network", + "theme": "dark", + "timeout": 30, + "is_feature_enabled": False, + } + + # 2. Valid optional params (theme is URL encoded, timeout is valid int string) + uri2 = "resource://config/ui?theme=light%20blue&timeout=60&is_feature_enabled=true" + res2 = await client.read_resource(AnyUrl(uri2)) + assert res2.contents and isinstance(res2.contents[0], TextResourceContents) + data2 = json.loads(res2.contents[0].text) + assert data2 == { + "section": "ui", + "theme": "light blue", + "timeout": 60, + "is_feature_enabled": True, + } + + # 3.Invalid 'timeout'(optional int),valid 'theme','is_feature_enabled' not given + # timeout=abc should use default 30 + uri3 = "resource://config/storage?theme=grayscale&timeout=abc" + res3 = await client.read_resource(AnyUrl(uri3)) + assert res3.contents and isinstance(res3.contents[0], TextResourceContents) + data3 = json.loads(res3.contents[0].text) + assert data3 == { + "section": "storage", + "theme": "grayscale", + "timeout": 30, # Fallback to default + "is_feature_enabled": False, # Fallback to default + } + + # 4.Invalid 'is_feature_enabled'(optional bool),'timeout'valid,'theme' not given + # is_feature_enabled=notbool should use default False + uri4 = "resource://config/user?timeout=15&is_feature_enabled=notbool" + res4 = await client.read_resource(AnyUrl(uri4)) + assert res4.contents and isinstance(res4.contents[0], TextResourceContents) + data4 = json.loads(res4.contents[0].text) + assert data4 == { + "section": "user", + "theme": "dark", # Fallback to default + "timeout": 15, + "is_feature_enabled": False, # Fallback to default + } + + # 5. Empty value for optional 'theme' (string type) + uri5 = "resource://config/general?theme=" + res5 = await client.read_resource(AnyUrl(uri5)) + assert res5.contents and isinstance(res5.contents[0], TextResourceContents) + data5 = json.loads(res5.contents[0].text) + assert data5 == { + "section": "general", + "theme": "dark", # Fallback to default because param is removed by parse_qs + "timeout": 30, + "is_feature_enabled": False, + } + + # 6. Empty value for optional 'timeout' (int type) + # timeout= (empty value) should fall back to default + uri6 = "resource://config/advanced?timeout=" + res6 = await client.read_resource(AnyUrl(uri6)) + assert res6.contents and isinstance(res6.contents[0], TextResourceContents) + data6 = json.loads(res6.contents[0].text) + assert data6 == { + "section": "advanced", + "theme": "dark", + "timeout": 30, # Fallback to default because param is removed by parse_qs + "is_feature_enabled": False, + } + + # 7. Invalid required path param type + # This scenario is more about the FastMCP.read_resource and its error handling + @mcp.resource("resource://item/{item_code}/check") # item_code is string here + def check_item(item_code: int) -> dict[str, str | bool]: # but int in function + return {"item_code_type": str(type(item_code)), "valid_code": item_code > 0} + + uri7 = "resource://item/notaninteger/check" + with pytest.raises(Exception, match="Error creating resource from template"): + # The err is caught by FastMCP.read_resource and re-raised as ResourceError, + # which the client sees as a general McpError or similar. + await client.read_resource(AnyUrl(uri7)) diff --git a/tests/server/fastmcp/resources/test_resource_template.py b/tests/server/fastmcp/resources/test_resource_template.py index f9b91a0a1..69af98709 100644 --- a/tests/server/fastmcp/resources/test_resource_template.py +++ b/tests/server/fastmcp/resources/test_resource_template.py @@ -46,6 +46,67 @@ def my_func(key: str, value: int) -> dict[str, Any]: assert template.matches("test://foo") is None assert template.matches("other://foo/123") is None + def test_template_with_optional_parameters(self): + """Test templates with optional parameters via query string.""" + + def my_func(key: str, sort: str = "asc", limit: int = 10) -> dict[str, str | int]: + return {"key": key, "sort": sort, "limit": limit} + + template = ResourceTemplate.from_function( + fn=my_func, + uri_template="test://{key}", + name="test", + ) + + # Verify required/optional params + assert template.required_params == {"key"} + assert template.optional_params == {"sort", "limit"} + + # Match with no query params - should only extract path param + params = template.matches("test://foo") + assert params == {"key": "foo"} + + # Match with query params + params = template.matches("test://foo?sort=desc&limit=20") + assert params == {"key": "foo", "sort": "desc", "limit": "20"} + + # Match with partial query params + params = template.matches("test://foo?sort=desc") + assert params == {"key": "foo", "sort": "desc"} + + # Match with unknown query params - should ignore + params = template.matches("test://foo?unknown=value") + assert params == {"key": "foo"} + + def test_template_validation(self): + """Test template validation with required/optional parameters.""" + + # Valid: required param in path + def valid_func(key: str, optional: str = "default") -> str: + return f"{key}-{optional}" + + template = ResourceTemplate.from_function( + fn=valid_func, + uri_template="test://{key}", + name="test", + ) + assert template.required_params == {"key"} + assert template.optional_params == {"optional"} + + # Invalid: missing required param in path + def invalid_func(key: str, value: str) -> str: + return f"{key}-{value}" + + with pytest.raises( + ValueError, + match="Mismatch between URI path parameters .* and required function parameters .*", + ): + ResourceTemplate.from_function( + fn=invalid_func, + uri_template="test://{key}", + name="test", + ) + @pytest.mark.anyio async def test_create_resource(self): """Test creating a resource from a template.""" @@ -186,3 +247,286 @@ def get_data(value: str) -> CustomData: assert isinstance(resource, FunctionResource) content = await resource.read() assert content == '"hello"' + + @pytest.mark.anyio + async def test_create_resource_with_optional_params(self): + """Test creating resources with optional parameters.""" + + def my_func(key: str, sort: str = "asc", limit: int = 10) -> dict[str, str | int]: + return {"key": key, "sort": sort, "limit": limit} + + template = ResourceTemplate.from_function( + fn=my_func, + uri_template="test://{key}", + name="test", + ) + + # Create with only required params + params = {"key": "foo"} + resource = await template.create_resource("test://foo", params) + result = await resource.read() + assert isinstance(result, str) + assert json.loads(result) == {"key": "foo", "sort": "asc", "limit": 10} + + # Create with all params + params = {"key": "foo", "sort": "desc", "limit": "20"} + resource = await template.create_resource("test://foo?sort=desc&limit=20", params) + result = await resource.read() + assert isinstance(result, str) + assert json.loads(result) == {"key": "foo", "sort": "desc", "limit": 20} + + def test_template_with_form_style_query_expansion(self): + """Test templates with RFC 6570 form-style query expansion.""" + + def my_func( + category: str, + id: str, + filter: str = "all", + sort: str = "name", + limit: int = 10, + ) -> dict[str, str | int]: + return { + "category": category, + "id": id, + "filter": filter, + "sort": sort, + "limit": limit, + } + + template = ResourceTemplate.from_function( + fn=my_func, + uri_template="test://{category}/{id}{?filter,sort,limit}", + name="test", + ) + + # Verify required/optional params + assert template.required_params == {"category", "id"} + assert template.optional_params == {"filter", "sort", "limit"} + + # Match with no query params - should only extract path params + params = template.matches("test://electronics/1234") + assert params == {"category": "electronics", "id": "1234"} + + # Match with all query params + params = template.matches("test://electronics/1234?filter=new&sort=price&limit=20") + assert params == { + "category": "electronics", + "id": "1234", + "filter": "new", + "sort": "price", + "limit": "20", + } + + # Match with partial query params + params = template.matches("test://electronics/1234?filter=new&sort=price") + assert params == { + "category": "electronics", + "id": "1234", + "filter": "new", + "sort": "price", + } + + # Match with unknown query params - should ignore + params = template.matches("test://electronics/1234?filter=new&unknown=value") + assert params == {"category": "electronics", "id": "1234", "filter": "new"} + + def test_form_style_query_validation(self): + """Test validation of form-style query parameters.""" + + # Valid: query params are subset of optional params + def valid_func(key: str, opt1: str = "default", opt2: int = 10, opt3: bool = False) -> str: + return f"{key}-{opt1}-{opt2}-{opt3}" + + template = ResourceTemplate.from_function( + fn=valid_func, + uri_template="test://{key}{?opt1,opt2}", + name="test", + ) + assert template.required_params == {"key"} + assert template.optional_params == {"opt1", "opt2", "opt3"} + + # Invalid: query param not optional in function + def invalid_func(key: str, required: str) -> str: + return f"{key}-{required}" + + with pytest.raises( + ValueError, + match="Mismatch between URI path parameters .* and required function parameters .*", + ): + ResourceTemplate.from_function( + fn=invalid_func, + uri_template="test://{key}{?required}", + name="test", + ) + + @pytest.mark.anyio + async def test_create_resource_with_form_style_query(self): + """Test creating resources with form-style query parameters.""" + + def item_func( + category: str, + id: str, + filter: str = "all", + sort: str = "name", + limit: int = 10, + ) -> dict[str, str | int]: + return { + "category": category, + "id": id, + "filter": filter, + "sort": sort, + "limit": limit, + } + + template = ResourceTemplate.from_function( + fn=item_func, + uri_template="items://{category}/{id}{?filter,sort,limit}", + name="item", + ) + + # Create with only required params + params = {"category": "electronics", "id": "1234"} + resource = await template.create_resource("items://electronics/1234", params) + result = await resource.read() + assert isinstance(result, str) + assert json.loads(result) == { + "category": "electronics", + "id": "1234", + "filter": "all", + "sort": "name", + "limit": 10, + } + + # Create with all params (limit will be string "20",Pydantic handles conversion) + uri = "items://electronics/1234?filter=new&sort=price&limit=20" + params = { + "category": "electronics", + "id": "1234", + "filter": "new", + "sort": "price", + "limit": "20", # value from URI is a string + } + resource = await template.create_resource(uri, params) + result = await resource.read() + assert isinstance(result, str) + assert json.loads(result) == { + "category": "electronics", + "id": "1234", + "filter": "new", + "sort": "price", + "limit": 20, # Pydantic converted "20" to 20 + } + + @pytest.mark.anyio + async def test_create_resource_optional_param_validation_fallback(self): + """ + Test that if optional parameters fail Pydantic validation, + their default values are used due to the + use_defaults_on_optional_validation_error decorator. + """ + + def func_with_optional_typed_params( + key: str, opt_int: int = 42, opt_bool: bool = True + ) -> dict[str, str | int | bool]: + return {"key": key, "opt_int": opt_int, "opt_bool": opt_bool} + + template = ResourceTemplate.from_function( + fn=func_with_optional_typed_params, + uri_template="test://{key}{?opt_int,opt_bool}", + name="test_optional_fallback", + ) + + # Case 1: opt_int is invalid, opt_bool is not provided + # URI like "test://mykey?opt_int=notanint" + params_invalid_int = {"key": "mykey", "opt_int": "notanint"} + resource1 = await template.create_resource("test://mykey?opt_int=notanint", params_invalid_int) + result1_str = await resource1.read() + result1 = json.loads(result1_str) + assert result1["key"] == "mykey" + assert result1["opt_int"] == 42 # Default used + assert result1["opt_bool"] is True # Default used + + # Case 2: opt_bool is invalid, opt_int is valid + # URI like "test://mykey?opt_int=100&opt_bool=notabool" + params_invalid_bool = { + "key": "mykey", + "opt_int": "100", # Valid string for int + "opt_bool": "notabool", + } + resource2 = await template.create_resource("test://mykey?opt_int=100&opt_bool=notabool", params_invalid_bool) + result2_str = await resource2.read() + result2 = json.loads(result2_str) + assert result2["key"] == "mykey" + assert result2["opt_int"] == 100 # Provided valid value used + assert result2["opt_bool"] is True # Default used + + # Case 3: Both opt_int and opt_bool are invalid + # URI like "test://mykey?opt_int=bad&opt_bool=bad" + params_both_invalid = { + "key": "mykey", + "opt_int": "bad", + "opt_bool": "bad", + } + resource3 = await template.create_resource("test://mykey?opt_int=bad&opt_bool=bad", params_both_invalid) + result3_str = await resource3.read() + result3 = json.loads(result3_str) + assert result3["key"] == "mykey" + assert result3["opt_int"] == 42 # Default used + assert result3["opt_bool"] is True # Default used + + # Case 4: Empty value for opt_int (should fall back to default) + # URI like "test://mykey?opt_int=" + params_empty_int = {"key": "mykey"} + resource4 = await template.create_resource("test://mykey?opt_int=", params_empty_int) + result4_str = await resource4.read() + result4 = json.loads(result4_str) + assert result4["key"] == "mykey" + assert result4["opt_int"] == 42 # Default used + assert result4["opt_bool"] is True # Default used + + # Case 5: Empty value for opt_bool (should fall back to default) + # URI like "test://mykey?opt_bool=" + params_empty_bool = {"key": "mykey"} + resource5 = await template.create_resource("test://mykey?opt_bool=", params_empty_bool) + result5_str = await resource5.read() + result5 = json.loads(result5_str) + assert result5["key"] == "mykey" + assert result5["opt_int"] == 42 # Default used + assert result5["opt_bool"] is True # Default used + + # Case 6: Optional string param with empty value, should use default value + def func_opt_str(key: str, opt_s: str = "default_val") -> dict[str, str]: + return {"key": key, "opt_s": opt_s} + + template_str = ResourceTemplate.from_function( + fn=func_opt_str, uri_template="test://{key}{?opt_s}", name="test_opt_str" + ) + params_empty_str = {"key": "mykey"} + resource6 = await template_str.create_resource("test://mykey?opt_s=", params_empty_str) + result6_str = await resource6.read() + result6 = json.loads(result6_str) + assert result6["key"] == "mykey" + assert result6["opt_s"] == "default_val" # Pydantic allows empty string for str type + + @pytest.mark.anyio + async def test_create_resource_required_param_validation_error(self): + """ + Test that if a required parameter fails Pydantic validation, an error is raised + and not suppressed by the new decorator. + """ + + def func_with_required_typed_param(req_int: int, key: str) -> dict[str, int | str]: + return {"req_int": req_int, "key": key} + + template = ResourceTemplate.from_function( + fn=func_with_required_typed_param, + uri_template="test://{key}/{req_int}", # req_int is part of path + name="test_req_error", + ) + + # req_int is "notanint", which is invalid for int type + params_invalid_req = {"key": "mykey", "req_int": "notanint"} + with pytest.raises(ValueError, match="Error creating resource from template"): + # This ValueError comes from ResourceTemplate.create_resource own try-except + # which catches Pydantic's ValidationError. + await template.create_resource("test://mykey/notanint", params_invalid_req) diff --git a/tests/server/fastmcp/test_func_metadata.py b/tests/server/fastmcp/test_func_metadata.py index 830cf816b..817032ef1 100644 --- a/tests/server/fastmcp/test_func_metadata.py +++ b/tests/server/fastmcp/test_func_metadata.py @@ -3,16 +3,19 @@ # pyright: reportMissingParameterType=false # pyright: reportUnknownArgumentType=false # pyright: reportUnknownLambdaType=false -from collections.abc import Callable +from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import Annotated, Any, TypedDict import annotated_types import pytest from dirty_equals import IsPartialDict -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, ValidationError, validate_call -from mcp.server.fastmcp.utilities.func_metadata import func_metadata +from mcp.server.fastmcp.utilities.func_metadata import ( + func_metadata, + use_defaults_on_optional_validation_error, +) class SomeInputModelA(BaseModel): @@ -1094,3 +1097,118 @@ def func_with_reserved_json( assert result["json"] == {"nested": "data"} assert result["model_dump"] == [1, 2, 3] assert result["normal"] == "plain string" + + +# Test functions for use_defaults_on_optional_validation_error decorator + + +def sync_func_for_decorator(req_param: str, opt_int: int = 10, opt_bool: bool = False) -> dict[str, str | int | bool]: + return {"req_param": req_param, "opt_int": opt_int, "opt_bool": opt_bool} + + +async def async_func_for_decorator(req_param: str, opt_int: int = 20, opt_str: str = "default") -> dict[str, str | int]: + return {"req_param": req_param, "opt_int": opt_int, "opt_str": opt_str} + + +class TestUseDefaultsOnOptionalValidationErrorDecorator: + @pytest.fixture + def decorated_sync_func(self) -> Callable[..., dict[str, str | int | bool]]: + # Apply validate_call first, then our decorator + return use_defaults_on_optional_validation_error(validate_call(sync_func_for_decorator)) + + @pytest.fixture + def decorated_async_func(self) -> Callable[..., Awaitable[dict[str, str | int]]]: + # Apply validate_call first, then our decorator + return use_defaults_on_optional_validation_error(validate_call(async_func_for_decorator)) + + def test_sync_all_valid(self, decorated_sync_func: Callable[..., dict[str, str | int | bool]]): + result = decorated_sync_func(req_param="test", opt_int=100, opt_bool=True) + assert result == {"req_param": "test", "opt_int": 100, "opt_bool": True} + + def test_sync_omit_optionals(self, decorated_sync_func: Callable[..., dict[str, str | int | bool]]): + result = decorated_sync_func(req_param="test") + assert result == {"req_param": "test", "opt_int": 10, "opt_bool": False} + + def test_sync_invalid_opt_int(self, decorated_sync_func: Callable[..., dict[str, str | int | bool]]): + # opt_int="bad" should cause ValidationError, decorator catches, uses default 10 + result = decorated_sync_func(req_param="test", opt_int="bad") + assert result == {"req_param": "test", "opt_int": 10, "opt_bool": False} + + def test_sync_invalid_opt_bool(self, decorated_sync_func: Callable[..., dict[str, str | int | bool]]): + # opt_bool="bad" should cause ValidationError, decorator catches, uses default False + result = decorated_sync_func(req_param="test", opt_bool="bad") + assert result == {"req_param": "test", "opt_int": 10, "opt_bool": False} + + def test_sync_invalid_opt_int_and_valid_opt_bool( + self, decorated_sync_func: Callable[..., dict[str, str | int | bool]] + ): + result = decorated_sync_func(req_param="test", opt_int="bad", opt_bool=True) + assert result == {"req_param": "test", "opt_int": 10, "opt_bool": True} + + def test_sync_all_optionals_invalid(self, decorated_sync_func: Callable[..., dict[str, str | int | bool]]): + result = decorated_sync_func(req_param="test", opt_int="bad", opt_bool="bad") + assert result == {"req_param": "test", "opt_int": 10, "opt_bool": False} + + def test_sync_required_param_missing(self, decorated_sync_func: Callable[..., dict[str, str | int | bool]]): + with pytest.raises(ValidationError): + decorated_sync_func(opt_int=100) # Missing req_param + + def test_sync_required_param_invalid(self, decorated_sync_func): + # If req_param itself was typed, e.g., req_param: int, and we passed "bad" + # For this test, sync_func_for_decorator has req_param: str, which is flexible. + # Let's define a quick one for this specific case. + def temp_sync_func(req_int_param: int, opt_str: str = "s") -> dict[str, int | str]: + return {"req_int_param": req_int_param, "opt_str": opt_str} + + decorated_temp_func = use_defaults_on_optional_validation_error(validate_call(temp_sync_func)) + with pytest.raises(ValidationError): + decorated_temp_func(req_int_param="notanint") + + @pytest.mark.anyio + async def test_async_all_valid(self, decorated_async_func: Callable[..., Awaitable[dict[str, str | int]]]): + result = await decorated_async_func(req_param="async_test", opt_int=200, opt_str="custom") + assert result == { + "req_param": "async_test", + "opt_int": 200, + "opt_str": "custom", + } + + @pytest.mark.anyio + async def test_async_omit_optionals(self, decorated_async_func: Callable[..., Awaitable[dict[str, str | int]]]): + result = await decorated_async_func(req_param="async_test") + assert result == { + "req_param": "async_test", + "opt_int": 20, + "opt_str": "default", + } + + @pytest.mark.anyio + async def test_async_invalid_opt_int(self, decorated_async_func: Callable[..., Awaitable[dict[str, str | int]]]): + result = await decorated_async_func(req_param="async_test", opt_int="bad") + assert result == { + "req_param": "async_test", + "opt_int": 20, # Default + "opt_str": "default", + } + + @pytest.mark.anyio + async def test_async_invalid_opt_str_but_is_int( + self, decorated_async_func: Callable[..., Awaitable[dict[str, str | int]]] + ): + # opt_str=123 (int) for str type should cause ValidationError, decorator uses default "default" + # Note: pydantic's validate_call might auto-convert int to str if not in strict mode. + # Let's assume default strictness where int is not directly valid for str. + # If validate_call is not strict, this test might need adjustment or a stricter type. + result = await decorated_async_func(req_param="async_test", opt_str=123) + assert result == { + "req_param": "async_test", + "opt_int": 20, + "opt_str": "default", # Default + } + + @pytest.mark.anyio + async def test_async_required_param_missing( + self, decorated_async_func: Callable[..., Awaitable[dict[str, str | int]]] + ): + with pytest.raises(ValidationError): + await decorated_async_func(opt_int=100) diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py index 3f921b588..9cd28752d 100644 --- a/tests/server/fastmcp/test_server.py +++ b/tests/server/fastmcp/test_server.py @@ -1,4 +1,5 @@ import base64 +import json from pathlib import Path from typing import TYPE_CHECKING, Any from unittest.mock import patch @@ -677,6 +678,41 @@ async def test_file_resource_binary(self, tmp_path: Path): assert isinstance(result.contents[0], BlobResourceContents) assert result.contents[0].blob == base64.b64encode(b"Binary file data").decode() + @pytest.mark.anyio + async def test_resource_with_form_style_query(self): + """Test that resources with form-style query expansion work correctly""" + mcp = FastMCP() + + @mcp.resource("resource://{category}/{id}{?filter,sort,limit}") + def get_item( + category: str, + id: str, + filter: str = "all", + sort: str = "name", + limit: int = 10, + ) -> str: + return f"Item {id} in {category}, filtered by {filter}, sorted by {sort}, limited to {limit}" + + async with client_session(mcp._mcp_server) as client: + # Test with default values + result = await client.read_resource(AnyUrl("resource://electronics/1234")) + assert isinstance(result.contents[0], TextResourceContents) + assert result.contents[0].text == "Item 1234 in electronics, filtered by all, sorted by name, limited to 10" + + # Test with query parameters + result = await client.read_resource(AnyUrl("resource://electronics/1234?filter=new&sort=price&limit=20")) + assert isinstance(result.contents[0], TextResourceContents) + assert ( + result.contents[0].text == "Item 1234 in electronics, filtered by new, sorted by price, limited to 20" + ) + + # Test with partial query parameters + result = await client.read_resource(AnyUrl("resource://electronics/1234?filter=used")) + assert isinstance(result.contents[0], TextResourceContents) + assert ( + result.contents[0].text == "Item 1234 in electronics, filtered by used, sorted by name, limited to 10" + ) + @pytest.mark.anyio async def test_function_resource(self): mcp = FastMCP() @@ -703,7 +739,10 @@ async def test_resource_with_params(self): parameters don't match""" mcp = FastMCP() - 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://data") def get_data_fn(param: str) -> str: @@ -714,7 +753,10 @@ async def test_resource_with_uri_params(self): """Test that a resource with URI parameters is automatically a template""" mcp = FastMCP() - 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://{param}") def get_data() -> str: @@ -743,12 +785,40 @@ def get_data(name: str) -> str: assert isinstance(result.contents[0], TextResourceContents) assert result.contents[0].text == "Data for test" + @pytest.mark.anyio + async def test_resource_with_optional_params(self): + """Test that resources with optional parameters work correctly""" + mcp = FastMCP() + + @mcp.resource("resource://{name}/data") + def get_data_with_options(name: str, format: str = "text", limit: int = 10) -> str: + return f"Data for {name} in {format} format with limit {limit}" + + async with client_session(mcp._mcp_server) as client: + # Test with default values + result = await client.read_resource(AnyUrl("resource://test/data")) + assert isinstance(result.contents[0], TextResourceContents) + assert result.contents[0].text == "Data for test in text format with limit 10" + + # Test with query parameters + result = await client.read_resource(AnyUrl("resource://test/data?format=json&limit=20")) + assert isinstance(result.contents[0], TextResourceContents) + assert result.contents[0].text == "Data for test in json format with limit 20" + + # Test with partial query parameters + result = await client.read_resource(AnyUrl("resource://test/data?format=xml")) + assert isinstance(result.contents[0], TextResourceContents) + assert result.contents[0].text == "Data for test in xml format with limit 10" + @pytest.mark.anyio async def test_resource_mismatched_params(self): """Test that mismatched parameters raise an error""" mcp = FastMCP() - 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://{name}/data") def get_data(user: str) -> str: @@ -773,7 +843,10 @@ async def test_resource_multiple_mismatched_params(self): """Test that mismatched parameters raise an error""" mcp = FastMCP() - 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://{org}/{repo}/data") def get_data_mismatched(org: str, repo_2: str) -> str: @@ -810,6 +883,116 @@ def get_data(name: str) -> str: result = await resource.read() assert result == "Data for test" + @pytest.mark.anyio + async def test_resource_optional_param_validation_fallback_and_url_encoding( + self, + ): + """Test handling of optional param validation fallback & URL encoding.""" + mcp = FastMCP() + + @mcp.resource("resource://test_item/{item_id}{?name,count,active}") + def get_test_item_details( + item_id: str, + name: str = "default_name", + count: int = 0, + active: bool = False, + ) -> dict[str, str | int | bool]: + return { + "item_id": item_id, + "name": name, + "count": count, + "active": active, + } + + async with client_session(mcp._mcp_server) as client: + # 1. All defaults + res1_uri = "resource://test_item/item001" + res1_content_result = await client.read_resource(AnyUrl(res1_uri)) + assert res1_content_result.contents and isinstance(res1_content_result.contents[0], TextResourceContents) + data1 = json.loads(res1_content_result.contents[0].text) + assert data1 == { + "item_id": "item001", + "name": "default_name", + "count": 0, + "active": False, + } + + # 2. Valid optional params (name is URL encoded) + res2_uri = "resource://test_item/item002?name=My%20Product&count=10&active=true" + res2_content_result = await client.read_resource(AnyUrl(res2_uri)) + assert res2_content_result.contents and isinstance(res2_content_result.contents[0], TextResourceContents) + data2 = json.loads(res2_content_result.contents[0].text) + assert data2 == { + "item_id": "item002", + "name": "My Product", # Decoded + "count": 10, + "active": True, + } + + # 3. Invalid 'count' (optional int), valid 'name', 'active' not provided + # count=notanint should make it use default_count = 0 + res3_uri = "resource://test_item/item003?name=Another%20Item&count=notanint" + res3_content_result = await client.read_resource(AnyUrl(res3_uri)) + assert res3_content_result.contents and isinstance(res3_content_result.contents[0], TextResourceContents) + data3 = json.loads(res3_content_result.contents[0].text) + assert data3 == { + "item_id": "item003", + "name": "Another Item", + "count": 0, # Fallback to default + "active": False, # Fallback to default + } + + # 4. Invalid 'active' (optional bool), valid 'count', 'name' not provided + # active=notabool should make it use default_active = False + res4_uri = "resource://test_item/item004?count=50&active=notabool" + res4_content_result = await client.read_resource(AnyUrl(res4_uri)) + assert res4_content_result.contents and isinstance(res4_content_result.contents[0], TextResourceContents) + data4 = json.loads(res4_content_result.contents[0].text) + assert data4 == { + "item_id": "item004", + "name": "default_name", # Fallback to default + "count": 50, + "active": False, # Fallback to default + } + + # 5. Empty value for optional 'name' (string type) + # name= (empty value) should fall back to default + res5_uri = "resource://test_item/item005?name=" + res5_content_result = await client.read_resource(AnyUrl(res5_uri)) + assert res5_content_result.contents and isinstance(res5_content_result.contents[0], TextResourceContents) + data5 = json.loads(res5_content_result.contents[0].text) + assert data5 == { + "item_id": "item005", + "name": "default_name", # Fallback to default + "count": 0, + "active": False, + } + + # 6. Empty value for optional 'count' (int type) + # count= (empty value) should fall back to default + res6_uri = "resource://test_item/item006?count=" + res6_content_result = await client.read_resource(AnyUrl(res6_uri)) + assert res6_content_result.contents and isinstance(res6_content_result.contents[0], TextResourceContents) + data6 = json.loads(res6_content_result.contents[0].text) + assert data6 == { + "item_id": "item006", + "name": "default_name", + "count": 0, # Fallback to default because param is removed by parse_qs + "active": False, + } + + # Test required param failing validation at server level + @mcp.resource("resource://req_fail/{req_id}/details") + def get_req_details(req_id: int, detail_type: str = "summary") -> dict[str, str | int]: + return {"req_id": req_id, "detail_type": detail_type} + + async with client_session(mcp._mcp_server) as client: + invalid_req_uri = "resource://req_fail/notanint/details" + # The FastMCP.read_resource wraps internal errors, + # from template.create_resource, into a ResourceError, as McpError. + with pytest.raises(McpError, match="Error creating resource from template"): + await client.read_resource(AnyUrl(invalid_req_uri)) + class TestContextInjection: """Test context injection in tools."""