"""
Elicitation API Endpoints
Production-ready FastAPI endpoints for handling MCP elicitations.
These endpoints provide the frontend with everything needed to work
with elicitations without hardcoding specific types.
Design Principles:
- DRY: One set of endpoints for all elicitation types
- KISS: Simple, predictable API
- Dynamic: Handles any elicitation automatically
- Production-ready: Full error handling and validation
"""
from typing import Optional, Dict, Any
from fastapi import HTTPException, status
from pydantic import BaseModel, Field
from src.observability import get_logger
from .elicitations_handler import elicitations_handler
logger = get_logger(__name__)
# Request/Response Models
class ElicitationResponseRequest(BaseModel):
"""Request model for elicitation responses."""
elicitation_id: str = Field(description="ID of the elicitation")
action: str = Field(description="User action: accept, decline, cancel")
data: Optional[Dict[str, Any]] = Field(default=None, description="Form data for accept action")
class ElicitationStatusResponse(BaseModel):
"""Response model for elicitation status."""
success: bool
elicitation_id: Optional[str] = None
status: Optional[str] = None
created_at: Optional[str] = None
expires_at: Optional[str] = None
response_action: Optional[str] = None
has_response: Optional[bool] = None
error: Optional[str] = None
class ElicitationCancelResponse(BaseModel):
"""Response model for elicitation cancellation."""
success: bool
elicitation_id: Optional[str] = None
message: Optional[str] = None
error: Optional[str] = None
class SystemStatsResponse(BaseModel):
"""Response model for system statistics."""
success: bool
stats: Optional[Dict[str, Any]] = None
error: Optional[str] = None
def register_elicitation_endpoints(app):
"""Register all elicitation endpoints with the FastAPI app."""
@app.post("/elicitations/response", response_model=Dict[str, Any])
async def handle_elicitation_response(request: ElicitationResponseRequest):
"""
Handle response from frontend elicitation UI.
This endpoint processes user responses to MCP elicitations,
including form submissions, declines, and cancellations.
"""
try:
logger.info(
"elicitation_response_received",
elicitation_id=request.elicitation_id,
action=request.action,
has_data=bool(request.data)
)
result = await elicitations_handler.handle_frontend_response(
elicitation_id=request.elicitation_id,
action=request.action,
data=request.data
)
if not result["success"]:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=result["error"]
)
logger.info(
"elicitation_response_processed",
elicitation_id=request.elicitation_id,
action=request.action
)
return result
except HTTPException:
raise
except Exception as e:
logger.error(
"failed_to_handle_elicitation_response",
elicitation_id=request.elicitation_id,
action=request.action,
error=str(e),
error_type=type(e).__name__
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to process elicitation response: {str(e)}"
)
@app.get("/elicitations/{elicitation_id}/status", response_model=ElicitationStatusResponse)
async def get_elicitation_status(elicitation_id: str):
"""
Get the status of an elicitation.
This endpoint allows the frontend to check the status
of an active elicitation session.
"""
try:
logger.info(
"elicitation_status_requested",
elicitation_id=elicitation_id
)
result = await elicitations_handler.get_elicitation_status(elicitation_id)
if not result["success"]:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=result["error"]
)
return ElicitationStatusResponse(**result)
except HTTPException:
raise
except Exception as e:
logger.error(
"failed_to_get_elicitation_status",
elicitation_id=elicitation_id,
error=str(e),
error_type=type(e).__name__
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to get elicitation status: {str(e)}"
)
@app.post("/elicitations/{elicitation_id}/cancel", response_model=ElicitationCancelResponse)
async def cancel_elicitation(elicitation_id: str):
"""
Cancel an active elicitation.
This endpoint allows the frontend to cancel an active
elicitation session.
"""
try:
logger.info(
"elicitation_cancellation_requested",
elicitation_id=elicitation_id
)
result = await elicitations_handler.cancel_elicitation(elicitation_id)
if not result["success"]:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=result["error"]
)
return ElicitationCancelResponse(**result)
except HTTPException:
raise
except Exception as e:
logger.error(
"failed_to_cancel_elicitation",
elicitation_id=elicitation_id,
error=str(e),
error_type=type(e).__name__
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to cancel elicitation: {str(e)}"
)
@app.post("/elicitations/cleanup", response_model=Dict[str, Any])
async def cleanup_expired_elicitations():
"""
Clean up expired elicitation sessions.
This endpoint is typically called by a background job
to clean up expired sessions and free up resources.
"""
try:
logger.info("elicitation_cleanup_requested")
result = await elicitations_handler.cleanup_expired_sessions()
logger.info(
"elicitation_cleanup_completed",
cleaned_count=result["cleaned_count"],
remaining_sessions=result["remaining_sessions"]
)
return result
except Exception as e:
logger.error(
"failed_to_cleanup_elicitations",
error=str(e),
error_type=type(e).__name__
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to cleanup elicitations: {str(e)}"
)
@app.get("/elicitations/stats", response_model=SystemStatsResponse)
async def get_elicitation_stats():
"""
Get elicitation system statistics.
This endpoint provides statistics about the elicitation
system for monitoring and debugging.
"""
try:
logger.info("elicitation_stats_requested")
result = elicitations_handler.get_system_stats()
return SystemStatsResponse(**result)
except Exception as e:
logger.error(
"failed_to_get_elicitation_stats",
error=str(e),
error_type=type(e).__name__
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to get elicitation stats: {str(e)}"
)
logger.info("elicitation_endpoints_registered")
