Spaces:
Running
Running
| import asyncio | |
| import logging | |
| from datetime import datetime, timezone | |
| from typing import Any, Dict, List, Optional | |
| from pydantic import BaseModel, Field | |
| from core.integration_registry import IntegrationRegistry | |
| from core.circuit_breaker import circuit_breaker | |
| from core.universal_communication_bridge import UniversalCommunicationBridge | |
| from sqlalchemy.orm import Session | |
| logger = logging.getLogger(__name__) | |
| class UnifiedIncomingMessage(BaseModel): | |
| """Standardized incoming message from any communication platform""" | |
| platform: str | |
| sender_id: str | |
| recipient_id: str | |
| text: str | |
| timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) | |
| thread_id: Optional[str] = None | |
| metadata: Dict[str, Any] = Field(default_factory=dict) | |
| raw_payload: Dict[str, Any] = Field(default_factory=dict) | |
| class WebhookBridge: | |
| """ | |
| Standardized bridge for incoming webhooks dispatching events | |
| to the IntegrationRegistry and ChatOrchestrator. | |
| """ | |
| def __init__(self): | |
| self._orchestrator = None | |
| def _get_orchestrator(self): | |
| """Lazy logic to handle circular dependencies.""" | |
| if self._orchestrator is None: | |
| try: | |
| from integrations.chat_orchestrator import ChatOrchestrator | |
| self._orchestrator = ChatOrchestrator(workspace_id="default") | |
| except Exception as e: | |
| logger.error(f"Failed to initialize ChatOrchestrator: {e}") | |
| return self._orchestrator | |
| async def process_event( | |
| self, | |
| platform: str, | |
| tenant_id: str, | |
| data: Dict[str, Any], | |
| registry: IntegrationRegistry, | |
| db: Session | |
| ) -> Dict[str, Any]: | |
| """Process an incoming platform event via the UniversalCommunicationBridge.""" | |
| logger.info(f"Webhook Bridge: Dispatching event from {platform} for tenant {tenant_id}") | |
| # 0. Circuit Breaker Check | |
| cb_key = f"{platform}:{tenant_id}" | |
| if not await circuit_breaker.is_enabled(cb_key): | |
| logger.warning(f"Webhook Bridge: Circuit breaker OPEN for {cb_key}. Ignoring event.") | |
| return {"status": "ignored", "reason": "circuit_breaker_open"} | |
| try: | |
| # 1. Use UniversalCommunicationBridge for standardized normalization | |
| ucb = UniversalCommunicationBridge(db) | |
| ucb_result = await ucb.receive_message( | |
| tenant_id=tenant_id, | |
| platform=platform, | |
| payload=data | |
| ) | |
| if not ucb_result: | |
| return {"status": "ignored", "reason": "ucb_ignored_or_error"} | |
| # Handle standardized interactions (buttons, etc.) | |
| if ucb_result.get("type") == "interaction": | |
| return { | |
| "status": "success", | |
| "processed": True, | |
| "type": "interaction", | |
| "result": ucb_result.get("result") | |
| } | |
| # Handle standard text messages | |
| if ucb_result.get("type") != "message": | |
| return {"status": "ignored", "reason": "unsupported_ucb_type"} | |
| unified_msg = ucb_result["message"] | |
| text_content = unified_msg.content | |
| sender_id = unified_msg.sender_id | |
| # 2. Command Handling (e.g., /run) | |
| if text_content.startswith('/'): | |
| # Convert to local model for compatibility with _handle_command | |
| compat_msg = UnifiedIncomingMessage( | |
| platform=platform, | |
| sender_id=sender_id, | |
| recipient_id=unified_msg.recipient_id or "bot", | |
| text=text_content, | |
| thread_id=unified_msg.thread_id, | |
| metadata=unified_msg.metadata or {} | |
| ) | |
| return await self._handle_command(compat_msg, tenant_id, registry) | |
| # 3. Chat Orchestrator Integration | |
| orchestrator = self._get_orchestrator() | |
| if not orchestrator: | |
| return {"status": "error", "message": "ChatOrchestrator unavailable"} | |
| session_id = f"{platform}_{sender_id}" | |
| response = await orchestrator.process_chat_message( | |
| message=text_content, | |
| session_id=session_id, | |
| user_id=f"ext_{sender_id}", | |
| context={ | |
| "platform": platform, | |
| "tenant_id": tenant_id, | |
| "sender_id": sender_id, | |
| "recipient_id": unified_msg.recipient_id, | |
| "thread_id": unified_msg.thread_id | |
| } | |
| ) | |
| # 4. Auto-Response Dispatch (Optional based on response) | |
| if response and response.get("message"): | |
| # Use UCB for standard response | |
| ucb = UniversalCommunicationBridge(db) | |
| await ucb.send_message( | |
| tenant_id=tenant_id, | |
| platform=platform, | |
| target_id=sender_id, | |
| content=response["message"], | |
| metadata={"thread_ts": unified_msg.thread_id} | |
| ) | |
| return { | |
| "status": "success", | |
| "processed": True, | |
| "orchestrator_response": response | |
| } | |
| except Exception as e: | |
| logger.error(f"Webhook Bridge Error ({platform}): {e}") | |
| return {"status": "error", "message": str(e)} | |
| async def _handle_command(self, msg: UnifiedIncomingMessage, tenant_id: str, registry: IntegrationRegistry) -> Dict[str, Any]: | |
| """Handle platform commands (e.g., /run) via Registry.""" | |
| parts = msg.text[1:].split(' ', 2) | |
| command = parts[0].lower() | |
| if command == "run" and len(parts) > 1: | |
| agent_name = parts[1] | |
| task_input = parts[2] if len(parts) > 2 else "Run Default" | |
| # Use registry to trigger agent task (simulated execution) | |
| # In production, we'd use core.agent_routes.execute_agent_task | |
| return {"status": "command_triggered", "command": "run", "agent": agent_name} | |
| return {"status": "command_ignored", "command": command} | |
| # Global instance | |
| webhook_bridge = WebhookBridge() | |