Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 62 additions & 18 deletions examples/mcp/elicitations/elicitation_forms_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

import logging
import sys
from typing import Optional, TypedDict

from mcp import ReadResourceResult
from mcp.server.elicitation import (
Expand All @@ -30,20 +31,63 @@
mcp = FastMCP("Elicitation Forms Demo Server", log_level="INFO")


class TitledEnumOption(TypedDict):
"""Type definition for oneOf/anyOf schema options."""

const: str
title: str


def _create_enum_schema_options(data: dict[str, str]) -> list[TitledEnumOption]:
"""Convert a dictionary to oneOf/anyOf schema format.

Args:
data: Dictionary mapping enum values to display titles

Returns:
List of schema options with 'const' and 'title' fields

Example:
>>> _create_enum_schema_options({"dark": "Dark Mode", "light": "Light Mode"})
[{"const": "dark", "title": "Dark Mode"}, {"const": "light", "title": "Light Mode"}]
"""
return [{"const": k, "title": v} for k, v in data.items()]


@mcp.resource(uri="elicitation://event-registration")
async def event_registration() -> ReadResourceResult:
"""Register for a tech conference event."""
workshop_names = {
"ai_basics": "AI Fundamentals",
"llm_apps": "Building LLM Applications",
"prompt_eng": "Prompt Engineering",
"rag_systems": "RAG Systems",
"fine_tuning": "Model Fine-tuning",
"deployment": "Production Deployment",
}

class EventRegistration(BaseModel):
name: str = Field(description="Your full name", min_length=2, max_length=100)
email: str = Field(description="Your email address", json_schema_extra={"format": "email"})
company_website: str | None = Field(
None, description="Your company website (optional)", json_schema_extra={"format": "uri"}
)
workshops: list[str] = Field(
description="Select the workshops you want to attend",
min_length=1,
max_length=3,
json_schema_extra={
"items": {
"enum": list(workshop_names.keys()),
"enumNames": list(workshop_names.values()),
},
"uniqueItems": True,
},
)
event_date: str = Field(
description="Which event date works for you?", json_schema_extra={"format": "date"}
)
dietary_requirements: str | None = Field(
dietary_requirements: Optional[str] = Field(
None, description="Any dietary requirements? (optional)", max_length=200
)

Expand All @@ -60,7 +104,10 @@ class EventRegistration(BaseModel):
f"🏢 Company: {data.company_website or 'Not provided'}",
f"📅 Event Date: {data.event_date}",
f"🍽️ Dietary Requirements: {data.dietary_requirements or 'None'}",
f"🎓 Workshops ({len(data.workshops)} selected):",
]
for workshop in data.workshops:
lines.append(f" • {workshop_names.get(workshop, workshop)}")
response = "\n".join(lines)
case DeclinedElicitation():
response = "Registration declined - no ticket reserved"
Expand All @@ -79,6 +126,13 @@ class EventRegistration(BaseModel):
@mcp.resource(uri="elicitation://product-review")
async def product_review() -> ReadResourceResult:
"""Submit a product review with rating and comments."""
categories = {
"electronics": "Electronics",
"books": "Books & Media",
"clothing": "Clothing",
"home": "Home & Garden",
"sports": "Sports & Outdoors",
}

class ProductReview(BaseModel):
rating: int = Field(description="Rate this product (1-5 stars)", ge=1, le=5)
Expand All @@ -87,16 +141,7 @@ class ProductReview(BaseModel):
)
category: str = Field(
description="What type of product is this?",
json_schema_extra={
"enum": ["electronics", "books", "clothing", "home", "sports"],
"enumNames": [
"Electronics",
"Books & Media",
"Clothing",
"Home & Garden",
"Sports & Outdoors",
],
},
json_schema_extra={"oneOf": _create_enum_schema_options(categories)},
)
review_text: str = Field(
description="Tell us about your experience",
Expand All @@ -112,7 +157,7 @@ class ProductReview(BaseModel):

Overall, highly recommended!""",
min_length=10,
max_length=1000
max_length=1000,
)

result = await mcp.get_context().elicit(
Expand All @@ -127,7 +172,7 @@ class ProductReview(BaseModel):
"🎯 Product Review Submitted!",
f"⭐ Rating: {stars} ({data.rating}/5)",
f"📊 Satisfaction: {data.satisfaction}/10.0",
f"📦 Category: {data.category.replace('_', ' ').title()}",
f"📦 Category: {categories.get(data.category, data.category)}",
f"💬 Review: {data.review_text}",
]
response = "\n".join(lines)
Expand All @@ -149,16 +194,15 @@ class ProductReview(BaseModel):
async def account_settings() -> ReadResourceResult:
"""Configure your account settings and preferences."""

themes = {"light": "Light Theme", "dark": "Dark Theme", "auto": "Auto (System)"}

class AccountSettings(BaseModel):
email_notifications: bool = Field(True, description="Receive email notifications?")
marketing_emails: bool = Field(False, description="Subscribe to marketing emails?")
theme: str = Field(
"dark",
description="Choose your preferred theme",
json_schema_extra={
"enum": ["light", "dark", "auto"],
"enumNames": ["Light Theme", "Dark Theme", "Auto (System)"],
},
json_schema_extra={"oneOf": _create_enum_schema_options(themes)},
)
privacy_public: bool = Field(False, description="Make your profile public?")
items_per_page: int = Field(
Expand All @@ -173,7 +217,7 @@ class AccountSettings(BaseModel):
"⚙️ Account Settings Updated!",
f"📧 Email notifications: {'On' if data.email_notifications else 'Off'}",
f"📬 Marketing emails: {'On' if data.marketing_emails else 'Off'}",
f"🎨 Theme: {data.theme.title()}",
f"🎨 Theme: {themes.get(data.theme, data.theme)}",
f"👥 Public profile: {'Yes' if data.privacy_public else 'No'}",
f"📄 Items per page: {data.items_per_page}",
]
Expand Down
4 changes: 3 additions & 1 deletion examples/mcp/elicitations/forms_demo.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,8 @@ async def main():
# Example 1: Event Registration
console.print("[bold yellow]Example 1: Event Registration Form[/bold yellow]")
console.print(
"[dim]Demonstrates: string validation, email format, URL format, date format[/dim]"
"[dim]Demonstrates: string validation, email format, URL format, date format, "
"multi-select enums[/dim]"
)
result = await agent["forms-demo"].get_resource("elicitation://event-registration")

Expand Down Expand Up @@ -95,6 +96,7 @@ async def main():
console.print("• [green]String validation[/green] (min/max length)")
console.print("• [green]Number validation[/green] (range constraints)")
console.print("• [green]Radio selections[/green] (enum dropdowns)")
console.print("• [green]Multi-select enums[/green] (checkbox lists)")
console.print("• [green]Boolean selections[/green] (checkboxes)")
console.print("• [green]Format validation[/green] (email, URL, date, datetime)")
console.print("• [green]Multiline text[/green] (expandable text areas)")
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ classifiers = [
requires-python = ">=3.13.5,<3.14"
dependencies = [
"fastapi>=0.121.0",
"mcp==1.22.0",
"mcp==1.23.1",
"opentelemetry-distro>=0.55b0",
"opentelemetry-exporter-otlp-proto-http>=1.7.0",
"pydantic-settings>=2.7.0",
Expand Down
59 changes: 59 additions & 0 deletions src/fast_agent/human_input/form_elements.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
"""Custom form elements for elicitation forms."""

from typing import Optional, Sequence, TypeVar

from prompt_toolkit.formatted_text import AnyFormattedText
from prompt_toolkit.validation import ValidationError
from prompt_toolkit.widgets import CheckboxList

_T = TypeVar("_T")


class ValidatedCheckboxList(CheckboxList[_T]):
"""CheckboxList with min/max items validation."""

def __init__(
self,
values: Sequence[tuple[_T, AnyFormattedText]],
default_values: Optional[Sequence[_T]] = None,
min_items: Optional[int] = None,
max_items: Optional[int] = None,
):
"""
Initialize checkbox list with validation.

Args:
values: List of (value, label) tuples
default_values: Initially selected values
min_items: Minimum number of items that must be selected
max_items: Maximum number of items that can be selected
"""
super().__init__(values, default_values=default_values)
self.min_items = min_items
self.max_items = max_items

@property
def validation_error(self) -> Optional[ValidationError]:
"""
Check if current selection is valid.

Returns:
ValidationError if invalid, None if valid
"""
selected_count = len(self.current_values)

if self.min_items is not None and selected_count < self.min_items:
if self.min_items == 1:
message = "At least 1 selection required"
else:
message = f"At least {self.min_items} selections required"
return ValidationError(message=message)

if self.max_items is not None and selected_count > self.max_items:
if self.max_items == 1:
message = "Only 1 selection allowed"
else:
message = f"Maximum {self.max_items} selections allowed"
return ValidationError(message=message)

return None
Loading
Loading