Handler
Usage
Actions
The handler returns one of three actions:| Action | Description |
|---|---|
accept | Submit collected content to the server |
decline | Reject the request without canceling the workflow |
cancel | Terminate the operation entirely |
Gather user input during tool execution
from dedalus_mcp.client import ClientCapabilitiesConfig, open_connection
from dedalus_mcp.types import (
ElicitRequestParams,
ElicitResult,
ErrorData,
)
async def elicitation_handler(
context: object,
params: ElicitRequestParams,
) -> ElicitResult | ErrorData:
"""Handle elicitation requests from the server."""
print(f"\nServer requests input: {params.message}")
# Collect values for each field in the schema
content = {}
schema = params.requestedSchema
properties = schema.get("properties", {})
required = schema.get("required", [])
for field, field_schema in properties.items():
field_type = field_schema.get("type", "string")
is_required = field in required
value = input(f"{field} ({field_type}): ")
if not value and is_required:
return ElicitResult(action="decline")
# Type coercion
if field_type == "boolean":
content[field] = value.lower() in ("true", "yes", "1", "y")
elif field_type == "integer":
content[field] = int(value)
elif field_type == "number":
content[field] = float(value)
else:
content[field] = value
return ElicitResult(action="accept", content=content)
capabilities = ClientCapabilitiesConfig(elicitation=elicitation_handler)
async with open_connection(
url="http://127.0.0.1:8000/mcp",
transport="streamable-http",
capabilities=capabilities,
) as client:
# Server can now request user input during tool execution
result = await client.call_tool("deploy", {"env": "production"})
| Action | Description |
|---|---|
accept | Submit collected content to the server |
decline | Reject the request without canceling the workflow |
cancel | Terminate the operation entirely |
# Accept with content
return ElicitResult(action="accept", content={"confirmed": True})
# Decline (user refused)
return ElicitResult(action="decline")
# Cancel (abort operation)
return ElicitResult(action="cancel")
async def elicitation_handler(context, params):
"""Auto-accept with default values for testing."""
content = {}
properties = params.requestedSchema.get("properties", {})
for field, schema in properties.items():
field_type = schema.get("type", "string")
if field_type == "boolean":
content[field] = True
elif field_type in ("integer", "number"):
content[field] = 0
else:
content[field] = "auto-filled"
return ElicitResult(action="accept", content=content)
Was this page helpful?