"""
Web Chat Elicitations Handler
Production-ready integration between web chat API and elicitation system.
Handles the complete flow from MCP tool responses to user interactions.
Design Principles:
- DRY: Single handler for all elicitation types
- KISS: Simple, predictable API endpoints
- Dynamic: Handles any elicitation automatically
- Production-ready: Full error handling and logging
"""
from typing import Dict, Any, Optional
from fastapi import HTTPException
import json
from src.observability import get_logger
from .elicitations_manager import elicitation_manager, ElicitationStatus
from .elicitations_ui import form_renderer
logger = get_logger(__name__)
class ElicitationsHandler:
"""
Production-ready handler for web chat elicitation integration.
This handler bridges the gap between MCP tool responses and the
frontend elicitation UI, providing a clean API for the web chat
interface to work with elicitations.
"""
def __init__(self):
self.manager = elicitation_manager
self.renderer = form_renderer
logger.info("elicitations_handler_initialized")
async def process_tool_response(
self,
tool_response: Dict[str, Any],
user_context: Dict[str, Any],
response_callback: Optional[callable] = None
) -> Dict[str, Any]:
"""
Process tool response and handle elicitation if needed.
Args:
tool_response: Response from MCP tool
user_context: User context for the request
response_callback: Callback for handling elicitation responses
Returns:
Dict containing processed response or elicitation data
"""
try:
# Check if response requires elicitation
if not tool_response.get("requires_elicitation"):
return {
"type": "regular_response",
"data": tool_response
}
elicitation_type = tool_response.get("elicitation_type")
if elicitation_type == "mcp_form":
return await self._handle_mcp_elicitation(
tool_response,
user_context,
response_callback
)
elif elicitation_type == "llm_followup":
return {
"type": "llm_followup",
"data": tool_response
}
else:
logger.warning(
"unknown_elicitation_type",
elicitation_type=elicitation_type
)
return {
"type": "error",
"error": f"Unknown elicitation type: {elicitation_type}"
}
except Exception as e:
logger.error(
"failed_to_process_tool_response",
error=str(e),
error_type=type(e).__name__
)
return {
"type": "error",
"error": str(e)
}
async def _handle_mcp_elicitation(
self,
tool_response: Dict[str, Any],
user_context: Dict[str, Any],
response_callback: Optional[callable] = None
) -> Dict[str, Any]:
"""Handle MCP form elicitation."""
try:
elicitation_request = tool_response.get("elicitation_request")
if not elicitation_request:
return {
"type": "error",
"error": "Missing elicitation request"
}
# Validate elicitation request
validation = await self.manager.validate_elicitation_request(
elicitation_request
)
if not validation["valid"]:
return {
"type": "error",
"error": validation["error"]
}
# Create elicitation session
elicitation_result = await self.manager.create_elicitation(
elicitation_request,
response_callback
)
if not elicitation_result["success"]:
return {
"type": "error",
"error": elicitation_result["error"]
}
# Return elicitation data for frontend
return {
"type": "mcp_elicitation",
"elicitation_id": elicitation_result["elicitation_id"],
"rendered": elicitation_result["rendered"],
"expires_at": elicitation_result["expires_at"],
"metadata": elicitation_result["session"]["metadata"]
}
except Exception as e:
logger.error(
"failed_to_handle_mcp_elicitation",
error=str(e),
error_type=type(e).__name__
)
return {
"type": "error",
"error": str(e)
}
async def handle_frontend_response(
self,
elicitation_id: str,
action: str,
data: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Handle response from frontend elicitation UI.
Args:
elicitation_id: ID of the elicitation
action: User action (accept, decline, cancel)
data: Form data (for accept action)
Returns:
Dict containing response handling result
"""
try:
result = await self.manager.handle_response(
elicitation_id,
action,
data
)
if not result["success"]:
raise Exception(result["error"])
# Return result for frontend
return {
"success": True,
"action": action,
"elicitation_id": elicitation_id,
"mcp_response": result["mcp_response"],
"session": result["session"]
}
except Exception as e:
logger.error(
"failed_to_handle_frontend_response",
elicitation_id=elicitation_id,
action=action,
error=str(e),
error_type=type(e).__name__
)
return {
"success": False,
"error": str(e)
}
async def get_elicitation_status(self, elicitation_id: str) -> Dict[str, Any]:
"""Get status of an elicitation."""
try:
result = await self.manager.get_session(elicitation_id)
if not result["success"]:
return {
"success": False,
"error": result["error"]
}
session = result["session"]
return {
"success": True,
"elicitation_id": elicitation_id,
"status": session["status"],
"created_at": session["created_at"],
"expires_at": session["expires_at"],
"response_action": session["response_action"],
"has_response": session["response_data"] is not None
}
except Exception as e:
logger.error(
"failed_to_get_elicitation_status",
elicitation_id=elicitation_id,
error=str(e),
error_type=type(e).__name__
)
return {
"success": False,
"error": str(e)
}
async def cancel_elicitation(self, elicitation_id: str) -> Dict[str, Any]:
"""Cancel an active elicitation."""
try:
result = await self.manager.cancel_elicitation(elicitation_id)
if not result["success"]:
return {
"success": False,
"error": result["error"]
}
return {
"success": True,
"elicitation_id": elicitation_id,
"message": "Elicitation cancelled successfully"
}
except Exception as e:
logger.error(
"failed_to_cancel_elicitation",
elicitation_id=elicitation_id,
error=str(e),
error_type=type(e).__name__
)
return {
"success": False,
"error": str(e)
}
async def cleanup_expired_sessions(self) -> Dict[str, Any]:
"""Clean up expired elicitation sessions."""
try:
result = await self.manager.cleanup_expired_sessions()
return {
"success": True,
"cleaned_count": result["expired_count"],
"remaining_sessions": result["remaining_sessions"]
}
except Exception as e:
logger.error(
"failed_to_cleanup_expired_sessions",
error=str(e),
error_type=type(e).__name__
)
return {
"success": False,
"error": str(e)
}
def get_system_stats(self) -> Dict[str, Any]:
"""Get elicitation system statistics."""
try:
stats = self.manager.get_stats()
return {
"success": True,
"stats": stats
}
except Exception as e:
logger.error(
"failed_to_get_system_stats",
error=str(e),
error_type=type(e).__name__
)
return {
"success": False,
"error": str(e)
}
def build_frontend_response(
self,
response_type: str,
data: Dict[str, Any]
) -> Dict[str, Any]:
"""
Build response for frontend consumption.
Args:
response_type: Type of response (regular, elicitation, error)
data: Response data
Returns:
Dict formatted for frontend
"""
if response_type == "regular_response":
return {
"type": "message",
"content": data.get("data", {}).get("data", {}).get("formatted_response", "Processing complete"),
"metadata": {
"tool_name": data.get("data", {}).get("meta", {}).get("tool_name"),
"trace_id": data.get("data", {}).get("meta", {}).get("trace_id")
}
}
elif response_type == "llm_followup":
return {
"type": "follow_up",
"content": data.get("follow_up_prompt", "I need more information"),
"metadata": {
"tool_name": data.get("data", {}).get("meta", {}).get("tool_name"),
"trace_id": data.get("data", {}).get("meta", {}).get("trace_id"),
"missing_fields": data.get("missing_fields", [])
}
}
elif response_type == "mcp_elicitation":
return {
"type": "elicitation",
"elicitation_id": data["elicitation_id"],
"form_html": data["rendered"]["form_html"],
"javascript": data["rendered"]["javascript"],
"css": data["rendered"]["css"],
"mode": data["rendered"]["mode"],
"expires_at": data["expires_at"],
"metadata": data["metadata"]
}
elif response_type == "error":
return {
"type": "error",
"content": data.get("error", "An error occurred"),
"metadata": {
"error_type": data.get("error_type", "unknown")
}
}
else:
return {
"type": "unknown",
"content": "Unknown response type",
"metadata": {"response_type": response_type}
}
# Global instance for easy access
elicitations_handler = ElicitationsHandler()
