diff --git a/src/fastmcp/prompts/prompt.py b/src/fastmcp/prompts/prompt.py index d9bae2b9a..cc1be6bde 100644 --- a/src/fastmcp/prompts/prompt.py +++ b/src/fastmcp/prompts/prompt.py @@ -3,14 +3,22 @@ from __future__ import annotations as _annotations import inspect +import json from collections.abc import Awaitable, Callable, Sequence -from typing import TYPE_CHECKING, Annotated, Any +from typing import TYPE_CHECKING, Annotated, Any, get_origin import pydantic_core from mcp.types import EmbeddedResource, ImageContent, PromptMessage, Role, TextContent from mcp.types import Prompt as MCPPrompt from mcp.types import PromptArgument as MCPPromptArgument -from pydantic import BaseModel, BeforeValidator, Field, TypeAdapter, validate_call +from pydantic import ( + BaseModel, + BeforeValidator, + Field, + PrivateAttr, + TypeAdapter, + validate_call, +) from fastmcp.exceptions import PromptError from fastmcp.server.dependencies import get_context @@ -78,6 +86,7 @@ class Prompt(BaseModel): None, description="Arguments that can be passed to the prompt" ) fn: Callable[..., PromptResult | Awaitable[PromptResult]] + _original_param_types: dict[str, Any] = PrivateAttr(default_factory=dict) @classmethod def from_function( @@ -97,12 +106,13 @@ class Prompt(BaseModel): """ from fastmcp.server.context import Context - func_name = name or getattr(fn, "__name__", None) or fn.__class__.__name__ + original_fn_for_signature = fn + func_name = name or original_fn_for_signature.__name__ or fn.__class__.__name__ if func_name == "": raise ValueError("You must provide a name for lambda functions") # Reject functions with *args or **kwargs - sig = inspect.signature(fn) + sig = inspect.signature(original_fn_for_signature) for param in sig.parameters.values(): if param.kind == inspect.Parameter.VAR_POSITIONAL: raise ValueError("Functions with *args are not supported as prompts") @@ -120,7 +130,9 @@ class Prompt(BaseModel): # Auto-detect context parameter if not provided - context_kwarg = find_kwarg_by_type(fn, kwarg_type=Context) + context_kwarg = find_kwarg_by_type( + original_fn_for_signature, kwarg_type=Context + ) if context_kwarg: prune_params = [context_kwarg] else: @@ -140,16 +152,25 @@ class Prompt(BaseModel): ) ) - # ensure the arguments are properly cast - fn = validate_call(fn) + # ensure the arguments are properly cast by Pydantic's validate_call + validated_fn = validate_call(original_fn_for_signature) - return cls( + # Store original parameter types + original_param_types_dict = { + p.name: p.annotation + for p in sig.parameters.values() + if p.annotation != inspect.Parameter.empty + } + + instance = cls( name=func_name, - description=description, + description=description or original_fn_for_signature.__doc__, arguments=arguments, - fn=fn, + fn=validated_fn, tags=tags or set(), ) + instance._original_param_types = original_param_types_dict + return instance async def render( self, @@ -167,8 +188,48 @@ class Prompt(BaseModel): raise ValueError(f"Missing required arguments: {missing}") try: - # Prepare arguments with context + # Prepare arguments kwargs = arguments.copy() if arguments else {} + + # <<< NEW: Attempt to deserialize JSON strings for complex types >>> + if self._original_param_types: + for param_name, param_value in list(kwargs.items()): + if param_name in self._original_param_types and isinstance( + param_value, str + ): + target_type_hint = self._original_param_types[param_name] + + # Determine the actual base type to check (e.g., list from list[float]) + origin_type = get_origin(target_type_hint) + # Fallback to the hint itself if no origin (e.g. for non-generic BaseModel) + type_to_check_for_complex = ( + origin_type if origin_type else target_type_hint + ) + + is_json_candidate = False + if type_to_check_for_complex in (list, dict): + is_json_candidate = True + elif inspect.isclass(type_to_check_for_complex) and issubclass( + type_to_check_for_complex, BaseModel + ): + # BaseModel itself is imported from pydantic, so this check is fine + is_json_candidate = True + + if is_json_candidate: + try: + kwargs[param_name] = json.loads(param_value) + logger.debug( + f"FastMCP: Auto-deserialized JSON string for param '{param_name}' in prompt '{self.name}'." + ) + except json.JSONDecodeError: + # Not valid JSON, pass the original string to Pydantic validation + logger.debug( + f"FastMCP: Param '{param_name}' for prompt '{self.name}' is a string " + "but not valid JSON. Passing as string to Pydantic validation." + ) + # <<< END NEW LOGIC >>> + + # Prepare arguments with context context_kwarg = find_kwarg_by_type(self.fn, kwarg_type=Context) if context_kwarg and context_kwarg not in kwargs: kwargs[context_kwarg] = get_context() diff --git a/tests/prompts/test_prompt.py b/tests/prompts/test_prompt.py index a0cda7b2d..92d9e99d9 100644 --- a/tests/prompts/test_prompt.py +++ b/tests/prompts/test_prompt.py @@ -3,6 +3,7 @@ from mcp.types import EmbeddedResource, TextResourceContents from pydantic import FileUrl from fastmcp.prompts.prompt import ( + BaseModel, Message, Prompt, PromptMessage, @@ -10,6 +11,11 @@ from fastmcp.prompts.prompt import ( ) +class MyTestModel(BaseModel): + key: str + value: int + + class TestRenderPrompt: async def test_basic_fn(self): def fn() -> str: @@ -240,3 +246,67 @@ class TestRenderPrompt: ), ) ] + + async def test_render_with_json_string_list_arg(self): + """Test that JSON string for a list argument is auto-deserialized.""" + + def prompt_with_list(my_list: list[int]) -> str: + return f"List sum: {sum(my_list)}" + + prompt = Prompt.from_function(prompt_with_list) + rendered_messages = await prompt.render(arguments={"my_list": "[1, 2, 3, 4]"}) + assert len(rendered_messages) == 1 + assert isinstance(rendered_messages[0].content, TextContent) + assert rendered_messages[0].content.text == "List sum: 10" + + async def test_render_with_json_string_dict_arg(self): + """Test that JSON string for a dict argument is auto-deserialized.""" + + def prompt_with_dict(my_dict: dict[str, int]) -> str: + return f"Value for 'b': {my_dict.get('b')}" + + prompt = Prompt.from_function(prompt_with_dict) + rendered_messages = await prompt.render( + arguments={"my_dict": '{"a": 1, "b": 2}'} + ) # escaped JSON string + assert len(rendered_messages) == 1 + assert isinstance(rendered_messages[0].content, TextContent) + assert rendered_messages[0].content.text == "Value for 'b': 2" + + async def test_render_with_json_string_basemodel_arg(self): + """Test that JSON string for a Pydantic BaseModel argument is auto-deserialized.""" + + def prompt_with_model(my_model: MyTestModel) -> str: + return f"Model: {my_model.key}={my_model.value}" + + prompt = Prompt.from_function(prompt_with_model) + rendered_messages = await prompt.render( + arguments={"my_model": '{"key": "test", "value": 123}'} + ) # escaped JSON string + assert len(rendered_messages) == 1 + assert isinstance(rendered_messages[0].content, TextContent) + assert rendered_messages[0].content.text == "Model: test=123" + + async def test_render_with_malformed_json_string_arg(self): + """Test that a malformed JSON string for a list arg is passed as string (and Pydantic errors).""" + + def prompt_with_list(my_list: list[int]) -> str: + return f"List sum: {sum(my_list)}" + + prompt = Prompt.from_function(prompt_with_list) + with pytest.raises( + ValueError, match="Error rendering prompt prompt_with_list." + ): + await prompt.render(arguments={"my_list": "not a valid json list"}) + + async def test_render_with_non_json_string_for_string_arg(self): + """Test that a regular string for a string argument is not json.loads-ed.""" + + def prompt_with_string(my_string: str) -> str: + return f"String: {my_string}" + + prompt = Prompt.from_function(prompt_with_string) + rendered_messages = await prompt.render(arguments={"my_string": '{"a": 1}'}) + assert len(rendered_messages) == 1 + assert isinstance(rendered_messages[0].content, TextContent) + assert rendered_messages[0].content.text == 'String: {"a": 1}'