tiphys.server

361def create_router() -> APIRouter:
362    """Create and return the API router."""
363    return router

 28class EventBus:
 29    """
 30    WebSocket-based event bus with typed events.
 31
 32    Manages client connections and broadcasts events to all connected clients.
 33    Also supports local event handlers for internal subscriptions.
 34    """
 35
 36    def __init__(self) -> None:
 37        self._clients: dict[str, WebSocket] = {}
 38        self._handlers: dict[str, list[EventHandler]] = {}
 39        self._lock = asyncio.Lock()
 40
 41    @property
 42    def client_count(self) -> int:
 43        """Number of connected clients."""
 44        return len(self._clients)
 45
 46    async def connect(self, client_id: str, websocket: WebSocket) -> None:
 47        """Register a new client connection."""
 48        await websocket.accept()
 49        async with self._lock:
 50            self._clients[client_id] = websocket
 51        await logger.ainfo("Client connected", client_id=client_id, total_clients=self.client_count)
 52
 53    async def disconnect(self, client_id: str) -> None:
 54        """Remove a client connection."""
 55        async with self._lock:
 56            self._clients.pop(client_id, None)
 57        await logger.ainfo(
 58            "Client disconnected", client_id=client_id, total_clients=self.client_count
 59        )
 60
 61    async def broadcast(self, event: BaseEvent) -> None:
 62        """Send event to all connected clients and local handlers."""
 63        message = event.model_dump_json()
 64
 65        # Send to WebSocket clients
 66        async with self._lock:
 67            dead_clients: list[str] = []
 68            for client_id, ws in self._clients.items():
 69                try:
 70                    await ws.send_text(message)
 71                except Exception as e:
 72                    await logger.awarning(
 73                        "Failed to send to client", client_id=client_id, error=str(e)
 74                    )
 75                    dead_clients.append(client_id)
 76
 77            # Clean up dead connections
 78            for client_id in dead_clients:
 79                self._clients.pop(client_id, None)
 80
 81        # Trigger local handlers
 82        event_type = getattr(event, "type", event.__class__.__name__)
 83        handlers = self._handlers.get(event_type, [])
 84        for handler in handlers:
 85            try:
 86                await handler(event)
 87            except Exception as e:
 88                await logger.aerror("Handler error", event_type=event_type, error=str(e))
 89
 90    async def send_to(self, client_id: str, event: BaseEvent) -> bool:
 91        """Send event to a specific client. Returns True if sent successfully."""
 92        if client_id not in self._clients:
 93            return False
 94
 95        try:
 96            await self._clients[client_id].send_text(event.model_dump_json())
 97            return True
 98        except Exception as e:
 99            await logger.awarning("Failed to send to client", client_id=client_id, error=str(e))
100            async with self._lock:
101                self._clients.pop(client_id, None)
102            return False
103
104    def subscribe(self, event_type: str, handler: EventHandler) -> None:
105        """Register an event handler for a specific event type."""
106        self._handlers.setdefault(event_type, []).append(handler)
107
108    def unsubscribe(self, event_type: str, handler: EventHandler) -> None:
109        """Remove an event handler."""
110        if event_type in self._handlers:
111            with contextlib.suppress(ValueError):
112                self._handlers[event_type].remove(handler)
113
114    async def handle_websocket(self, client_id: str, websocket: WebSocket) -> None:
115        """
116        Handle a WebSocket connection lifecycle.
117
118        Call this from a WebSocket endpoint to manage the connection.
119        """
120        await self.connect(client_id, websocket)
121        try:
122            while True:
123                # Receive messages from client
124                data = await websocket.receive_text()
125                await self._handle_client_message(client_id, data)
126        except WebSocketDisconnect:
127            pass
128        finally:
129            await self.disconnect(client_id)
130
131    async def _handle_client_message(self, client_id: str, data: str) -> None:
132        """Handle incoming message from a client."""
133        await logger.adebug("Received from client", client_id=client_id, data=data[:100])
134
135        try:
136            message = json.loads(data)
137        except json.JSONDecodeError:
138            await logger.awarning("Invalid JSON from client", client_id=client_id)
139            return
140
141        msg_type = message.get("type", "message")
142
143        if msg_type == "message":
144            # Process user message through agent
145            content = message.get("content", "")
146            if content:
147                await self._process_user_message(client_id, content)
148        elif msg_type == "permission_response":
149            # Handle permission response
150            from tiphys.events import PermissionResponseEvent
151
152            await self.broadcast(
153                PermissionResponseEvent(
154                    request_id=message.get("request_id", ""),
155                    approved=message.get("approved", False),
156                )
157            )
158
159    async def _process_user_message(self, client_id: str, content: str) -> None:
160        """Process a user message through the agent."""
161        from tiphys.agent.runner import AgentRunner
162        from tiphys.config import load_config
163        from tiphys.tools import ToolRegistry, register_builtin_tools
164
165        config = load_config()
166        session_key = f"{config.agent_id}:client:{client_id}"
167
168        # Create tool registry and runner
169        registry = register_builtin_tools(ToolRegistry())
170        runner = AgentRunner(
171            config=config,
172            tool_registry=registry,
173            event_bus=self,
174        )
175
176        # Run agent and stream responses
177        try:
178            async for _block in runner.stream(
179                message=content,
180                session_key=session_key,
181            ):
182                pass  # Events are emitted to WebSocket via event_bus
183        except Exception as e:
184            await logger.aerror("Agent error", client_id=client_id, error=str(e))
185            from tiphys.events import ErrorEvent
186
187            await self.broadcast(
188                ErrorEvent(
189                    message=str(e),
190                    code=type(e).__name__,
191                )
192            )

199def get_event_bus() -> EventBus:
200    """
201    Get the global event bus instance.
202
203    .. deprecated::
204        Use AppContext.event_bus instead. This function will be removed
205        in a future version.
206
207    Returns:
208        The global EventBus instance.
209    """
210    import warnings
211
212    from tiphys.context import get_context, has_context
213
214    # If AppContext is available, use it (preferred path)
215    if has_context():
216        warnings.warn(
217            "get_event_bus() is deprecated. Use AppContext.event_bus instead.",
218            DeprecationWarning,
219            stacklevel=2,
220        )
221        return get_context().event_bus
222
223    # Fallback to global singleton for backwards compatibility
224    global _event_bus
225    if _event_bus is None:
226        _event_bus = EventBus()
227    return _event_bus