import logging import asyncio from contextlib import asynccontextmanager from typing import Optional, Dict, Any from fastapi import FastAPI from pydantic import BaseModel import uvicorn # from app.services.mcp_service import AgrotaMcpClient # Configuración de logs logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger("mcp-webhook-server") # Variables globales para el cliente MCP mcp_client = None mcp_client_ready = False import pybreaker # Circuit Breaker Configuration circuit_breaker = pybreaker.CircuitBreaker( fail_max=3, reset_timeout=60, exclude=[ValueError], # Exclude non-connection errors if needed ) @asynccontextmanager async def lifespan(app: FastAPI): # Inicio: Conectar cliente MCP global mcp_client, mcp_client_ready logger.info("🚀 Iniciando servidor y conectando Cliente MCP...") # URL del MCP Server (SSE endpoint) from app.services.mcp_service import AgrotaMcpClient import traceback mcp_client = AgrotaMcpClient("http://localhost:8088/sse") # Run connection in background to avoid blocking server startup async def connect_mcp(): global mcp_client_ready try: logger.info("⏳ Intentando conectar con MCP Server...") # Use circuit breaker for connection await circuit_breaker.call(mcp_client.connect) if mcp_client.is_connected(): logger.info("✅ Cliente MCP conectado e inicializado!") mcp_client_ready = True else: logger.warning("⚠️ Cliente MCP iniciado pero no conectado") except pybreaker.CircuitBreakerError: logger.error("❌ Circuit Breaker OPEN: Skipping MCP connection attempt.") except BaseException as e: logger.error(f"❌ Error CRÍTICO al iniciar Cliente MCP: {e}") logger.error(traceback.format_exc()) asyncio.create_task(connect_mcp()) yield # Cierre: Desconectar cliente MCP if mcp_client: logger.info("🛑 Desconectando Cliente MCP...") try: await mcp_client.disconnect() except Exception as e: logger.error(f"Error cerrando cliente MCP: {e}") app = FastAPI(lifespan=lifespan) # Modelos de solicitud/respuesta (simplificados para lo que envía BackAgrota) class WhatsAppText(BaseModel): body: str class WhatsAppMessage(BaseModel): type: str = "text" text: Optional[WhatsAppText] = None class WebhookPayload(BaseModel): messaging_product: str = "whatsapp" recipient_type: str = "individual" to: str type: str = "text" text: Optional[WhatsAppText] = None @app.post("/webhook") async def process_message(payload: Dict[str, Any]): logger.info(f"📩 Payload recibido: {payload}") global mcp_client, mcp_client_ready # Wait for initial connection to complete (up to 10 seconds) if not mcp_client_ready: logger.info("⏳ Esperando a que el cliente MCP esté listo...") for i in range(20): # 20 attemps * 0.5s = 10 seconds max if mcp_client_ready: break await asyncio.sleep(0.5) if not mcp_client_ready: logger.warning( "⚠️ Timeout esperando conexión inicial, intentando de todos modos..." ) # Check if client is initialized and connected if not mcp_client: from app.services.mcp_service import AgrotaMcpClient mcp_client = AgrotaMcpClient("http://localhost:8088/sse") if not mcp_client.is_connected(): logger.warning("⚠️ mcp_client not connected. Attempting reconnection...") try: await mcp_client.connect() except Exception as e: logger.error(f"❌ Reconnection failed: {e}") if not mcp_client.is_connected(): logger.error("❌ mcp_client is not ready.") return { "status": "error", "replies": [ { "type": "text", "text": { "body": "El servicio de IA no está disponible en este momento. La sesión no se ha podido iniciar." }, } ], } try: # Extraer el mensaje del usuario user_message = "" if "text" in payload and "body" in payload["text"]: user_message = payload["text"]["body"] else: # Fallback si la estructura varía user_message = str(payload) logger.info(f"📥 [MENSAJE RECIBIDO] Contenido: {user_message}") logger.debug(f"🔍 Payload completo: {payload}") # Llamar a la herramienta "consultarAgrota" # Add timeout to call_tool to prevent indefinite hangs logger.info("⏳ Llamando a consultarAgrota...") try: result = await circuit_breaker.call( lambda: asyncio.wait_for( mcp_client.call_tool("consultarAgrota", {"prompt": user_message}), timeout=120.0, ) ) except pybreaker.CircuitBreakerError: logger.error("❌ Circuit Breaker OPEN: Preventing tool call.") return { "status": "error", "replies": [ { "messaging_product": "whatsapp", "recipient_type": "individual", "to": payload.get("to", ""), "type": "text", "text": { "body": "El servicio de IA no está disponible temporalmente (Circuit Breaker Open)." }, } ], } reply_text = "No response from tool" if result: reply_text = str(result) logger.info(f"📤 [RESPUESTA ENVIADA] Contenido: {reply_text}") # Extraer el número de destinatario del payload original recipient = payload.get("to", "") # Construir respuesta formato BackAgrota (MessageResponseDto) response = { "status": "success", "replies": [ { "messaging_product": "whatsapp", "recipient_type": "individual", "to": recipient, "type": "text", "text": {"body": reply_text}, } ], } return response except asyncio.TimeoutError: logger.error( "❌ Timeout waiting for MCP tool response (120s). Possible SSE connection issue." ) return { "status": "error", "replies": [ { "messaging_product": "whatsapp", "recipient_type": "individual", "to": payload.get("to", ""), "type": "text", "text": { "body": "La consulta está tardando demasiado. El servidor MCP puede tener problemas de conexión SSE." }, } ], } except Exception as e: logger.error(f"❌ Error procesando mensaje: {e}") # If the error suggests session issues, disconnect so next request retries if "session" in str(e).lower() or "not started" in str(e).lower(): await mcp_client.disconnect() return { "status": "error", "replies": [ { "messaging_product": "whatsapp", "recipient_type": "individual", "to": payload.get("to", ""), "type": "text", "text": {"body": f"Error interno en MCP Client: {str(e)}"}, } ], } if __name__ == "__main__": # Correr en puerto 8000 uvicorn.run(app, host="0.0.0.0", port=8086)