From b2f5551d229ba85dc6c134e187daa24f21fa7c93 Mon Sep 17 00:00:00 2001 From: Bill Easton Date: Sun, 1 Feb 2026 20:28:44 -0600 Subject: [PATCH] Fix Field() handling in prompts (#3050) Co-authored-by: Bill Easton Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com> --- src/fastmcp/prompts/function_prompt.py | 18 ++++- tests/prompts/test_prompt.py | 82 +++++++++++++++++++++++ tests/resources/test_resource_template.py | 80 ++++++++++++++++++++++ 3 files changed, 178 insertions(+), 2 deletions(-) diff --git a/src/fastmcp/prompts/function_prompt.py b/src/fastmcp/prompts/function_prompt.py index 4648c269f..c2ba6f44a 100644 --- a/src/fastmcp/prompts/function_prompt.py +++ b/src/fastmcp/prompts/function_prompt.py @@ -297,13 +297,27 @@ class FunctionPrompt(Prompt): # Convert string arguments to expected types BEFORE validation kwargs = self._convert_string_arguments(kwargs) + # Filter out arguments that aren't in the function signature + # This is important for security: dependencies should not be overridable + # from external callers. self.fn is wrapped by without_injected_parameters, + # so we only accept arguments that are in the wrapped function's signature. + sig = inspect.signature(self.fn) + valid_params = set(sig.parameters.keys()) + kwargs = {k: v for k, v in kwargs.items() if k in valid_params} + + # Use type adapter to validate arguments and handle Field() defaults + # This matches the behavior of tools in function_tool + type_adapter = get_cached_typeadapter(self.fn) + # self.fn is wrapped by without_injected_parameters which handles # dependency resolution internally if inspect.iscoroutinefunction(self.fn): - result = await self.fn(**kwargs) + result = await type_adapter.validate_python(kwargs) else: # Run sync functions in threadpool to avoid blocking the event loop - result = await call_sync_fn_in_threadpool(self.fn, **kwargs) + result = await call_sync_fn_in_threadpool( + type_adapter.validate_python, kwargs + ) # Handle sync wrappers that return awaitables (e.g., partial(async_fn)) if inspect.isawaitable(result): result = await result diff --git a/tests/prompts/test_prompt.py b/tests/prompts/test_prompt.py index 30c8336e7..05d7c0f17 100644 --- a/tests/prompts/test_prompt.py +++ b/tests/prompts/test_prompt.py @@ -552,6 +552,88 @@ class TestPromptResult: assert mcp_result.meta == {"key": "value"} +class TestPromptFieldDefaults: + """Test prompts with Field() defaults.""" + + async def test_field_with_default(self): + """Test that Field(default=...) correctly provides default values.""" + + from pydantic import Field + + def prompt_with_defaults( + required: str = Field(description="Required parameter"), + optional: str = Field( + default="default_value", description="Optional parameter" + ), + ) -> str: + return f"required={required}, optional={optional}" + + prompt = Prompt.from_function(prompt_with_defaults) + result = await prompt.render(arguments={"required": "test"}) + assert result.messages == [Message("required=test, optional=default_value")] + + async def test_annotated_field_with_default_in_signature(self): + """Test that Annotated[type, Field(...)] with default in signature works.""" + from typing import Annotated + + from pydantic import Field + + def prompt_with_annotated( + required: Annotated[str, Field(description="Required parameter")], + optional: Annotated[ + str, Field(description="Optional parameter") + ] = "default_value", + ) -> str: + return f"required={required}, optional={optional}" + + prompt = Prompt.from_function(prompt_with_annotated) + result = await prompt.render(arguments={"required": "test"}) + assert result.messages == [Message("required=test, optional=default_value")] + + async def test_multiple_field_defaults(self): + """Test multiple parameters with Field() defaults.""" + from pydantic import Field + + def prompt_with_multiple_defaults( + name: str = Field(description="Name"), + greeting: str = Field(default="Hello", description="Greeting"), + punctuation: str = Field(default="!", description="Punctuation"), + ) -> str: + return f"{greeting}, {name}{punctuation}" + + prompt = Prompt.from_function(prompt_with_multiple_defaults) + + # Test with only required parameter + result1 = await prompt.render(arguments={"name": "World"}) + assert result1.messages == [Message("Hello, World!")] + + # Test overriding one default + result2 = await prompt.render(arguments={"name": "World", "greeting": "Hi"}) + assert result2.messages == [Message("Hi, World!")] + + # Test overriding all defaults + result3 = await prompt.render( + arguments={"name": "World", "greeting": "Greetings", "punctuation": "."} + ) + assert result3.messages == [Message("Greetings, World.")] + + async def test_field_defaults_with_type_conversion(self): + """Test Field() defaults work with type conversion for non-string types.""" + from pydantic import Field + + def prompt_with_typed_defaults( + count: int = Field(description="Count"), + multiplier: int = Field(default=2, description="Multiplier"), + ) -> str: + return f"result={count * multiplier}" + + prompt = Prompt.from_function(prompt_with_typed_defaults) + + # Pass count as string (MCP requirement), should use default for multiplier + result = await prompt.render(arguments={"count": "5"}) + assert result.messages == [Message("result=10")] + + class TestPromptCallableAndConcurrency: """Test prompts with callable objects and concurrent execution.""" diff --git a/tests/resources/test_resource_template.py b/tests/resources/test_resource_template.py index 2cb1d1d4a..ed0b37376 100644 --- a/tests/resources/test_resource_template.py +++ b/tests/resources/test_resource_template.py @@ -1007,3 +1007,83 @@ class TestQueryParameterWithWildcards: assert result["path"] == "src/test/data.txt" assert result["encoding"] == "utf-8" # default assert result["lines"] == 50 # provided + + +class TestResourceTemplateFieldDefaults: + """Test resource templates with Field() defaults.""" + + async def test_field_with_default(self): + """Test that Field(default=...) correctly provides default values in resource templates.""" + from pydantic import Field + + def get_data( + id: str = Field(description="Resource ID"), + format: str = Field(default="json", description="Output format"), + ) -> str: + return f"id={id}, format={format}" + + template = ResourceTemplate.from_function( + fn=get_data, + uri_template="data://{id}{?format}", + name="test", + ) + + # Test with only required parameter + resource = await template.create_resource("data://123", {"id": "123"}) + result = await resource.read() + assert result == "id=123, format=json" + + # Test with override + resource = await template.create_resource( + "data://123?format=xml", {"id": "123", "format": "xml"} + ) + result = await resource.read() + assert result == "id=123, format=xml" + + async def test_multiple_field_defaults(self): + """Test multiple query parameters with Field() defaults.""" + from typing import Any + + from pydantic import Field + + def fetch_data( + resource_id: str = Field(description="Resource ID"), + limit: int = Field(default=10, description="Result limit"), + offset: int = Field(default=0, description="Result offset"), + format: str = Field(default="json", description="Output format"), + ) -> dict[str, Any]: + return { + "resource_id": resource_id, + "limit": limit, + "offset": offset, + "format": format, + } + + template = ResourceTemplate.from_function( + fn=fetch_data, + uri_template="api://{resource_id}{?limit,offset,format}", + name="test", + ) + + # Test with only required parameter - all defaults should apply + resource1 = await template.create_resource( + "api://user123", {"resource_id": "user123"} + ) + result1 = await resource1.read() + assert isinstance(result1, dict) + assert result1["resource_id"] == "user123" + assert result1["limit"] == 10 + assert result1["offset"] == 0 + assert result1["format"] == "json" + + # Test with some overrides + resource2 = await template.create_resource( + "api://user123?limit=50&format=xml", + {"resource_id": "user123", "limit": "50", "format": "xml"}, + ) + result2 = await resource2.read() + assert isinstance(result2, dict) + assert result2["resource_id"] == "user123" + assert result2["limit"] == 50 # overridden + assert result2["offset"] == 0 # default + assert result2["format"] == "xml" # overridden