mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Compare commits
4 commits
main
...
fix-remove
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3acdbf1376 | ||
|
|
86f38b9acc | ||
|
|
5edcee4513 | ||
|
|
9c14f0f73c |
4 changed files with 562 additions and 62 deletions
|
|
@ -198,6 +198,79 @@ Without `expose_headers=["mcp-session-id"]`, browsers will receive the session I
|
|||
**Production Security**: Never use `allow_origins=["*"]` in production. Specify the exact origins of your browser-based clients. Using wildcards exposes your server to unauthorized access from any website.
|
||||
</Warning>
|
||||
|
||||
### SSE Polling for Long-Running Operations
|
||||
|
||||
<VersionBadge version="2.14.0" />
|
||||
|
||||
<Note>
|
||||
This feature only applies to the **StreamableHTTP transport** (the default for `http_app()`). It does not apply to the legacy SSE transport (`transport="sse"`).
|
||||
</Note>
|
||||
|
||||
When running tools that take a long time to complete, you may encounter issues with load balancers or proxies terminating connections that stay idle too long. [SEP-1699](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699) introduces SSE polling to solve this by allowing the server to gracefully close connections and have clients automatically reconnect.
|
||||
|
||||
To enable SSE polling, configure an `EventStore` when creating your HTTP application:
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP, Context
|
||||
from fastmcp.server.event_store import EventStore
|
||||
|
||||
mcp = FastMCP("My Server")
|
||||
|
||||
@mcp.tool
|
||||
async def long_running_task(ctx: Context) -> str:
|
||||
"""A task that takes several minutes to complete."""
|
||||
for i in range(100):
|
||||
await ctx.report_progress(i, 100)
|
||||
|
||||
# Periodically close the connection to avoid load balancer timeouts
|
||||
# Client will automatically reconnect and resume receiving progress
|
||||
if i % 30 == 0 and i > 0:
|
||||
await ctx.close_sse_stream()
|
||||
|
||||
await do_expensive_work()
|
||||
|
||||
return "Done!"
|
||||
|
||||
# Configure with EventStore for resumability
|
||||
event_store = EventStore()
|
||||
app = mcp.http_app(
|
||||
event_store=event_store,
|
||||
retry_interval=2000, # Client reconnects after 2 seconds
|
||||
)
|
||||
```
|
||||
|
||||
**How it works:**
|
||||
|
||||
1. When `event_store` is configured, the server stores all events (progress updates, results) with unique IDs
|
||||
2. Calling `ctx.close_sse_stream()` gracefully closes the HTTP connection
|
||||
3. The client automatically reconnects with a `Last-Event-ID` header
|
||||
4. The server replays any events the client missed during the disconnection
|
||||
|
||||
The `retry_interval` parameter (in milliseconds) controls how long clients wait before reconnecting. Choose a value that balances responsiveness with server load.
|
||||
|
||||
<Note>
|
||||
`close_sse_stream()` is a no-op if called without an `EventStore` configured, so you can safely include it in tools that may run in different deployment configurations.
|
||||
</Note>
|
||||
|
||||
#### Custom Storage Backends
|
||||
|
||||
By default, `EventStore` uses in-memory storage. For production deployments with multiple server instances, you can provide a custom storage backend using the `key_value` package:
|
||||
|
||||
```python
|
||||
from fastmcp.server.event_store import EventStore
|
||||
from key_value.aio.stores.redis import RedisStore
|
||||
|
||||
# Use Redis for distributed deployments
|
||||
redis_store = RedisStore(url="redis://localhost:6379")
|
||||
event_store = EventStore(
|
||||
storage=redis_store,
|
||||
max_events_per_stream=100, # Keep last 100 events per stream
|
||||
ttl=3600, # Events expire after 1 hour
|
||||
)
|
||||
|
||||
app = mcp.http_app(event_store=event_store)
|
||||
```
|
||||
|
||||
## Integration with Web Frameworks
|
||||
|
||||
If you already have a web application running, you can add MCP capabilities by mounting a FastMCP server as a sub-application. This allows you to expose MCP tools alongside your existing API endpoints, sharing the same domain and infrastructure. The MCP server becomes just another route in your application, making it easy to manage and deploy.
|
||||
|
|
@ -513,11 +586,66 @@ When deploying to production, you'll want to optimize your server for performanc
|
|||
# Run with basic configuration
|
||||
uvicorn app:app --host 0.0.0.0 --port 8000
|
||||
|
||||
# Ensure stateless HTTP mode is enabled (stateless_http=True)
|
||||
# Run with multiple workers for production
|
||||
# Run with multiple workers for production (requires stateless mode - see below)
|
||||
uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4
|
||||
```
|
||||
|
||||
### Horizontal Scaling
|
||||
|
||||
<VersionBadge version="2.10.2" />
|
||||
|
||||
When deploying FastMCP behind a load balancer or running multiple server instances, you need to understand how the HTTP transport handles sessions and configure your server appropriately.
|
||||
|
||||
#### Understanding Sessions
|
||||
|
||||
By default, FastMCP's Streamable HTTP transport maintains server-side sessions. Sessions enable stateful MCP features like [elicitation](/servers/elicitation) and [sampling](/servers/sampling), where the server needs to maintain context across multiple requests from the same client.
|
||||
|
||||
This works perfectly for single-instance deployments. However, sessions are stored in memory on each server instance, which creates challenges when scaling horizontally.
|
||||
|
||||
#### Without Stateless Mode
|
||||
|
||||
When running multiple server instances behind a load balancer (Traefik, nginx, HAProxy, Kubernetes, etc.), requests from the same client may be routed to different instances:
|
||||
|
||||
1. Client connects to Instance A → session created on Instance A
|
||||
2. Next request routes to Instance B → session doesn't exist → **request fails**
|
||||
|
||||
You might expect sticky sessions (session affinity) to solve this, but they don't work reliably with MCP clients.
|
||||
|
||||
<Warning>
|
||||
**Why sticky sessions don't work:** Most MCP clients—including Cursor and Claude Code—use `fetch()` internally and don't properly forward `Set-Cookie` headers. Without cookies, load balancers can't identify which instance should handle subsequent requests. This is a limitation in how these clients implement HTTP, not something you can fix with load balancer configuration.
|
||||
</Warning>
|
||||
|
||||
#### Enabling Stateless Mode
|
||||
|
||||
For horizontally scaled deployments, enable stateless HTTP mode. In stateless mode, each request creates a fresh transport context, eliminating the need for session affinity entirely.
|
||||
|
||||
**Option 1: Via constructor**
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
|
||||
mcp = FastMCP("My Server", stateless_http=True)
|
||||
|
||||
@mcp.tool
|
||||
def process(data: str) -> str:
|
||||
return f"Processed: {data}"
|
||||
|
||||
app = mcp.http_app()
|
||||
```
|
||||
|
||||
**Option 2: Via `run()`**
|
||||
|
||||
```python
|
||||
if __name__ == "__main__":
|
||||
mcp.run(transport="http", stateless_http=True)
|
||||
```
|
||||
|
||||
**Option 3: Via environment variable**
|
||||
|
||||
```bash
|
||||
FASTMCP_STATELESS_HTTP=true uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4
|
||||
```
|
||||
|
||||
### Environment Variables
|
||||
|
||||
Production deployments should never hardcode sensitive information like API keys or authentication tokens. Instead, use environment variables to configure your server at runtime. This keeps your code secure and makes it easy to deploy the same code to different environments with different configurations.
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ title: User Elicitation
|
|||
sidebarTitle: Elicitation
|
||||
description: Request structured input from users during tool execution through the MCP context.
|
||||
icon: message-question
|
||||
tag: NEW
|
||||
---
|
||||
|
||||
import { VersionBadge } from '/snippets/version-badge.mdx'
|
||||
|
|
@ -209,10 +210,10 @@ Often you'll want to constrain the user's response to a specific set of values.
|
|||
async def set_priority(ctx: Context) -> str:
|
||||
"""Set task priority level."""
|
||||
result = await ctx.elicit(
|
||||
"What priority level?",
|
||||
"What priority level?",
|
||||
response_type=["low", "medium", "high"],
|
||||
)
|
||||
|
||||
|
||||
if result.action == "accept":
|
||||
return f"Priority set to: {result.data}"
|
||||
```
|
||||
|
|
@ -223,10 +224,10 @@ from typing import Literal
|
|||
async def set_priority(ctx: Context) -> str:
|
||||
"""Set task priority level."""
|
||||
result = await ctx.elicit(
|
||||
"What priority level?",
|
||||
"What priority level?",
|
||||
response_type=Literal["low", "medium", "high"]
|
||||
)
|
||||
|
||||
|
||||
if result.action == "accept":
|
||||
return f"Priority set to: {result.data}"
|
||||
return "No priority set"
|
||||
|
|
@ -237,19 +238,124 @@ from enum import Enum
|
|||
class Priority(Enum):
|
||||
LOW = "low"
|
||||
MEDIUM = "medium"
|
||||
HIGH = "high"
|
||||
HIGH = "high"
|
||||
|
||||
@mcp.tool
|
||||
async def set_priority(ctx: Context) -> str:
|
||||
"""Set task priority level."""
|
||||
result = await ctx.elicit("What priority level?", response_type=Priority)
|
||||
|
||||
|
||||
if result.action == "accept":
|
||||
return f"Priority set to: {result.data.value}"
|
||||
return "No priority set"
|
||||
```
|
||||
</CodeGroup>
|
||||
|
||||
#### Multi-Select
|
||||
|
||||
<VersionBadge version="2.14.0" />
|
||||
|
||||
Enable multi-select by wrapping your choices in an additional list level. This allows users to select multiple values from the available options.
|
||||
|
||||
<CodeGroup>
|
||||
```python {6-8} title="List of a list of strings"
|
||||
@mcp.tool
|
||||
async def select_tags(ctx: Context) -> str:
|
||||
"""Select multiple tags."""
|
||||
result = await ctx.elicit(
|
||||
"Choose tags",
|
||||
response_type=[["bug", "feature", "documentation"]] # Note: list of a list
|
||||
)
|
||||
|
||||
if result.action == "accept":
|
||||
tags = result.data # List of selected strings
|
||||
return f"Selected tags: {', '.join(tags)}"
|
||||
```
|
||||
|
||||
```python {1, 3-6, 11-14} title="list[Enum] type annotation"
|
||||
from enum import Enum
|
||||
|
||||
class Tag(Enum):
|
||||
BUG = "bug"
|
||||
FEATURE = "feature"
|
||||
DOCS = "documentation"
|
||||
|
||||
@mcp.tool
|
||||
async def select_tags(ctx: Context) -> str:
|
||||
result = await ctx.elicit(
|
||||
"Choose tags",
|
||||
response_type=list[Tag] # Type annotation for multi-select
|
||||
)
|
||||
if result.action == "accept":
|
||||
tags = [tag.value for tag in result.data]
|
||||
return f"Selected: {', '.join(tags)}"
|
||||
```
|
||||
</CodeGroup>
|
||||
|
||||
For titled multi-select, wrap a dict in a list (see [Titled Options](#titled-options) for dict syntax):
|
||||
|
||||
```python {6-12}
|
||||
@mcp.tool
|
||||
async def select_priorities(ctx: Context) -> str:
|
||||
"""Select multiple priorities."""
|
||||
result = await ctx.elicit(
|
||||
"Choose priorities",
|
||||
response_type=[{ # Note: list containing a dict
|
||||
"low": {"title": "Low Priority"},
|
||||
"medium": {"title": "Medium Priority"},
|
||||
"high": {"title": "High Priority"}
|
||||
}]
|
||||
)
|
||||
|
||||
if result.action == "accept":
|
||||
priorities = result.data # List of selected strings
|
||||
return f"Selected: {', '.join(priorities)}"
|
||||
```
|
||||
|
||||
#### Titled Options
|
||||
|
||||
<VersionBadge version="2.14.0" />
|
||||
|
||||
For better UI display, you can provide human-readable titles for enum options. FastMCP generates SEP-1330 compliant schemas using the `oneOf` pattern with `const` and `title` fields.
|
||||
|
||||
Use a dict to specify titles for enum values:
|
||||
|
||||
```python {6-10}
|
||||
@mcp.tool
|
||||
async def set_priority(ctx: Context) -> str:
|
||||
"""Set task priority level."""
|
||||
result = await ctx.elicit(
|
||||
"What priority level?",
|
||||
response_type={
|
||||
"low": {"title": "Low Priority"},
|
||||
"medium": {"title": "Medium Priority"},
|
||||
"high": {"title": "High Priority"}
|
||||
}
|
||||
)
|
||||
|
||||
if result.action == "accept":
|
||||
return f"Priority set to: {result.data}"
|
||||
```
|
||||
|
||||
For multi-select with titles, wrap the dict in a list:
|
||||
|
||||
```python {6-12}
|
||||
@mcp.tool
|
||||
async def select_priorities(ctx: Context) -> str:
|
||||
"""Select multiple priorities."""
|
||||
result = await ctx.elicit(
|
||||
"Choose priorities",
|
||||
response_type=[{ # List containing a dict for multi-select
|
||||
"low": {"title": "Low Priority"},
|
||||
"medium": {"title": "Medium Priority"},
|
||||
"high": {"title": "High Priority"}
|
||||
}]
|
||||
)
|
||||
|
||||
if result.action == "accept":
|
||||
priorities = result.data # List of selected strings
|
||||
return f"Selected: {', '.join(priorities)}"
|
||||
```
|
||||
|
||||
### Structured Responses
|
||||
|
||||
|
|
@ -282,6 +388,8 @@ async def create_task(ctx: Context) -> str:
|
|||
|
||||
### Default Values
|
||||
|
||||
<VersionBadge version="2.14.0" />
|
||||
|
||||
You can provide default values for elicitation fields using Pydantic's `Field(default=...)`. Clients will pre-populate form fields with these defaults, making it easier for users to provide input.
|
||||
|
||||
Default values are supported for all primitive types:
|
||||
|
|
|
|||
|
|
@ -38,49 +38,63 @@ class ElicitationJsonSchema(GenerateJsonSchema):
|
|||
"""
|
||||
|
||||
def generate_inner(self, schema: core_schema.CoreSchema) -> JsonSchemaValue:
|
||||
"""Override to prevent ref generation for enums."""
|
||||
"""Override to prevent ref generation for enums and handle list schemas."""
|
||||
# For enum schemas, bypass the ref mechanism entirely
|
||||
if schema["type"] == "enum":
|
||||
# Directly call our custom enum_schema without going through handler
|
||||
# This prevents the ref/defs mechanism from being invoked
|
||||
return self.enum_schema(schema)
|
||||
# For list schemas, check if items are enums
|
||||
if schema["type"] == "list":
|
||||
return self.list_schema(schema)
|
||||
# For all other types, use the default implementation
|
||||
return super().generate_inner(schema)
|
||||
|
||||
def list_schema(self, schema: core_schema.ListSchema) -> JsonSchemaValue:
|
||||
"""Generate schema for list types, detecting enum items for multi-select."""
|
||||
items_schema = schema.get("items_schema")
|
||||
|
||||
# Check if items are enum/Literal
|
||||
if items_schema and items_schema.get("type") == "enum":
|
||||
# Generate array with enum items
|
||||
items = self.enum_schema(items_schema)
|
||||
# If items have oneOf pattern, convert to anyOf for multi-select per SEP-1330
|
||||
if "oneOf" in items:
|
||||
items = {"anyOf": items["oneOf"]}
|
||||
return {
|
||||
"type": "array",
|
||||
"items": items, # Will be {"enum": [...]} or {"anyOf": [...]}
|
||||
}
|
||||
|
||||
# Check if items are Literal (which Pydantic represents differently)
|
||||
if items_schema:
|
||||
# Try to detect Literal patterns
|
||||
items_result = super().generate_inner(items_schema)
|
||||
# If it's a const pattern or enum-like, allow it
|
||||
if (
|
||||
"const" in items_result
|
||||
or "enum" in items_result
|
||||
or "oneOf" in items_result
|
||||
):
|
||||
# Convert oneOf to anyOf for multi-select
|
||||
if "oneOf" in items_result:
|
||||
items_result = {"anyOf": items_result["oneOf"]}
|
||||
return {
|
||||
"type": "array",
|
||||
"items": items_result,
|
||||
}
|
||||
|
||||
# Default behavior for non-enum arrays
|
||||
return super().list_schema(schema)
|
||||
|
||||
def enum_schema(self, schema: core_schema.EnumSchema) -> JsonSchemaValue:
|
||||
"""Generate inline enum schema with optional enumNames for better UI.
|
||||
"""Generate inline enum schema.
|
||||
|
||||
If enum members have a _display_name_ attribute or custom __str__,
|
||||
we'll include enumNames for better UI representation.
|
||||
Always generates enum pattern: {"enum": [value, ...]}
|
||||
Titled enums are handled separately via dict-based syntax in ctx.elicit().
|
||||
"""
|
||||
# Get the base schema from parent
|
||||
result = super().enum_schema(schema)
|
||||
|
||||
# Try to add enumNames if the enum has display-friendly names
|
||||
enum_cls = schema.get("cls")
|
||||
if enum_cls:
|
||||
members = schema.get("members", [])
|
||||
enum_names = []
|
||||
has_custom_names = False
|
||||
|
||||
for member in members:
|
||||
# Check if member has a custom display name attribute
|
||||
if hasattr(member, "_display_name_"):
|
||||
enum_names.append(member._display_name_)
|
||||
has_custom_names = True
|
||||
# Or use the member name with better formatting
|
||||
else:
|
||||
# Convert SNAKE_CASE to Title Case for display
|
||||
display_name = member.name.replace("_", " ").title()
|
||||
enum_names.append(display_name)
|
||||
if display_name != member.value:
|
||||
has_custom_names = True
|
||||
|
||||
# Only add enumNames if they differ from the values
|
||||
if has_custom_names:
|
||||
result["enumNames"] = enum_names
|
||||
|
||||
return result
|
||||
# Get the base schema from parent - always use simple enum pattern
|
||||
return super().enum_schema(schema)
|
||||
|
||||
|
||||
# we can't use the low-level AcceptedElicitation because it only works with BaseModels
|
||||
|
|
@ -96,6 +110,27 @@ class ScalarElicitationType(Generic[T]):
|
|||
value: T
|
||||
|
||||
|
||||
def _dict_to_enum_schema(
|
||||
enum_dict: dict[str, dict[str, str]], multi_select: bool = False
|
||||
) -> dict[str, Any]:
|
||||
"""Convert dict enum to SEP-1330 compliant schema pattern.
|
||||
|
||||
Args:
|
||||
enum_dict: {"low": {"title": "Low Priority"}, "medium": {"title": "Medium Priority"}}
|
||||
multi_select: If True, use anyOf pattern; if False, use oneOf pattern
|
||||
|
||||
Returns:
|
||||
{"oneOf": [{"const": "low", "title": "Low Priority"}, ...]} for single-select
|
||||
{"anyOf": [{"const": "low", "title": "Low Priority"}, ...]} for multi-select
|
||||
"""
|
||||
pattern_key = "anyOf" if multi_select else "oneOf"
|
||||
pattern = []
|
||||
for value, metadata in enum_dict.items():
|
||||
title = metadata.get("title", value)
|
||||
pattern.append({"const": value, "title": title})
|
||||
return {pattern_key: pattern}
|
||||
|
||||
|
||||
def get_elicitation_schema(response_type: type[T]) -> dict[str, Any]:
|
||||
"""Get the schema for an elicitation response.
|
||||
|
||||
|
|
@ -197,20 +232,7 @@ def validate_elicitation_json_schema(schema: dict[str, Any]) -> None:
|
|||
)
|
||||
continue
|
||||
|
||||
# Check if it's a primitive type
|
||||
if prop_type not in ALLOWED_TYPES:
|
||||
raise TypeError(
|
||||
f"Elicitation schema field '{prop_name}' has type '{prop_type}' which is not "
|
||||
f"a primitive type. Only {ALLOWED_TYPES} are allowed in elicitation schemas."
|
||||
)
|
||||
|
||||
# Check for nested objects or arrays of objects (not allowed)
|
||||
if prop_type == "object":
|
||||
raise TypeError(
|
||||
f"Elicitation schema field '{prop_name}' is an object, but nested objects are not allowed. "
|
||||
"Elicitation schemas must be flat objects with primitive properties only."
|
||||
)
|
||||
|
||||
# Check for arrays before checking primitive types
|
||||
if prop_type == "array":
|
||||
items_schema = prop_schema.get("items", {})
|
||||
if items_schema.get("type") == "object":
|
||||
|
|
@ -218,3 +240,35 @@ def validate_elicitation_json_schema(schema: dict[str, Any]) -> None:
|
|||
f"Elicitation schema field '{prop_name}' is an array of objects, but arrays of objects are not allowed. "
|
||||
"Elicitation schemas must be flat objects with primitive properties only."
|
||||
)
|
||||
|
||||
# Allow arrays with enum patterns (for multi-select)
|
||||
if "enum" in items_schema:
|
||||
continue # Allowed: {"type": "array", "items": {"enum": [...]}}
|
||||
|
||||
# Allow arrays with oneOf/anyOf const patterns (SEP-1330)
|
||||
if "oneOf" in items_schema or "anyOf" in items_schema:
|
||||
union_schemas = items_schema.get("oneOf", []) + items_schema.get(
|
||||
"anyOf", []
|
||||
)
|
||||
if union_schemas and all("const" in s for s in union_schemas):
|
||||
continue # Allowed: {"type": "array", "items": {"anyOf": [{"const": ...}, ...]}}
|
||||
|
||||
# Reject other array types (e.g., arrays of primitives without enum pattern)
|
||||
raise TypeError(
|
||||
f"Elicitation schema field '{prop_name}' is an array, but arrays are only allowed "
|
||||
"when items are enums (for multi-select). Only enum arrays are supported in elicitation schemas."
|
||||
)
|
||||
|
||||
# Check for nested objects (not allowed)
|
||||
if prop_type == "object":
|
||||
raise TypeError(
|
||||
f"Elicitation schema field '{prop_name}' is an object, but nested objects are not allowed. "
|
||||
"Elicitation schemas must be flat objects with primitive properties only."
|
||||
)
|
||||
|
||||
# Check if it's a primitive type
|
||||
if prop_type not in ALLOWED_TYPES:
|
||||
raise TypeError(
|
||||
f"Elicitation schema field '{prop_name}' has type '{prop_type}' which is not "
|
||||
f"a primitive type. Only {ALLOWED_TYPES} are allowed in elicitation schemas."
|
||||
)
|
||||
|
|
|
|||
|
|
@ -691,8 +691,8 @@ def test_enum_elicitation_schema_inline():
|
|||
assert schema["properties"]["title"]["type"] == "string"
|
||||
|
||||
|
||||
def test_enum_elicitation_schema_with_enum_names():
|
||||
"""Test that enum schemas can include enumNames for better UI display."""
|
||||
def test_enum_elicitation_schema_inline_untitled():
|
||||
"""Test that enum schemas generate simple enum pattern (no automatic titles)."""
|
||||
|
||||
class TaskStatus(Enum):
|
||||
NOT_STARTED = "not_started"
|
||||
|
|
@ -713,7 +713,10 @@ def test_enum_elicitation_schema_with_enum_names():
|
|||
assert "$ref" not in str(schema)
|
||||
|
||||
status_schema = schema["properties"]["status"]
|
||||
# Should generate simple enum pattern (no automatic title generation)
|
||||
assert "enum" in status_schema
|
||||
assert "oneOf" not in status_schema
|
||||
assert "enumNames" not in status_schema
|
||||
assert status_schema["enum"] == [
|
||||
"not_started",
|
||||
"in_progress",
|
||||
|
|
@ -721,14 +724,221 @@ def test_enum_elicitation_schema_with_enum_names():
|
|||
"on_hold",
|
||||
]
|
||||
|
||||
# Check if enumNames were added for display
|
||||
assert "enumNames" in status_schema
|
||||
assert status_schema["enumNames"] == [
|
||||
"Not Started",
|
||||
"In Progress",
|
||||
"Completed",
|
||||
"On Hold",
|
||||
]
|
||||
|
||||
async def test_dict_based_titled_single_select():
|
||||
"""Test dict-based titled single-select enum."""
|
||||
mcp = FastMCP("TestServer")
|
||||
|
||||
@mcp.tool
|
||||
async def my_tool(ctx: Context) -> str:
|
||||
result = await ctx.elicit(
|
||||
"Choose priority",
|
||||
response_type={
|
||||
"low": {"title": "Low Priority"},
|
||||
"high": {"title": "High Priority"},
|
||||
},
|
||||
)
|
||||
if result.action == "accept":
|
||||
return result.data # type: ignore[attr-defined]
|
||||
return "declined"
|
||||
|
||||
async def elicitation_handler(message, response_type, params, ctx):
|
||||
return ElicitResult(action="accept", content={"value": "low"})
|
||||
|
||||
async with Client(mcp, elicitation_handler=elicitation_handler) as client:
|
||||
result = await client.call_tool("my_tool", {})
|
||||
assert result.data == "low"
|
||||
|
||||
|
||||
async def test_list_list_multi_select_untitled():
|
||||
"""Test list[list[str]] for multi-select untitled shorthand."""
|
||||
mcp = FastMCP("TestServer")
|
||||
|
||||
@mcp.tool
|
||||
async def my_tool(ctx: Context) -> str:
|
||||
result = await ctx.elicit(
|
||||
"Choose tags",
|
||||
response_type=[["bug", "feature", "documentation"]],
|
||||
)
|
||||
if result.action == "accept":
|
||||
return ",".join(result.data) # type: ignore[attr-defined]
|
||||
return "declined"
|
||||
|
||||
async def elicitation_handler(message, response_type, params, ctx):
|
||||
# Verify schema has array with enum pattern
|
||||
schema = params.requestedSchema
|
||||
assert schema["type"] == "object"
|
||||
assert "value" in schema["properties"]
|
||||
value_schema = schema["properties"]["value"]
|
||||
assert value_schema["type"] == "array"
|
||||
assert "enum" in value_schema["items"]
|
||||
assert value_schema["items"]["enum"] == ["bug", "feature", "documentation"]
|
||||
|
||||
return ElicitResult(action="accept", content={"value": ["bug", "feature"]})
|
||||
|
||||
async with Client(mcp, elicitation_handler=elicitation_handler) as client:
|
||||
result = await client.call_tool("my_tool", {})
|
||||
assert result.data == "bug,feature"
|
||||
|
||||
|
||||
async def test_list_dict_multi_select_titled():
|
||||
"""Test list[dict] for multi-select titled."""
|
||||
mcp = FastMCP("TestServer")
|
||||
|
||||
@mcp.tool
|
||||
async def my_tool(ctx: Context) -> str:
|
||||
result = await ctx.elicit(
|
||||
"Choose priorities",
|
||||
response_type=[
|
||||
{
|
||||
"low": {"title": "Low Priority"},
|
||||
"high": {"title": "High Priority"},
|
||||
}
|
||||
],
|
||||
)
|
||||
if result.action == "accept":
|
||||
return ",".join(result.data) # type: ignore[attr-defined]
|
||||
return "declined"
|
||||
|
||||
async def elicitation_handler(message, response_type, params, ctx):
|
||||
# Verify schema has array with anyOf pattern
|
||||
schema = params.requestedSchema
|
||||
assert schema["type"] == "object"
|
||||
assert "value" in schema["properties"]
|
||||
value_schema = schema["properties"]["value"]
|
||||
assert value_schema["type"] == "array"
|
||||
assert "anyOf" in value_schema["items"]
|
||||
any_of = value_schema["items"]["anyOf"]
|
||||
assert {"const": "low", "title": "Low Priority"} in any_of
|
||||
assert {"const": "high", "title": "High Priority"} in any_of
|
||||
|
||||
return ElicitResult(action="accept", content={"value": ["low", "high"]})
|
||||
|
||||
async with Client(mcp, elicitation_handler=elicitation_handler) as client:
|
||||
result = await client.call_tool("my_tool", {})
|
||||
assert result.data == "low,high"
|
||||
|
||||
|
||||
async def test_list_enum_multi_select():
|
||||
"""Test list[Enum] for multi-select with enum in dataclass field."""
|
||||
|
||||
class Priority(Enum):
|
||||
LOW = "low"
|
||||
MEDIUM = "medium"
|
||||
HIGH = "high"
|
||||
|
||||
@dataclass
|
||||
class TaskRequest:
|
||||
priorities: list[Priority]
|
||||
|
||||
schema = get_elicitation_schema(TaskRequest)
|
||||
|
||||
priorities_schema = schema["properties"]["priorities"]
|
||||
assert priorities_schema["type"] == "array"
|
||||
assert "items" in priorities_schema
|
||||
items_schema = priorities_schema["items"]
|
||||
# Should have enum pattern for untitled enums
|
||||
assert "enum" in items_schema
|
||||
assert items_schema["enum"] == ["low", "medium", "high"]
|
||||
|
||||
|
||||
async def test_list_enum_multi_select_direct():
|
||||
"""Test list[Enum] type annotation passed directly to ctx.elicit()."""
|
||||
mcp = FastMCP("TestServer")
|
||||
|
||||
class Priority(Enum):
|
||||
LOW = "low"
|
||||
MEDIUM = "medium"
|
||||
HIGH = "high"
|
||||
|
||||
@mcp.tool
|
||||
async def my_tool(ctx: Context) -> str:
|
||||
result = await ctx.elicit(
|
||||
"Choose priorities",
|
||||
response_type=list[Priority], # Type annotation for multi-select
|
||||
)
|
||||
if result.action == "accept":
|
||||
priorities = result.data # type: ignore[attr-defined]
|
||||
return ",".join(
|
||||
[p.value if isinstance(p, Priority) else str(p) for p in priorities]
|
||||
)
|
||||
return "declined"
|
||||
|
||||
async def elicitation_handler(message, response_type, params, ctx):
|
||||
# Verify schema has array with enum pattern
|
||||
schema = params.requestedSchema
|
||||
assert schema["type"] == "object"
|
||||
assert "value" in schema["properties"]
|
||||
value_schema = schema["properties"]["value"]
|
||||
assert value_schema["type"] == "array"
|
||||
assert "enum" in value_schema["items"]
|
||||
assert value_schema["items"]["enum"] == ["low", "medium", "high"]
|
||||
|
||||
return ElicitResult(action="accept", content={"value": ["low", "high"]})
|
||||
|
||||
async with Client(mcp, elicitation_handler=elicitation_handler) as client:
|
||||
result = await client.call_tool("my_tool", {})
|
||||
assert result.data == "low,high"
|
||||
|
||||
|
||||
async def test_validation_allows_enum_arrays():
|
||||
"""Test validation accepts arrays with enum items."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"priorities": {
|
||||
"type": "array",
|
||||
"items": {"enum": ["low", "medium", "high"]},
|
||||
}
|
||||
},
|
||||
}
|
||||
validate_elicitation_json_schema(schema) # Should not raise
|
||||
|
||||
|
||||
async def test_validation_allows_enum_arrays_with_anyof():
|
||||
"""Test validation accepts arrays with anyOf enum pattern."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"priorities": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"anyOf": [
|
||||
{"const": "low", "title": "Low Priority"},
|
||||
{"const": "high", "title": "High Priority"},
|
||||
]
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
validate_elicitation_json_schema(schema) # Should not raise
|
||||
|
||||
|
||||
async def test_validation_rejects_non_enum_arrays():
|
||||
"""Test validation still rejects arrays of objects."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"users": {
|
||||
"type": "array",
|
||||
"items": {"type": "object", "properties": {"name": {"type": "string"}}},
|
||||
}
|
||||
},
|
||||
}
|
||||
with pytest.raises(TypeError, match="array of objects"):
|
||||
validate_elicitation_json_schema(schema)
|
||||
|
||||
|
||||
async def test_validation_rejects_primitive_arrays():
|
||||
"""Test validation rejects arrays of primitives without enum pattern."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"names": {"type": "array", "items": {"type": "string"}},
|
||||
},
|
||||
}
|
||||
with pytest.raises(TypeError, match="arrays are only allowed"):
|
||||
validate_elicitation_json_schema(schema)
|
||||
|
||||
|
||||
class TestElicitationDefaults:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue