Compare commits

...

4 commits

Author SHA1 Message Date
zzstoatzz
8f1bae544e ok 2025-05-27 19:38:32 -05:00
zzstoatzz
7d2f1f9671 Fix: Prompt arg handling & docs for #541
Addresses #541:
- Server now auto-deserializes JSON string args (list, dict, BaseModel) for prompts. This simplifies server-side prompt logic by reducing boilerplate `json.loads()` calls.
- Docs updated to clarify `list_resource_templates` usage for templatized resources.
- Docs updated to require client-side `json.dumps()` for complex `get_prompt` arguments, resolving the original Pydantic error.
- Adds a new example (`examples/dynamic_story_prompt/`) demonstrating the server-side deserialization benefit and correct client-side serialization.

Closes #541.

---
Notes for Reviewers:
- **Server-Side Auto-Deserialization:** This change introduces a "magic" `json.loads()` in `Prompt.render`. This is an intentional DX improvement. It only triggers for `str` inputs targeting `list`, `dict`, or `BaseModel` type hints. If `json.loads()` fails (e.g., malformed JSON), the original string is passed to Pydantic's `validate_call`, ensuring robust error handling. This avoids boilerplate in user prompt functions.
- **Client `get_prompt()` Return Value:** The `examples/dynamic_story_prompt/story_client.py` parses the result of `client.get_prompt()` by iterating and looking for a `('messages', ...)` tuple. This reflects the observed behavior of the current `client.get_prompt()`. This commit does *not* change `client.get_prompt()`'s return behavior; the example merely adapts to it. A separate discussion might be warranted for potentially simplifying `client.get_prompt()`'s return signature in the future.
2025-05-27 19:38:32 -05:00
zzstoatzz
5aa19c381b also rules 2025-05-27 19:38:32 -05:00
zzstoatzz
d09fcc80eb merge conflict 2025-05-27 19:38:25 -05:00
10 changed files with 478 additions and 14 deletions

7
.cursor/rules/debug.mdc Normal file
View file

@ -0,0 +1,7 @@
---
description:
globs:
alwaysApply: false
---
- prefer using your shell tool to find and `rg` through site-packages to your web tool when you need to know details on upstream libraries
- iterative, empirical observations are better than rumination

10
.cursor/rules/mre.mdc Normal file
View file

@ -0,0 +1,10 @@
---
description:
globs:
alwaysApply: false
---
- put a MINIMAL reproducible example in repros/{issue_number}.py
- afterwards, change the library as needed, if necessary
- test against the MRE
- add unit tests (in a sensible place) if applicable
- after adding tests, critically review them and the library changes

View file

