mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 21:44:18 +02:00
Merge pull request #908 from jlowin/feature/server-side-type-conversion
This commit is contained in:
Jeremiah Lowin
GitHub
commit
951de7634a
4 changed files with 466 additions and 37 deletions
|
|
@ -57,6 +57,84 @@ def generate_code_request(language: str, task_description: str) -> PromptMessage
|
|||
Functions with `*args` or `**kwargs` are not supported as prompts. This restriction exists because FastMCP needs to generate a complete parameter schema for the MCP protocol, which isn't possible with variable argument lists.
|
||||
</Tip>
|
||||
|
||||
### Argument Types
|
||||
|
||||
<VersionBadge version="2.9.0" />
|
||||
|
||||
The MCP specification requires that all prompt arguments be passed as strings, but FastMCP allows you to use typed annotations for better developer experience. When you use complex types like `list[int]` or `dict[str, str]`, FastMCP:
|
||||
|
||||
1. **Automatically converts** string arguments from MCP clients to the expected types
|
||||
2. **Generates helpful descriptions** showing the exact JSON string format needed
|
||||
3. **Preserves direct usage** - you can still call prompts with properly typed arguments
|
||||
|
||||
Since the MCP specification only allows string arguments, clients need to know what string format to use for complex types. FastMCP solves this by automatically enhancing the argument descriptions with JSON schema information, making it clear to both humans and LLMs how to format their arguments.
|
||||
|
||||
<CodeGroup>
|
||||
|
||||
```python Python Code
|
||||
@mcp.prompt
|
||||
def analyze_data(
|
||||
numbers: list[int],
|
||||
metadata: dict[str, str],
|
||||
threshold: float
|
||||
) -> str:
|
||||
"""Analyze numerical data."""
|
||||
avg = sum(numbers) / len(numbers)
|
||||
return f"Average: {avg}, above threshold: {avg > threshold}"
|
||||
```
|
||||
|
||||
```json Resulting MCP Prompt
|
||||
{
|
||||
"name": "analyze_data",
|
||||
"description": "Analyze numerical data.",
|
||||
"arguments": [
|
||||
{
|
||||
"name": "numbers",
|
||||
"description": "Provide as a JSON string matching the following schema: {\"items\":{\"type\":\"integer\"},\"type\":\"array\"}",
|
||||
"required": true
|
||||
},
|
||||
{
|
||||
"name": "metadata",
|
||||
"description": "Provide as a JSON string matching the following schema: {\"additionalProperties\":{\"type\":\"string\"},\"type\":\"object\"}",
|
||||
"required": true
|
||||
},
|
||||
{
|
||||
"name": "threshold",
|
||||
"description": "Provide as a JSON string matching the following schema: {\"type\":\"number\"}",
|
||||
"required": true
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
</CodeGroup>
|
||||
|
||||
**MCP clients will call this prompt with string arguments:**
|
||||
```json
|
||||
{
|
||||
"numbers": "[1, 2, 3, 4, 5]",
|
||||
"metadata": "{\"source\": \"api\", \"version\": \"1.0\"}",
|
||||
"threshold": "2.5"
|
||||
}
|
||||
```
|
||||
|
||||
**But you can still call it directly with proper types:**
|
||||
```python
|
||||
# This also works for direct calls
|
||||
result = await prompt.render({
|
||||
"numbers": [1, 2, 3, 4, 5],
|
||||
"metadata": {"source": "api", "version": "1.0"},
|
||||
"threshold": 2.5
|
||||
})
|
||||
```
|
||||
|
||||
<Warning>
|
||||
Keep your type annotations simple when using this feature. Complex nested types or custom classes may not convert reliably from JSON strings. The automatically generated schema descriptions are the only guidance users receive about the expected format.
|
||||
|
||||
Good choices: `list[int]`, `dict[str, str]`, `float`, `bool`
|
||||
Avoid: Complex Pydantic models, deeply nested structures, custom classes
|
||||
</Warning>
|
||||
|
||||
### Return Values
|
||||
|
||||
FastMCP intelligently handles different return types from your prompt function:
|
||||
|
|
@ -78,33 +156,6 @@ def roleplay_scenario(character: str, situation: str) -> list[Message]:
|
|||
]
|
||||
```
|
||||
|
||||
### Type Annotations
|
||||
|
||||
Type annotations are important for prompts. They:
|
||||
1. Inform FastMCP about the expected types for each parameter.
|
||||
2. Allow validation of parameters received from clients.
|
||||
3. Are used to generate the prompt's schema for the MCP protocol.
|
||||
|
||||
```python
|
||||
from pydantic import Field
|
||||
from typing import Literal, Optional
|
||||
|
||||
@mcp.prompt
|
||||
def generate_content_request(
|
||||
topic: str = Field(description="The main subject to cover"),
|
||||
format: Literal["blog", "email", "social"] = "blog",
|
||||
tone: str = "professional",
|
||||
word_count: Optional[int] = None
|
||||
) -> str:
|
||||
"""Create a request for generating content in a specific format."""
|
||||
prompt = f"Please write a {format} post about {topic} in a {tone} tone."
|
||||
|
||||
if word_count:
|
||||
prompt += f" It should be approximately {word_count} words long."
|
||||
|
||||
return prompt
|
||||
```
|
||||
|
||||
|
||||
### Required vs. Optional Parameters
|
||||
|
||||
|
|
|
|||
|
|
@ -3,15 +3,16 @@
|
|||
from __future__ import annotations as _annotations
|
||||
|
||||
import inspect
|
||||
import json
|
||||
from abc import ABC, abstractmethod
|
||||
from collections.abc import Awaitable, Callable, Sequence
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from typing import Any
|
||||
|
||||
import pydantic_core
|
||||
from mcp.types import Prompt as MCPPrompt
|
||||
from mcp.types import PromptArgument as MCPPromptArgument
|
||||
from mcp.types import PromptMessage, Role, TextContent
|
||||
from pydantic import Field, TypeAdapter, validate_call
|
||||
from pydantic import Field, TypeAdapter
|
||||
|
||||
from fastmcp.exceptions import PromptError
|
||||
from fastmcp.server.dependencies import get_context
|
||||
|
|
@ -25,10 +26,6 @@ from fastmcp.utilities.types import (
|
|||
get_cached_typeadapter,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
pass
|
||||
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
|
|
@ -181,17 +178,43 @@ class FunctionPrompt(Prompt):
|
|||
arguments: list[PromptArgument] = []
|
||||
if "properties" in parameters:
|
||||
for param_name, param in parameters["properties"].items():
|
||||
arg_description = param.get("description")
|
||||
|
||||
# For non-string parameters, append JSON schema info to help users
|
||||
# understand the expected format when passing as strings (MCP requirement)
|
||||
if param_name in sig.parameters:
|
||||
sig_param = sig.parameters[param_name]
|
||||
if (
|
||||
sig_param.annotation != inspect.Parameter.empty
|
||||
and sig_param.annotation is not str
|
||||
and param_name != context_kwarg
|
||||
):
|
||||
# Get the JSON schema for this specific parameter type
|
||||
try:
|
||||
param_adapter = get_cached_typeadapter(sig_param.annotation)
|
||||
param_schema = param_adapter.json_schema()
|
||||
|
||||
# Create compact schema representation
|
||||
schema_str = json.dumps(param_schema, separators=(",", ":"))
|
||||
|
||||
# Append schema info to description
|
||||
schema_note = f"Provide as a JSON string matching the following schema: {schema_str}"
|
||||
if arg_description:
|
||||
arg_description = f"{arg_description}\n\n{schema_note}"
|
||||
else:
|
||||
arg_description = schema_note
|
||||
except Exception:
|
||||
# If schema generation fails, skip enhancement
|
||||
pass
|
||||
|
||||
arguments.append(
|
||||
PromptArgument(
|
||||
name=param_name,
|
||||
description=param.get("description"),
|
||||
description=arg_description,
|
||||
required=param_name in parameters.get("required", []),
|
||||
)
|
||||
)
|
||||
|
||||
# ensure the arguments are properly cast
|
||||
fn = validate_call(fn)
|
||||
|
||||
return cls(
|
||||
name=func_name,
|
||||
description=description,
|
||||
|
|
@ -201,6 +224,60 @@ class FunctionPrompt(Prompt):
|
|||
fn=fn,
|
||||
)
|
||||
|
||||
def _convert_string_arguments(self, kwargs: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Convert string arguments to expected types based on function signature."""
|
||||
from fastmcp.server.context import Context
|
||||
|
||||
sig = inspect.signature(self.fn)
|
||||
converted_kwargs = {}
|
||||
|
||||
# Find context parameter name if any
|
||||
context_param_name = find_kwarg_by_type(self.fn, kwarg_type=Context)
|
||||
|
||||
for param_name, param_value in kwargs.items():
|
||||
if param_name in sig.parameters:
|
||||
param = sig.parameters[param_name]
|
||||
|
||||
# Skip Context parameters - they're handled separately
|
||||
if param_name == context_param_name:
|
||||
converted_kwargs[param_name] = param_value
|
||||
continue
|
||||
|
||||
# If parameter has no annotation or annotation is str, pass as-is
|
||||
if (
|
||||
param.annotation == inspect.Parameter.empty
|
||||
or param.annotation is str
|
||||
):
|
||||
converted_kwargs[param_name] = param_value
|
||||
# If argument is not a string, pass as-is (already properly typed)
|
||||
elif not isinstance(param_value, str):
|
||||
converted_kwargs[param_name] = param_value
|
||||
else:
|
||||
# Try to convert string argument using type adapter
|
||||
try:
|
||||
adapter = get_cached_typeadapter(param.annotation)
|
||||
# Try JSON parsing first for complex types
|
||||
try:
|
||||
converted_kwargs[param_name] = adapter.validate_json(
|
||||
param_value
|
||||
)
|
||||
except (ValueError, TypeError, pydantic_core.ValidationError):
|
||||
# Fallback to direct validation
|
||||
converted_kwargs[param_name] = adapter.validate_python(
|
||||
param_value
|
||||
)
|
||||
except (ValueError, TypeError, pydantic_core.ValidationError) as e:
|
||||
# If conversion fails, provide informative error
|
||||
raise PromptError(
|
||||
f"Could not convert argument '{param_name}' with value '{param_value}' "
|
||||
f"to expected type {param.annotation}. Error: {e}"
|
||||
)
|
||||
else:
|
||||
# Parameter not in function signature, pass as-is
|
||||
converted_kwargs[param_name] = param_value
|
||||
|
||||
return converted_kwargs
|
||||
|
||||
async def render(
|
||||
self,
|
||||
arguments: dict[str, Any] | None = None,
|
||||
|
|
@ -223,6 +300,9 @@ class FunctionPrompt(Prompt):
|
|||
if context_kwarg and context_kwarg not in kwargs:
|
||||
kwargs[context_kwarg] = get_context()
|
||||
|
||||
# Convert string arguments to expected types when needed
|
||||
kwargs = self._convert_string_arguments(kwargs)
|
||||
|
||||
# Call function and check if result is a coroutine
|
||||
result = self.fn(**kwargs)
|
||||
if inspect.iscoroutine(result):
|
||||
|
|
|
|||
|
|
@ -240,3 +240,245 @@ class TestRenderPrompt:
|
|||
),
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
class TestPromptTypeConversion:
|
||||
async def test_list_of_integers_as_string_args(self):
|
||||
"""Test that prompts can handle complex types passed as strings from MCP spec."""
|
||||
|
||||
def sum_numbers(numbers: list[int]) -> str:
|
||||
"""Calculate the sum of a list of numbers."""
|
||||
total = sum(numbers)
|
||||
return f"The sum is: {total}"
|
||||
|
||||
prompt = Prompt.from_function(sum_numbers)
|
||||
|
||||
# MCP spec only allows string arguments, so this should work
|
||||
# after we implement type conversion
|
||||
result_from_string = await prompt.render(
|
||||
arguments={"numbers": "[1, 2, 3, 4, 5]"}
|
||||
)
|
||||
assert result_from_string == [
|
||||
PromptMessage(
|
||||
role="user", content=TextContent(type="text", text="The sum is: 15")
|
||||
)
|
||||
]
|
||||
|
||||
# Both should work now with string conversion
|
||||
result_from_list_string = await prompt.render(
|
||||
arguments={"numbers": "[1, 2, 3, 4, 5]"}
|
||||
)
|
||||
assert result_from_list_string == result_from_string
|
||||
|
||||
async def test_various_type_conversions(self):
|
||||
"""Test type conversion for various data types."""
|
||||
|
||||
def process_data(
|
||||
name: str,
|
||||
age: int,
|
||||
scores: list[float],
|
||||
metadata: dict[str, str],
|
||||
active: bool,
|
||||
) -> str:
|
||||
return f"{name} ({age}): {len(scores)} scores, active={active}, metadata keys={list(metadata.keys())}"
|
||||
|
||||
prompt = Prompt.from_function(process_data)
|
||||
|
||||
# All arguments as strings (as MCP would send them)
|
||||
result = await prompt.render(
|
||||
arguments={
|
||||
"name": "Alice",
|
||||
"age": "25",
|
||||
"scores": "[1.5, 2.0, 3.5]",
|
||||
"metadata": '{"project": "test", "version": "1.0"}',
|
||||
"active": "true",
|
||||
}
|
||||
)
|
||||
|
||||
expected_text = (
|
||||
"Alice (25): 3 scores, active=True, metadata keys=['project', 'version']"
|
||||
)
|
||||
assert result == [
|
||||
PromptMessage(
|
||||
role="user", content=TextContent(type="text", text=expected_text)
|
||||
)
|
||||
]
|
||||
|
||||
async def test_type_conversion_error_handling(self):
|
||||
"""Test that informative errors are raised for invalid type conversions."""
|
||||
from fastmcp.exceptions import PromptError
|
||||
|
||||
def typed_prompt(numbers: list[int]) -> str:
|
||||
return f"Got {len(numbers)} numbers"
|
||||
|
||||
prompt = Prompt.from_function(typed_prompt)
|
||||
|
||||
# Test with invalid JSON - should raise PromptError due to exception handling in render()
|
||||
with pytest.raises(PromptError) as exc_info:
|
||||
await prompt.render(arguments={"numbers": "not valid json"})
|
||||
|
||||
assert f"Error rendering prompt {prompt.name}" in str(exc_info.value)
|
||||
|
||||
async def test_json_parsing_fallback(self):
|
||||
"""Test that JSON parsing falls back to direct validation when needed."""
|
||||
|
||||
def data_prompt(value: int) -> str:
|
||||
return f"Value: {value}"
|
||||
|
||||
prompt = Prompt.from_function(data_prompt)
|
||||
|
||||
# This should work with JSON parsing (integer as string)
|
||||
result1 = await prompt.render(arguments={"value": "42"})
|
||||
assert result1 == [
|
||||
PromptMessage(
|
||||
role="user", content=TextContent(type="text", text="Value: 42")
|
||||
)
|
||||
]
|
||||
|
||||
# This should work with direct validation (already an integer string)
|
||||
result2 = await prompt.render(arguments={"value": "123"})
|
||||
assert result2 == [
|
||||
PromptMessage(
|
||||
role="user", content=TextContent(type="text", text="Value: 123")
|
||||
)
|
||||
]
|
||||
|
||||
async def test_mixed_string_and_typed_args(self):
|
||||
"""Test mixing string args (no conversion) with typed args (conversion needed)."""
|
||||
|
||||
def mixed_prompt(message: str, count: int) -> str:
|
||||
return f"{message} (repeated {count} times)"
|
||||
|
||||
prompt = Prompt.from_function(mixed_prompt)
|
||||
|
||||
result = await prompt.render(
|
||||
arguments={
|
||||
"message": "Hello world", # str - no conversion needed
|
||||
"count": "3", # int - conversion needed
|
||||
}
|
||||
)
|
||||
|
||||
assert result == [
|
||||
PromptMessage(
|
||||
role="user",
|
||||
content=TextContent(type="text", text="Hello world (repeated 3 times)"),
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
class TestPromptArgumentDescriptions:
|
||||
def test_enhanced_descriptions_for_non_string_types(self):
|
||||
"""Test that non-string argument types get enhanced descriptions with JSON schema."""
|
||||
|
||||
def analyze_data(
|
||||
name: str,
|
||||
numbers: list[int],
|
||||
metadata: dict[str, str],
|
||||
threshold: float,
|
||||
active: bool,
|
||||
) -> str:
|
||||
"""Analyze numerical data."""
|
||||
return f"Analyzed {name}"
|
||||
|
||||
prompt = Prompt.from_function(analyze_data)
|
||||
|
||||
assert prompt.arguments is not None
|
||||
# Check that string parameter has no schema enhancement
|
||||
name_arg = next((arg for arg in prompt.arguments if arg.name == "name"), None)
|
||||
assert name_arg is not None
|
||||
assert name_arg.description is None # No enhancement for string types
|
||||
|
||||
# Check that non-string parameters have schema enhancements
|
||||
numbers_arg = next(
|
||||
(arg for arg in prompt.arguments if arg.name == "numbers"), None
|
||||
)
|
||||
assert numbers_arg is not None
|
||||
assert numbers_arg.description is not None
|
||||
assert (
|
||||
"Provide as a JSON string matching the following schema:"
|
||||
in numbers_arg.description
|
||||
)
|
||||
assert '{"items":{"type":"integer"},"type":"array"}' in numbers_arg.description
|
||||
|
||||
metadata_arg = next(
|
||||
(arg for arg in prompt.arguments if arg.name == "metadata"), None
|
||||
)
|
||||
assert metadata_arg is not None
|
||||
assert metadata_arg.description is not None
|
||||
assert (
|
||||
"Provide as a JSON string matching the following schema:"
|
||||
in metadata_arg.description
|
||||
)
|
||||
assert (
|
||||
'{"additionalProperties":{"type":"string"},"type":"object"}'
|
||||
in metadata_arg.description
|
||||
)
|
||||
|
||||
threshold_arg = next(
|
||||
(arg for arg in prompt.arguments if arg.name == "threshold"), None
|
||||
)
|
||||
assert threshold_arg is not None
|
||||
assert threshold_arg.description is not None
|
||||
assert (
|
||||
"Provide as a JSON string matching the following schema:"
|
||||
in threshold_arg.description
|
||||
)
|
||||
assert '{"type":"number"}' in threshold_arg.description
|
||||
|
||||
active_arg = next(
|
||||
(arg for arg in prompt.arguments if arg.name == "active"), None
|
||||
)
|
||||
assert active_arg is not None
|
||||
assert active_arg.description is not None
|
||||
assert (
|
||||
"Provide as a JSON string matching the following schema:"
|
||||
in active_arg.description
|
||||
)
|
||||
assert '{"type":"boolean"}' in active_arg.description
|
||||
|
||||
def test_enhanced_descriptions_with_existing_descriptions(self):
|
||||
"""Test that existing parameter descriptions are preserved with schema appended."""
|
||||
from typing import Annotated
|
||||
|
||||
from pydantic import Field
|
||||
|
||||
def documented_prompt(
|
||||
numbers: Annotated[
|
||||
list[int], Field(description="A list of integers to process")
|
||||
],
|
||||
) -> str:
|
||||
"""Process numbers."""
|
||||
return "processed"
|
||||
|
||||
prompt = Prompt.from_function(documented_prompt)
|
||||
|
||||
assert prompt.arguments is not None
|
||||
numbers_arg = next(
|
||||
(arg for arg in prompt.arguments if arg.name == "numbers"), None
|
||||
)
|
||||
assert numbers_arg is not None
|
||||
# Should have both the original description and the schema
|
||||
assert numbers_arg.description is not None
|
||||
assert "A list of integers to process" in numbers_arg.description
|
||||
assert "\n\n" in numbers_arg.description # Should have newline separator
|
||||
assert (
|
||||
"Provide as a JSON string matching the following schema:"
|
||||
in numbers_arg.description
|
||||
)
|
||||
|
||||
def test_string_parameters_no_enhancement(self):
|
||||
"""Test that string parameters don't get schema enhancement."""
|
||||
|
||||
def string_only_prompt(message: str, name: str) -> str:
|
||||
return f"{message}, {name}"
|
||||
|
||||
prompt = Prompt.from_function(string_only_prompt)
|
||||
|
||||
assert prompt.arguments is not None
|
||||
for arg in prompt.arguments:
|
||||
# String parameters should not have schema enhancement
|
||||
if arg.description is not None:
|
||||
assert (
|
||||
"Provide as a JSON string matching the following schema:"
|
||||
not in arg.description
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1785,6 +1785,62 @@ class TestPrompts:
|
|||
assert prompts[0].arguments[1].name == "optional"
|
||||
assert prompts[0].arguments[1].required is False
|
||||
|
||||
async def test_list_prompts_with_enhanced_descriptions(self):
|
||||
"""Test that enhanced descriptions with JSON schema are visible via MCP protocol."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.prompt
|
||||
def analyze_data(
|
||||
name: str, numbers: list[int], metadata: dict[str, str], threshold: float
|
||||
) -> str:
|
||||
"""Analyze some data."""
|
||||
return f"Analyzed {name}"
|
||||
|
||||
async with Client(mcp) as client:
|
||||
prompts = await client.list_prompts()
|
||||
assert len(prompts) == 1
|
||||
prompt = prompts[0]
|
||||
assert prompt.name == "analyze_data"
|
||||
assert prompt.description == "Analyze some data."
|
||||
|
||||
# Find each argument and verify schema enhancements
|
||||
assert prompt.arguments is not None
|
||||
args_by_name = {arg.name: arg for arg in prompt.arguments}
|
||||
|
||||
# String parameter should not have schema enhancement
|
||||
name_arg = args_by_name["name"]
|
||||
assert name_arg.description is None
|
||||
|
||||
# Non-string parameters should have schema enhancements
|
||||
numbers_arg = args_by_name["numbers"]
|
||||
assert numbers_arg.description is not None
|
||||
assert (
|
||||
"Provide as a JSON string matching the following schema:"
|
||||
in numbers_arg.description
|
||||
)
|
||||
assert (
|
||||
'{"items":{"type":"integer"},"type":"array"}' in numbers_arg.description
|
||||
)
|
||||
|
||||
metadata_arg = args_by_name["metadata"]
|
||||
assert metadata_arg.description is not None
|
||||
assert (
|
||||
"Provide as a JSON string matching the following schema:"
|
||||
in metadata_arg.description
|
||||
)
|
||||
assert (
|
||||
'{"additionalProperties":{"type":"string"},"type":"object"}'
|
||||
in metadata_arg.description
|
||||
)
|
||||
|
||||
threshold_arg = args_by_name["threshold"]
|
||||
assert threshold_arg.description is not None
|
||||
assert (
|
||||
"Provide as a JSON string matching the following schema:"
|
||||
in threshold_arg.description
|
||||
)
|
||||
assert '{"type":"number"}' in threshold_arg.description
|
||||
|
||||
async def test_get_prompt(self):
|
||||
"""Test getting a prompt through MCP protocol."""
|
||||
mcp = FastMCP()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue