merge conflict

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 == "<lambda>":
             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()

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}'