From 759135e2a39381873a1c4d76d6044bf06e149138 Mon Sep 17 00:00:00 2001 From: Tapan Chugh Date: Sat, 2 Aug 2025 14:03:48 -0700 Subject: [PATCH 1/3] WIP: PoC demonstrating new enum schemas + multi-selection --- .../elicitations/elicitation_forms_server.py | 89 ++++++++++++++----- src/mcp_agent/human_input/elicitation_form.py | 80 +++++++++++++++++ src/mcp_agent/human_input/form_elements.py | 59 ++++++++++++ 3 files changed, 206 insertions(+), 22 deletions(-) create mode 100644 src/mcp_agent/human_input/form_elements.py diff --git a/examples/mcp/elicitations/elicitation_forms_server.py b/examples/mcp/elicitations/elicitation_forms_server.py index 44ac8129..e16d79cf 100644 --- a/examples/mcp/elicitations/elicitation_forms_server.py +++ b/examples/mcp/elicitations/elicitation_forms_server.py @@ -7,7 +7,7 @@ import logging import sys -from typing import Optional +from typing import List, Optional from mcp import ReadResourceResult from mcp.server.elicitation import ( @@ -31,9 +31,43 @@ mcp = FastMCP("Elicitation Forms Demo Server", log_level="INFO") +async def _elicit_skip_validation(ctx, message: str, schema: type[BaseModel]): + """Helper to use session.elicit directly to bypass type validation. + + Note: This is just a temporary workound until we update the SDK.""" + json_schema = schema.model_json_schema() + + elicit_result = await ctx.request_context.session.elicit( + message=message, requestedSchema=json_schema, related_request_id=ctx.request_id + ) + + # Convert the result to match the expected format + if elicit_result.action == "accept" and elicit_result.content: + logger.info(f"Elicit result content: {elicit_result.content}") + logger.info(f"Content type: {type(elicit_result.content)}") + validated_data = schema.model_validate(elicit_result.content) + return AcceptedElicitation(data=validated_data) + elif elicit_result.action == "decline": + return DeclinedElicitation() + else: + return CancelledElicitation() + + +def _serialize_enum(data: dict[str, str]) -> list[dict[str, str]]: + 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) @@ -44,12 +78,19 @@ class EventRegistration(BaseModel): event_date: str = Field( description="Which event date works for you?", json_schema_extra={"format": "date"} ) + workshops: List[str] = Field( + description="Select workshops to attend (2-4 required)", + min_length=2, + max_length=4, + json_schema_extra={"items": {"anyOf": _serialize_enum(workshop_names)}}, + ) dietary_requirements: Optional[str] = Field( None, description="Any dietary requirements? (optional)", max_length=200 ) - result = await mcp.get_context().elicit( - "Register for the fast-agent conference - fill out your details", + result = await _elicit_skip_validation( + ctx=mcp.get_context(), + message="Register for the fast-agent conference - fill out your details", schema=EventRegistration, ) @@ -61,7 +102,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" @@ -80,6 +124,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) @@ -88,23 +139,16 @@ 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": _serialize_enum(categories)}, ) review_text: str = Field( description="Tell us about your experience", min_length=10, max_length=1000 ) - result = await mcp.get_context().elicit( - "Share your product review - Help others make informed decisions!", schema=ProductReview + result = await _elicit_skip_validation( + mcp.get_context(), + "Share your product review - Help others make informed decisions!", + schema=ProductReview, ) match result: @@ -114,7 +158,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) @@ -136,20 +180,21 @@ 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( description="Choose your preferred theme", - json_schema_extra={ - "enum": ["light", "dark", "auto"], - "enumNames": ["Light Theme", "Dark Theme", "Auto (System)"], - }, + json_schema_extra={"oneOf": [_serialize_enum(themes)]}, ) privacy_public: bool = Field(False, description="Make your profile public?") items_per_page: int = Field(description="Items to show per page (10-100)", ge=10, le=100) - result = await mcp.get_context().elicit("Update your account settings", schema=AccountSettings) + result = await _elicit_skip_validation( + mcp.get_context(), "Update your account settings", schema=AccountSettings + ) match result: case AcceptedElicitation(data=data): @@ -157,7 +202,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}", ] diff --git a/src/mcp_agent/human_input/elicitation_form.py b/src/mcp_agent/human_input/elicitation_form.py index 0e118abb..18dc9457 100644 --- a/src/mcp_agent/human_input/elicitation_form.py +++ b/src/mcp_agent/human_input/elicitation_form.py @@ -26,6 +26,7 @@ from mcp_agent.human_input.elicitation_forms import ELICITATION_STYLE from mcp_agent.human_input.elicitation_state import elicitation_state +from mcp_agent.human_input.form_elements import ValidatedCheckboxList class SimpleNumberValidator(Validator): @@ -454,6 +455,24 @@ def _create_field(self, field_name: str, field_def: Dict[str, Any]): hints = [] format_hint = None + # Check if this is an array type with enum/oneOf/anyOf items + if field_type == "array" and "items" in field_def: + items_def = field_def["items"] + + # Add minItems/maxItems hints + min_items = field_def.get("minItems") + max_items = field_def.get("maxItems") + + if min_items is not None and max_items is not None: + if min_items == max_items: + hints.append(f"select exactly {min_items}") + else: + hints.append(f"select {min_items}-{max_items}") + elif min_items is not None: + hints.append(f"select at least {min_items}") + elif max_items is not None: + hints.append(f"select up to {max_items}") + if field_type == "string": constraints = self._extract_string_constraints(field_def) if constraints.get("minLength"): @@ -506,6 +525,7 @@ def _create_field(self, field_name: str, field_def: Dict[str, Any]): return HSplit([label, Frame(checkbox)]) elif field_type == "string" and "enum" in field_def: + # Leaving this here for existing enum schema enum_values = field_def["enum"] enum_names = field_def.get("enumNames", enum_values) values = [(val, name) for val, name in zip(enum_values, enum_names)] @@ -515,6 +535,51 @@ def _create_field(self, field_name: str, field_def: Dict[str, Any]): return HSplit([label, Frame(radio_list, height=min(len(values) + 2, 6))]) + elif field_type == "string" and "oneOf" in field_def: + # Handle oneOf pattern for single selection enums + values = [] + for option in field_def["oneOf"]: + if "const" in option: + value = option["const"] + title = option.get("title", str(value)) + values.append((value, title)) + + if values: + radio_list = RadioList(values=values) + self.field_widgets[field_name] = radio_list + return HSplit([label, Frame(radio_list, height=min(len(values) + 2, 6))]) + + elif field_type == "array" and "items" in field_def: + # Handle array types with enum/oneOf/anyOf items + items_def = field_def["items"] + values = [] + # oneOf/anyOf pattern + options = items_def.get("oneOf", []) + if not options: + options = items_def.get("anyOf", []) + + for option in options: + if "const" in option: + value = option["const"] + title = option.get("title", str(value)) + values.append((value, title)) + + if values: + # Create checkbox list for multi-selection + min_items = field_def.get("minItems") + max_items = field_def.get("maxItems") + + checkbox_list = ValidatedCheckboxList( + values=values, min_items=min_items, max_items=max_items + ) + + # Store the widget directly (consistent with other widgets) + self.field_widgets[field_name] = checkbox_list + + # Create scrollable frame if many options + height = min(len(values) + 2, 8) + return HSplit([label, Frame(checkbox_list, height=height)]) + else: # Text/number input validator = None @@ -620,6 +685,10 @@ def _validate_form(self) -> tuple[bool, Optional[str]]: if widget.validation_error: title = field_def.get("title", field_name) return False, f"'{title}': {widget.validation_error.message}" + elif isinstance(widget, ValidatedCheckboxList): + if widget.validation_error: + title = field_def.get("title", field_name) + return False, f"'{title}': {widget.validation_error.message}" # Then check if required fields are empty for field_name in self.required_fields: @@ -636,6 +705,10 @@ def _validate_form(self) -> tuple[bool, Optional[str]]: if widget.current_value is None: title = self.properties[field_name].get("title", field_name) return False, f"'{title}' is required" + elif isinstance(widget, ValidatedCheckboxList): + if not widget.current_values: + title = self.properties[field_name].get("title", field_name) + return False, f"'{title}' is required" return True, None @@ -677,6 +750,13 @@ def _get_form_data(self) -> Dict[str, Any]: if widget.current_value is not None: data[field_name] = widget.current_value + elif isinstance(widget, ValidatedCheckboxList): + selected_values = widget.current_values + if selected_values: + data[field_name] = list(selected_values) + elif field_name not in self.required_fields: + data[field_name] = [] + return data def _accept(self): diff --git a/src/mcp_agent/human_input/form_elements.py b/src/mcp_agent/human_input/form_elements.py new file mode 100644 index 00000000..36fe4fef --- /dev/null +++ b/src/mcp_agent/human_input/form_elements.py @@ -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 From 79add4a5e0ae06da0f289e8f85e5732f25ca562e Mon Sep 17 00:00:00 2001 From: Tapan Chugh Date: Wed, 27 Aug 2025 15:29:34 -0700 Subject: [PATCH 2/3] Cleanup checkbox impl --- .../elicitations/elicitation_forms_server.py | 54 ++++++++---------- src/mcp_agent/human_input/elicitation_form.py | 55 +++++++++++-------- 2 files changed, 56 insertions(+), 53 deletions(-) diff --git a/examples/mcp/elicitations/elicitation_forms_server.py b/examples/mcp/elicitations/elicitation_forms_server.py index 5da47d5c..fd3100c3 100644 --- a/examples/mcp/elicitations/elicitation_forms_server.py +++ b/examples/mcp/elicitations/elicitation_forms_server.py @@ -7,7 +7,7 @@ import logging import sys -from typing import List, Optional +from typing import List, Optional, TypedDict from mcp import ReadResourceResult from mcp.server.elicitation import ( @@ -31,29 +31,26 @@ mcp = FastMCP("Elicitation Forms Demo Server", log_level="INFO") -async def _elicit_skip_validation(ctx, message: str, schema: type[BaseModel]): - """Helper to use session.elicit directly to bypass type validation. +class TitledEnumOption(TypedDict): + """Type definition for oneOf/anyOf schema options.""" - Note: This is just a temporary workound until we update the SDK.""" - json_schema = schema.model_json_schema() + const: str + title: str - elicit_result = await ctx.request_context.session.elicit( - message=message, requestedSchema=json_schema, related_request_id=ctx.request_id - ) - # Convert the result to match the expected format - if elicit_result.action == "accept" and elicit_result.content: - logger.info(f"Elicit result content: {elicit_result.content}") - logger.info(f"Content type: {type(elicit_result.content)}") - validated_data = schema.model_validate(elicit_result.content) - return AcceptedElicitation(data=validated_data) - elif elicit_result.action == "decline": - return DeclinedElicitation() - else: - return CancelledElicitation() +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 -def _serialize_enum(data: dict[str, str]) -> list[dict[str, str]]: + 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()] @@ -79,18 +76,18 @@ class EventRegistration(BaseModel): description="Which event date works for you?", json_schema_extra={"format": "date"} ) workshops: List[str] = Field( + default=["ai_basics", "llm_apps"], description="Select workshops to attend (2-4 required)", min_length=2, max_length=4, - json_schema_extra={"items": {"anyOf": _serialize_enum(workshop_names)}}, + json_schema_extra={"items": {"anyOf": _create_enum_schema_options(workshop_names)}}, ) dietary_requirements: Optional[str] = Field( None, description="Any dietary requirements? (optional)", max_length=200 ) - result = await _elicit_skip_validation( - ctx=mcp.get_context(), - message="Register for the fast-agent conference - fill out your details", + result = await mcp.get_context().elicit( + "Register for the fast-agent conference - fill out your details", schema=EventRegistration, ) @@ -139,14 +136,13 @@ class ProductReview(BaseModel): ) category: str = Field( description="What type of product is this?", - json_schema_extra={"oneOf": _serialize_enum(categories)}, + json_schema_extra={"oneOf": _create_enum_schema_options(categories)}, ) review_text: str = Field( description="Tell us about your experience", min_length=10, max_length=1000 ) - result = await _elicit_skip_validation( - mcp.get_context(), + result = await mcp.get_context().elicit( "Share your product review - Help others make informed decisions!", schema=ProductReview, ) @@ -188,16 +184,14 @@ class AccountSettings(BaseModel): theme: str = Field( "dark", description="Choose your preferred theme", - json_schema_extra={"oneOf": [_serialize_enum(themes)]}, + json_schema_extra={"oneOf": _create_enum_schema_options(themes)}, ) privacy_public: bool = Field(False, description="Make your profile public?") items_per_page: int = Field( 25, description="Items to show per page (10-100)", ge=10, le=100 ) - result = await _elicit_skip_validation( - mcp.get_context(), "Update your account settings", schema=AccountSettings - ) + result = await mcp.get_context().elicit("Update your account settings", schema=AccountSettings) match result: case AcceptedElicitation(data=data): diff --git a/src/mcp_agent/human_input/elicitation_form.py b/src/mcp_agent/human_input/elicitation_form.py index 942b905d..8825d2dc 100644 --- a/src/mcp_agent/human_input/elicitation_form.py +++ b/src/mcp_agent/human_input/elicitation_form.py @@ -1,7 +1,7 @@ """Simplified, robust elicitation form dialog.""" from datetime import date, datetime -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional, Tuple from mcp.types import ElicitRequestedSchema from prompt_toolkit import Application @@ -414,6 +414,28 @@ def set_initial_focus(): self.app.invalidate() # Ensure layout is built set_initial_focus() + def _extract_enum_schema_options(self, schema_def: Dict[str, Any]) -> List[Tuple[str, str]]: + """Extract options from oneOf/anyOf schema patterns. + + Args: + schema_def: Schema definition potentially containing oneOf/anyOf + + Returns: + List of (value, title) tuples for the options + """ + values = [] + options = schema_def.get("oneOf", []) + if not options: + options = schema_def.get("anyOf", []) + + for option in options: + if "const" in option: + value = option["const"] + title = option.get("title", str(value)) + values.append((value, title)) + + return values + def _extract_string_constraints(self, field_def: Dict[str, Any]) -> Dict[str, Any]: """Extract string constraints from field definition, handling anyOf schemas.""" constraints = {} @@ -436,7 +458,6 @@ def _extract_string_constraints(self, field_def: Dict[str, Any]) -> Dict[str, An return constraints - def _create_field(self, field_name: str, field_def: Dict[str, Any]): """Create a field widget.""" @@ -539,40 +560,28 @@ def _create_field(self, field_name: str, field_def: Dict[str, Any]): elif field_type == "string" and "oneOf" in field_def: # Handle oneOf pattern for single selection enums - values = [] - for option in field_def["oneOf"]: - if "const" in option: - value = option["const"] - title = option.get("title", str(value)) - values.append((value, title)) - + values = self._extract_enum_schema_options(field_def) if values: - radio_list = RadioList(values=values) + default_value = field_def.get("default") + radio_list = RadioList(values=values, default=default_value) self.field_widgets[field_name] = radio_list return HSplit([label, Frame(radio_list, height=min(len(values) + 2, 6))]) elif field_type == "array" and "items" in field_def: # Handle array types with enum/oneOf/anyOf items items_def = field_def["items"] - values = [] - # oneOf/anyOf pattern - options = items_def.get("oneOf", []) - if not options: - options = items_def.get("anyOf", []) - - for option in options: - if "const" in option: - value = option["const"] - title = option.get("title", str(value)) - values.append((value, title)) - + values = self._extract_enum_schema_options(items_def) if values: # Create checkbox list for multi-selection min_items = field_def.get("minItems") max_items = field_def.get("maxItems") + default_values = field_def.get("default", []) checkbox_list = ValidatedCheckboxList( - values=values, min_items=min_items, max_items=max_items + values=values, + default_values=default_values, + min_items=min_items, + max_items=max_items, ) # Store the widget directly (consistent with other widgets) From f1069f3b192cdc6e51d5b099916f69399c0e1cc0 Mon Sep 17 00:00:00 2001 From: Tapan Chugh Date: Thu, 28 Aug 2025 10:56:45 -0700 Subject: [PATCH 3/3] add bare enum support --- src/mcp_agent/human_input/elicitation_form.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/mcp_agent/human_input/elicitation_form.py b/src/mcp_agent/human_input/elicitation_form.py index 8825d2dc..cebfbc83 100644 --- a/src/mcp_agent/human_input/elicitation_form.py +++ b/src/mcp_agent/human_input/elicitation_form.py @@ -415,15 +415,25 @@ def set_initial_focus(): set_initial_focus() def _extract_enum_schema_options(self, schema_def: Dict[str, Any]) -> List[Tuple[str, str]]: - """Extract options from oneOf/anyOf schema patterns. + """Extract options from oneOf/anyOf/enum schema patterns. Args: - schema_def: Schema definition potentially containing oneOf/anyOf + schema_def: Schema definition potentially containing oneOf/anyOf/enum Returns: List of (value, title) tuples for the options """ values = [] + + # First check for bare enum (most common pattern for arrays) + if "enum" in schema_def: + enum_values = schema_def["enum"] + enum_names = schema_def.get("enumNames", enum_values) + for val, name in zip(enum_values, enum_names): + values.append((val, str(name))) + return values + + # Then check for oneOf/anyOf patterns options = schema_def.get("oneOf", []) if not options: options = schema_def.get("anyOf", [])