mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-21 04:54:17 +02:00
SEP-1330 enum schema support (#2549)
* SEP-1330 enum schema support for elicitation * Add version badges for 2.14.0 elicitation features * Fix Context.elicit() to handle SEP-1330 enum syntaxes * Guard against empty list in elicit response_type * Add guards for empty dict/list edge cases in elicit * Refactor elicit: extract parsing and response handling to elicitation.py
This commit is contained in:
parent
9ea57f9ef8
commit
c3ffef677b
4 changed files with 626 additions and 114 deletions
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -8,9 +8,8 @@ from collections.abc import Generator, Mapping, Sequence
|
|||
from contextlib import contextmanager
|
||||
from contextvars import ContextVar, Token
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from logging import Logger
|
||||
from typing import Any, Literal, cast, get_origin, overload
|
||||
from typing import Any, overload
|
||||
|
||||
import anyio
|
||||
from mcp import LoggingLevel, ServerSession
|
||||
|
|
@ -41,12 +40,11 @@ from fastmcp.server.elicitation import (
|
|||
AcceptedElicitation,
|
||||
CancelledElicitation,
|
||||
DeclinedElicitation,
|
||||
ScalarElicitationType,
|
||||
get_elicitation_schema,
|
||||
handle_elicit_accept,
|
||||
parse_elicit_response_type,
|
||||
)
|
||||
from fastmcp.server.server import FastMCP
|
||||
from fastmcp.utilities.logging import _clamp_logger, get_logger
|
||||
from fastmcp.utilities.types import get_cached_typeadapter
|
||||
|
||||
logger: Logger = get_logger(name=__name__)
|
||||
to_client_logger: Logger = logger.getChild(suffix="to_client")
|
||||
|
|
@ -674,61 +672,21 @@ class Context:
|
|||
type or dataclass or BaseModel. If it is a primitive type, an
|
||||
object schema with a single "value" field will be generated.
|
||||
"""
|
||||
if response_type is None:
|
||||
schema = {"type": "object", "properties": {}}
|
||||
else:
|
||||
# if the user provided a list of strings, treat it as a Literal
|
||||
if isinstance(response_type, list):
|
||||
if not all(isinstance(item, str) for item in response_type):
|
||||
raise ValueError(
|
||||
"List of options must be a list of strings. Received: "
|
||||
f"{response_type}"
|
||||
)
|
||||
# Convert list of options to Literal type and wrap
|
||||
choice_literal = Literal[tuple(response_type)] # type: ignore
|
||||
response_type = ScalarElicitationType[choice_literal] # type: ignore
|
||||
# if the user provided a primitive scalar, wrap it in an object schema
|
||||
elif (
|
||||
response_type in {bool, int, float, str}
|
||||
or get_origin(response_type) is Literal
|
||||
or (isinstance(response_type, type) and issubclass(response_type, Enum))
|
||||
):
|
||||
response_type = ScalarElicitationType[response_type] # type: ignore
|
||||
|
||||
response_type = cast(type[T], response_type)
|
||||
|
||||
schema = get_elicitation_schema(response_type)
|
||||
config = parse_elicit_response_type(response_type)
|
||||
|
||||
result = await self.session.elicit(
|
||||
message=message,
|
||||
requestedSchema=schema,
|
||||
requestedSchema=config.schema,
|
||||
related_request_id=self.request_id,
|
||||
)
|
||||
|
||||
if result.action == "accept":
|
||||
if response_type is not None:
|
||||
type_adapter = get_cached_typeadapter(response_type)
|
||||
validated_data = cast(
|
||||
T | ScalarElicitationType[T],
|
||||
type_adapter.validate_python(result.content),
|
||||
)
|
||||
if isinstance(validated_data, ScalarElicitationType):
|
||||
return AcceptedElicitation[T](data=validated_data.value)
|
||||
else:
|
||||
return AcceptedElicitation[T](data=cast(T, validated_data))
|
||||
elif result.content:
|
||||
raise ValueError(
|
||||
"Elicitation expected an empty response, but received: "
|
||||
f"{result.content}"
|
||||
)
|
||||
else:
|
||||
return AcceptedElicitation[dict[str, Any]](data={})
|
||||
return handle_elicit_accept(config, result.content)
|
||||
elif result.action == "decline":
|
||||
return DeclinedElicitation()
|
||||
elif result.action == "cancel":
|
||||
return CancelledElicitation()
|
||||
else:
|
||||
# This should never happen, but handle it just in case
|
||||
raise ValueError(f"Unexpected elicitation action: {result.action}")
|
||||
|
||||
def set_state(self, key: str, value: Any) -> None:
|
||||
|
|
|
|||
|
|
@ -1,7 +1,8 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Generic, Literal
|
||||
from enum import Enum
|
||||
from typing import Any, Generic, Literal, get_origin
|
||||
|
||||
from mcp.server.elicitation import (
|
||||
CancelledElicitation,
|
||||
|
|
@ -20,8 +21,11 @@ __all__ = [
|
|||
"AcceptedElicitation",
|
||||
"CancelledElicitation",
|
||||
"DeclinedElicitation",
|
||||
"ElicitConfig",
|
||||
"ScalarElicitationType",
|
||||
"get_elicitation_schema",
|
||||
"handle_elicit_accept",
|
||||
"parse_elicit_response_type",
|
||||
]
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
|
@ -38,49 +42,63 @@ class ElicitationJsonSchema(GenerateJsonSchema):
|
|||
"""
|
||||
|
||||
def generate_inner(self, schema: core_schema.CoreSchema) -> JsonSchemaValue: # type: ignore[override]
|
||||
"""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) # type: ignore[arg-type]
|
||||
# For list schemas, check if items are enums
|
||||
if schema["type"] == "list":
|
||||
return self.list_schema(schema) # type: ignore[arg-type]
|
||||
# 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) # type: ignore[arg-type]
|
||||
# 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 +114,207 @@ class ScalarElicitationType(Generic[T]):
|
|||
value: T
|
||||
|
||||
|
||||
@dataclass
|
||||
class ElicitConfig:
|
||||
"""Configuration for an elicitation request.
|
||||
|
||||
Attributes:
|
||||
schema: The JSON schema to send to the client
|
||||
response_type: The type to validate responses with (None for raw schemas)
|
||||
is_raw: True if schema was built directly (extract "value" from response)
|
||||
"""
|
||||
|
||||
schema: dict[str, Any]
|
||||
response_type: type | None
|
||||
is_raw: bool
|
||||
|
||||
|
||||
def parse_elicit_response_type(response_type: Any) -> ElicitConfig:
|
||||
"""Parse response_type into schema and handling configuration.
|
||||
|
||||
Supports multiple syntaxes:
|
||||
- None: Empty object schema, expect empty response
|
||||
- dict: {"low": {"title": "..."}} -> single-select titled enum
|
||||
- list patterns:
|
||||
- [["a", "b"]] -> multi-select untitled
|
||||
- [{"low": {...}}] -> multi-select titled
|
||||
- ["a", "b"] -> single-select untitled
|
||||
- list[X] type annotation: multi-select with type
|
||||
- Scalar types (bool, int, float, str, Literal, Enum): single value
|
||||
- Other types (dataclass, BaseModel): use directly
|
||||
"""
|
||||
if response_type is None:
|
||||
return ElicitConfig(
|
||||
schema={"type": "object", "properties": {}},
|
||||
response_type=None,
|
||||
is_raw=False,
|
||||
)
|
||||
|
||||
if isinstance(response_type, dict):
|
||||
return _parse_dict_syntax(response_type)
|
||||
|
||||
if isinstance(response_type, list):
|
||||
return _parse_list_syntax(response_type)
|
||||
|
||||
if get_origin(response_type) is list:
|
||||
return _parse_generic_list(response_type)
|
||||
|
||||
if _is_scalar_type(response_type):
|
||||
return _parse_scalar_type(response_type)
|
||||
|
||||
# Other types (dataclass, BaseModel, etc.) - use directly
|
||||
return ElicitConfig(
|
||||
schema=get_elicitation_schema(response_type),
|
||||
response_type=response_type,
|
||||
is_raw=False,
|
||||
)
|
||||
|
||||
|
||||
def _is_scalar_type(response_type: Any) -> bool:
|
||||
"""Check if response_type is a scalar type that needs wrapping."""
|
||||
return (
|
||||
response_type in {bool, int, float, str}
|
||||
or get_origin(response_type) is Literal
|
||||
or (isinstance(response_type, type) and issubclass(response_type, Enum))
|
||||
)
|
||||
|
||||
|
||||
def _parse_dict_syntax(d: dict[str, Any]) -> ElicitConfig:
|
||||
"""Parse dict syntax: {"low": {"title": "..."}} -> single-select titled."""
|
||||
if not d:
|
||||
raise ValueError("Dict response_type cannot be empty.")
|
||||
enum_schema = _dict_to_enum_schema(d, multi_select=False)
|
||||
return ElicitConfig(
|
||||
schema={
|
||||
"type": "object",
|
||||
"properties": {"value": enum_schema},
|
||||
"required": ["value"],
|
||||
},
|
||||
response_type=None,
|
||||
is_raw=True,
|
||||
)
|
||||
|
||||
|
||||
def _parse_list_syntax(lst: list[Any]) -> ElicitConfig:
|
||||
"""Parse list patterns: [[...]], [{...}], or [...]."""
|
||||
# [["a", "b", "c"]] -> multi-select untitled
|
||||
if (
|
||||
len(lst) == 1
|
||||
and isinstance(lst[0], list)
|
||||
and lst[0]
|
||||
and all(isinstance(item, str) for item in lst[0])
|
||||
):
|
||||
return ElicitConfig(
|
||||
schema={
|
||||
"type": "object",
|
||||
"properties": {"value": {"type": "array", "items": {"enum": lst[0]}}},
|
||||
"required": ["value"],
|
||||
},
|
||||
response_type=None,
|
||||
is_raw=True,
|
||||
)
|
||||
|
||||
# [{"low": {"title": "..."}}] -> multi-select titled
|
||||
if len(lst) == 1 and isinstance(lst[0], dict) and lst[0]:
|
||||
enum_schema = _dict_to_enum_schema(lst[0], multi_select=True)
|
||||
return ElicitConfig(
|
||||
schema={
|
||||
"type": "object",
|
||||
"properties": {"value": {"type": "array", "items": enum_schema}},
|
||||
"required": ["value"],
|
||||
},
|
||||
response_type=None,
|
||||
is_raw=True,
|
||||
)
|
||||
|
||||
# ["a", "b", "c"] -> single-select untitled
|
||||
if lst and all(isinstance(item, str) for item in lst):
|
||||
choice_literal = Literal[tuple(lst)] # type: ignore[valid-type]
|
||||
wrapped = ScalarElicitationType[choice_literal] # type: ignore[valid-type]
|
||||
return ElicitConfig(
|
||||
schema=get_elicitation_schema(wrapped), # type: ignore[arg-type]
|
||||
response_type=wrapped, # type: ignore[assignment]
|
||||
is_raw=False,
|
||||
)
|
||||
|
||||
raise ValueError(f"Invalid list response_type format. Received: {lst}")
|
||||
|
||||
|
||||
def _parse_generic_list(response_type: Any) -> ElicitConfig:
|
||||
"""Parse list[X] type annotation -> multi-select."""
|
||||
wrapped = ScalarElicitationType[response_type] # type: ignore[valid-type]
|
||||
return ElicitConfig(
|
||||
schema=get_elicitation_schema(wrapped), # type: ignore[arg-type]
|
||||
response_type=wrapped, # type: ignore[assignment]
|
||||
is_raw=False,
|
||||
)
|
||||
|
||||
|
||||
def _parse_scalar_type(response_type: Any) -> ElicitConfig:
|
||||
"""Parse scalar types (bool, int, float, str, Literal, Enum)."""
|
||||
wrapped = ScalarElicitationType[response_type] # type: ignore[valid-type]
|
||||
return ElicitConfig(
|
||||
schema=get_elicitation_schema(wrapped), # type: ignore[arg-type]
|
||||
response_type=wrapped, # type: ignore[assignment]
|
||||
is_raw=False,
|
||||
)
|
||||
|
||||
|
||||
def handle_elicit_accept(
|
||||
config: ElicitConfig, content: Any
|
||||
) -> AcceptedElicitation[Any]:
|
||||
"""Handle an accepted elicitation response.
|
||||
|
||||
Args:
|
||||
config: The elicitation configuration from parse_elicit_response_type
|
||||
content: The response content from the client
|
||||
|
||||
Returns:
|
||||
AcceptedElicitation with the extracted/validated data
|
||||
"""
|
||||
# For raw schemas (dict/nested-list syntax), extract value directly
|
||||
if config.is_raw:
|
||||
if not isinstance(content, dict) or "value" not in content:
|
||||
raise ValueError("Elicitation response missing required 'value' field.")
|
||||
return AcceptedElicitation[Any](data=content["value"])
|
||||
|
||||
# For typed schemas, validate with Pydantic
|
||||
if config.response_type is not None:
|
||||
type_adapter = get_cached_typeadapter(config.response_type)
|
||||
validated_data = type_adapter.validate_python(content)
|
||||
if isinstance(validated_data, ScalarElicitationType):
|
||||
return AcceptedElicitation[Any](data=validated_data.value)
|
||||
return AcceptedElicitation[Any](data=validated_data)
|
||||
|
||||
# For None response_type, expect empty response
|
||||
if content:
|
||||
raise ValueError(
|
||||
f"Elicitation expected an empty response, but received: {content}"
|
||||
)
|
||||
return AcceptedElicitation[dict[str, Any]](data={})
|
||||
|
||||
|
||||
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 +416,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 +424,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."
|
||||
)
|
||||
|
|
|
|||
|
|
@ -515,7 +515,7 @@ class TestValidation:
|
|||
"""Test that nested object schemas are rejected."""
|
||||
|
||||
with pytest.raises(
|
||||
TypeError, match="has type 'object' which is not a primitive type"
|
||||
TypeError, match="is an object, but nested objects are not allowed"
|
||||
):
|
||||
validate_elicitation_json_schema(
|
||||
{
|
||||
|
|
@ -530,11 +530,9 @@ class TestValidation:
|
|||
)
|
||||
|
||||
async def test_schema_validation_rejects_arrays(self):
|
||||
"""Test that array schemas are rejected."""
|
||||
"""Test that non-enum array schemas are rejected."""
|
||||
|
||||
with pytest.raises(
|
||||
TypeError, match="has type 'array' which is not a primitive type"
|
||||
):
|
||||
with pytest.raises(TypeError, match="is an array, but arrays are only allowed"):
|
||||
validate_elicitation_json_schema(
|
||||
{
|
||||
"type": "object",
|
||||
|
|
@ -692,8 +690,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"
|
||||
|
|
@ -714,7 +712,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",
|
||||
|
|
@ -722,14 +723,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