diff --git a/README.md b/README.md index 4c8bb90dc..250979f81 100644 --- 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. \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 0a11a3b15..e8156ea3a 100644 --- 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" -] +] \ No newline at end of file diff --git a/src/mcp/server/fastmcp/resources/templates.py b/src/mcp/server/fastmcp/resources/templates.py index a30b18253..4edb64638 100644 --- 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 == "": 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 index e5b6c3acc..ce1a1518e 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, 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, diff --git a/src/mcp/server/fastmcp/utilities/func_metadata.py b/src/mcp/server/fastmcp/utilities/func_metadata.py index 374391325..320115f79 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 @@ -7,7 +8,14 @@ ForwardRef, ) -from pydantic import BaseModel, ConfigDict, Field, WithJsonSchema, create_model +from pydantic import ( + BaseModel, + ConfigDict, + Field, + ValidationError, + WithJsonSchema, + create_model, +) from pydantic._internal._typing_extra import eval_type_backport from pydantic.fields import FieldInfo from pydantic_core import PydanticUndefined @@ -212,3 +220,118 @@ def _get_typed_signature(call: Callable[..., Any]) -> inspect.Signature: ] typed_signature = inspect.Signature(typed_params) return typed_signature + + +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 3c17cd559..65a973da6 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,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) @@ -118,3 +171,115 @@ def get_user_profile(user_id: str) -> str: 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: + 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: # 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 f47244361..b07bfa9c3 100644 --- a/tests/server/fastmcp/resources/test_resource_template.py +++ b/tests/server/fastmcp/resources/test_resource_template.py @@ -46,6 +46,68 @@ def my_func(key: str, value: int) -> dict: 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: + 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 +248,307 @@ 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: + 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: + 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: + 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: + 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: + 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: + 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 b1828ffe9..53f165101 100644 --- a/tests/server/fastmcp/test_func_metadata.py +++ b/tests/server/fastmcp/test_func_metadata.py @@ -2,9 +2,12 @@ import annotated_types import pytest -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): @@ -414,3 +417,124 @@ def func_with_str_and_int(a: str, b: int): result = meta.pre_parse_json({"a": "123", "b": 123}) assert result["a"] == "123" assert result["b"] == 123 + + +# 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: + 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: + return {"req_param": req_param, "opt_int": opt_int, "opt_str": opt_str} + + +class TestUseDefaultsOnOptionalValidationErrorDecorator: + @pytest.fixture + def decorated_sync_func(self): + # 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): + # 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): + 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): + 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): + # 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): + # 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): + 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): + 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): + 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: + 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): + 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): + 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): + 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): + # 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): + 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 b817761ea..76f5a0d7c 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 from unittest.mock import patch @@ -441,6 +442,56 @@ async def test_file_resource_binary(self, tmp_path: Path): == 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}, " + f"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() @@ -467,7 +518,11 @@ 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: @@ -478,7 +533,11 @@ 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: @@ -507,12 +566,53 @@ 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: @@ -539,7 +639,11 @@ 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: @@ -576,6 +680,130 @@ 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: + 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: + 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."""