@ -229,6 +229,9 @@ The standard client methods return user-friendly representations that may change
templates = await client.list_resource_templates()
# templates -> list[mcp.types.ResourceTemplate]
```
<Tip>
If you define a resource with placeholders in its URI (e.g., `@mcp.resource("data://{user_id}/profile")`), it is considered a **resource template**. You will find it using `client.list_resource_templates()`. The `client.list_resources()` method is for resources defined with fully static URIs.
</Tip>
* **`read_resource(uri: str | AnyUrl)`**: Reads the content of a resource or a resolved template.
```python
# Read a static resource
@ -244,7 +247,20 @@ The standard client methods return user-friendly representations that may change
#### Prompt Operations
* **`list_prompts()`**: Retrieves available prompt templates.
* **`get_prompt(name: str, arguments: dict[str, Any] | None = None)`**: Retrieves a rendered prompt message list.
* **`get_prompt(name: str, arguments: dict[str, str] | None = None)`**: Retrieves a rendered prompt message list.
* **Important:** When passing complex data types (like lists or dictionaries) as values in the `arguments` dictionary, they **must be serialized to JSON strings**. For example, if your prompt expects a list of numbers for an argument named `data_points`, you should pass `{"data_points": json.dumps([1, 2, 3])}`.
* The FastMCP server will automatically attempt to deserialize these JSON strings back into appropriate Python objects (lists, dicts, Pydantic models) if your server-side prompt function is type-hinted accordingly. See the server-side [Prompts documentation](/servers/prompts#automatic-deserialization-of-string-arguments) for more details on this server behavior.
```python
import json
# Assuming 'analyze_data' prompt expects 'data_points' (list) and 'settings' (dict)
args = {
"data_points": json.dumps([1.0, 2.5, 3.0]),
"settings": json.dumps({"mode": "detailed", "threshold": 0.5})
}
prompt_messages = await client.get_prompt("analyze_data", args)
# prompt_messages -> list[mcp.types.PromptMessage]
```
### Raw MCP Protocol Objects

View file

@ -221,4 +221,104 @@ The duplicate behavior options are:
- `"warn"` (default): Logs a warning, and the new prompt replaces the old one.
- `"error"`: Raises a `ValueError`, preventing the duplicate registration.
- `"replace"`: Silently replaces the existing prompt with the new one.
- `"ignore"`: Keeps the original prompt and ignores the new registration attempt.
- `"ignore"`: Keeps the original prompt and ignores the new registration attempt.
### Automatic Deserialization of String Arguments
<VersionBadge version="2.4.1" />
FastMCP enhances developer experience by automatically handling the deserialization of string arguments for your prompt functions when they are intended to be complex Python types like lists, dictionaries, or Pydantic models.
When you define your prompt functions, you can use Pythonic type hints to specify the data structures you want to work with directly:
```python server.py
from fastmcp import FastMCP
from pydantic import BaseModel
mcp = FastMCP(name="PromptDemoServer")
class MyConfig(BaseModel):
param_a: str
param_b: int
is_active: bool = True
@mcp.prompt()
def process_user_data(
user_id: int,
preferences: list[str],
settings: MyConfig,
metadata: dict[str, str | int]
) -> str:
"""
Generates a prompt based on processed user data, preferences,
settings, and metadata.
"""
# Inside this function, you directly work with Python objects:
# - user_id is an int
# - preferences is a list of strings
# - settings is an instance of MyConfig
# - metadata is a dictionary
pref_string = ", ".join(preferences)
return (
f"User {user_id} preferences: {pref_string}. "
f"Settings: {settings.param_a} (Active: {settings.is_active}, Value: {settings.param_b}). "
f"Metadata count: {len(metadata)}."
)
```
**How it Works:**
If a client sends arguments for `preferences`, `settings`, or `metadata` as JSON strings (which is necessary for complex types due to the underlying MCP specification that expects `dict[str, str]` for prompt arguments see [Client documentation](/clients/client#prompt-operations)), FastMCP's server-side logic will:
1. Detect that the incoming argument is a string.
2. Check your function's type hint for that parameter (e.g., `list[str]`, `MyConfig`, `dict[str, str | int]`).
3. If the type hint is `list`, `dict`, or a Pydantic `BaseModel` (or a generic version like `list[Any]`), FastMCP attempts to parse the string using `json.loads()`.
4. **On Success:** Your function receives the deserialized Python object (a `list`, `dict`, or `MyConfig` instance).
5. **On `json.JSONDecodeError`:** If the string is not valid JSON, it's passed to your function as-is. Pydantic's standard validation will then take over, likely raising a `ValidationError` if the string doesn't match the expected type (e.g., a string like `"not-a-list"` for a `list` parameter).
This behavior significantly reduces boilerplate in your prompt functions, as you don't need to manually call `json.loads()` and handle potential errors for every complex parameter.
<details>
<summary>Hypothetical Client Example</summary>
```python client.py
import asyncio
import json
from fastmcp import Client
async def run_client():
client = Client("server.py") # Or appropriate transport to your server
async with client:
user_prefs = ["notifications", "dark_mode"]
user_settings = {"param_a": "profile_v2", "param_b": 42, "is_active": False}
user_metadata = {"login_count": 10, "last_ip": "192.168.1.100"}
client_args = {
"user_id": 123, # Simple types are passed directly
"preferences": json.dumps(user_prefs),
"settings": json.dumps(user_settings),
"metadata": json.dumps(user_metadata)
}
print(f"Sending to server: {client_args}")
try:
prompt_messages = await client.get_prompt("process_user_data", client_args)
for msg in prompt_messages:
print(f"Server responded with role '{msg.role}': {msg.content.text}")
except Exception as e:
print(f"An error occurred: {e}")
if __name__ == "__main__":
asyncio.run(run_client())
```
**Expected Client Output (if server responds directly with the generated string):**
```
# Sending to server: {'user_id': 123, 'preferences': '["notifications", "dark_mode"]', 'settings': '{"param_a": "profile_v2", "param_b": 42, "is_active": false}', 'metadata': '{"login_count": 10, "last_ip": "192.168.1.100"}'}
# Server responded with role 'user': User 123 preferences: notifications, dark_mode. Settings: profile_v2 (Active: False, Value: 42). Metadata count: 2.
```
</details>
## Advanced Usage

View file

@ -0,0 +1,37 @@
# Dynamic Story Prompt Generator Example
This example demonstrates how FastMCP's automatic JSON string deserialization for prompt arguments can simplify server-side prompt logic.
## Scenario
We have a FastMCP server with a prompt named `generate_dynamic_story_prompt`. This prompt is designed to create an engaging story starter based on several complex inputs:
1. **Character Details**: Information about the main character, structured as a Pydantic model (name, archetype, quirky trait).
2. **Mysterious Objects**: A list of unusual items the character stumbles upon.
3. **Active World Laws**: A dictionary describing peculiar rules or conditions currently affecting the story world.
The server-side prompt function takes these as a `Character` object, a `list[str]`, and a `dict[str, str]` respectively. It then combines them into a creative text prompt.
## Motivation for Auto-Deserialization
The developer writing the `generate_dynamic_story_prompt` function wants to work with these inputs as native Python objects for clarity and ease of use within their creative logic. They shouldn't need to manually parse JSON strings for each complex argument.
The client application (e.g., a web UI, another script) will gather this information and serialize the complex parts (character details, list of objects, world laws dictionary) into JSON strings before sending them to the FastMCP server.
FastMCP's auto-deserialization feature (for `list`, `dict`, and Pydantic `BaseModel` arguments) bridges this gap:
- The client sends complex data as JSON strings (as required by the MCP spec for prompt arguments: `dict[str, str]`).
- The server-side `FastMCP` prompt automatically attempts to `json.loads()` these strings into the Python types hinted in the prompt function signature.
This keeps the server-side prompt function clean, Pythonic, and focused on its core task of generating the story prompt, without boilerplate `json.loads()` calls.
## Files
* `story_server.py`: The FastMCP server code defining the `Character` model and the `generate_dynamic_story_prompt`.
* `story_client.py`: A conceptual Python client script showing how to call this prompt, including serializing complex arguments to JSON strings.
## How to Run (Conceptual)
1. Run the `story_server.py`.
2. In a separate terminal, run the `story_client.py`.
The client will output the creative story prompt generated by the server.

View file

@ -0,0 +1,96 @@
import asyncio
import json
from fastmcp import Client
SERVER_SCRIPT_PATH = __file__.replace("client", "server")
async def main():
print(f"Attempting to connect to server script: {SERVER_SCRIPT_PATH}\n")
client = Client(SERVER_SCRIPT_PATH)
async with client:
print("Successfully connected to server!")
# 1. Define the complex data for the prompt
character_data = {
"name": "Elara",
"archetype": "Reluctant Oracle",
"quirky_trait": "habit of humming ancient, forgotten tunes when nervous",
}
objects_data = [
"a tarnished silver locket that refuses to open",
"a smooth, obsidian sphere that whispers secrets in the dark",
"a single, petrified rose that blooms only in moonlight",
]
laws_data = {
"time": "flows like molasses uphill on Tuesdays",
"shadows": "have a mind of their own and occasionally steal small, shiny objects",
"laughter": "can briefly mend broken things",
}
# 2. Prepare arguments for the client, serializing complex types to JSON strings
prompt_args = {
# 'character_details' expects a Character Pydantic model
"character_details": json.dumps(character_data),
# 'mysterious_objects' expects a list[str]
"mysterious_objects": json.dumps(objects_data),
# 'world_laws' expects a dict[str, str]
"world_laws": json.dumps(laws_data),
}
print("--- Sending to server: ---")
for key, value in prompt_args.items():
print(f" {key}: {value}")
print("--------------------------\n")
try:
# 3. Call the prompt
results_iterable = await client.get_prompt(
"generate_dynamic_story_prompt", arguments=prompt_args
)
print("--- Generated Story Prompt from Server: ---")
# The client.get_prompt() seems to return an iterable of (key, value) pairs
# from the GetPromptResult model. We need to find the 'messages' key.
prompt_messages_list = None
if results_iterable:
for key, value in results_iterable:
if key == "messages":
prompt_messages_list = value
break # Found the messages list
if prompt_messages_list:
for message in (
prompt_messages_list
): # This should be a list of PromptMessage objects
if (
hasattr(message, "content")
and hasattr(message.content, "text")
and message.content.text is not None
):
print(message.content.text)
else:
print(
f"(Received message with unexpected content structure: {message!r})"
)
else:
print(
"(Could not find 'messages' in the prompt result or result was empty)"
)
print("-----------------------------------------")
except Exception as e:
print(f"Error calling prompt: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(main())

View file

@ -0,0 +1,69 @@
from pydantic import BaseModel, Field
from fastmcp import FastMCP
mcp = FastMCP(name="DynamicStoryServer")
class Character(BaseModel):
name: str = Field(..., description="The character's name.")
archetype: str = Field(
default="Mysterious Stranger",
description="The character's archetype (e.g., Brave Knight, Wily Rogue).",
)
quirky_trait: str = Field(
..., description="A unique or unusual trait of the character."
)
@mcp.prompt()
def generate_dynamic_story_prompt(
character_details: Character,
mysterious_objects: list[str],
world_laws: dict[str, str],
) -> str:
"""Generates a creative story prompt based on character, objects, and world laws."""
# Directly use the deserialized Python objects:
prompt = f"Our protagonist, {character_details.name}, a self-proclaimed '{character_details.archetype}', "
prompt += f"known for their {character_details.quirky_trait}, stumbles upon a peculiar collection: "
if mysterious_objects:
if len(mysterious_objects) == 1:
prompt += f"a single {mysterious_objects[0]}. "
else:
objects_str = (
", ".join(mysterious_objects[:-1]) + f", and a {mysterious_objects[-1]}"
)
prompt += f"{objects_str}. "
else:
prompt += "nothing but dust bunnies. "
prompt += (
"\n\nSuddenly, the world shifts. The very laws of reality seem to be in flux:"
)
if world_laws:
for law, description in world_laws.items():
prompt += f"\n- {law.capitalize()}: {description}."
else:
prompt += "\n- Everything is disconcertingly normal."
prompt += "\n\nWhat happens next?"
return prompt
if __name__ == "__main__":
print("Starting Dynamic Story Server...")
# To run this server, you would typically call mcp.run()
# For example, if you have uvicorn and want to run it as an ASGI app (if FastMCP supports it directly or via an adapter):
# import uvicorn
# uvicorn.run(mcp.asgi_app, host="0.0.0.0", port=8000)
# Or, if it runs via its own stdio mechanism, just mcp.run()
try:
mcp.run() # Assuming this is the standard way to run a FastMCP stdio server
except KeyboardInterrupt:
print("Server shutting down.")
except Exception as e:
print(f"Server failed to start or run: {e}")

View file

@ -6,4 +6,7 @@ test: build
# Run pyright on all files
typecheck:
uv run --frozen pyright
uv run --frozen pyright
docs:
cd docs && uv run mintlify dev

View file

@ -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,43 @@ class Prompt(BaseModel):
raise ValueError(f"Missing required arguments: {missing}")
try:
# Prepare arguments with context
# Prepare arguments
kwargs = arguments.copy() if arguments else {}
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]
origin_type = get_origin(target_type_hint)
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
):
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."
)
# 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()

View file

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