fastmcp/examples/dynamic_story_prompt/story_client.py
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

96 lines
3.4 KiB
Python

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())