From 111728915e387f939efec0160e597406fda6c782 Mon Sep 17 00:00:00 2001 From: Ishaan Jaffer Date: Thu, 7 May 2026 10:06:55 -0700 Subject: [PATCH] feat(v2 managed agents): add GET /v2/sessions/:id/events SSE handler --- litellm/managed_agents/endpoints/events.py | 194 +++++++++++++++++++++ 1 file changed, 194 insertions(+) create mode 100644 litellm/managed_agents/endpoints/events.py diff --git a/litellm/managed_agents/endpoints/events.py b/litellm/managed_agents/endpoints/events.py new file mode 100644 index 00000000000..59af407a4b9 --- /dev/null +++ b/litellm/managed_agents/endpoints/events.py @@ -0,0 +1,194 @@ +"""SSE events endpoint for managed agents v2 — `GET /v2/sessions/:id/events`. + +Streams normalized events for a session. The adapter yields +`(event_type, data_dict)` tuples — this handler formats them as SSE +frames per contract §6.6: + + event: + data: + \\n\\n + +The first yielded event is always `connected` (synthesized by the +adapter). Subsequent events are translated from the underlying provider +bus per the §7 event-mapping table. + +Pre-forward checks (per contract §7): +1. `Authorization` valid → handled by `user_api_key_auth` dependency. +2. Session row exists for `created_by = caller` → else 404. +3. `session.status == "ready"` → else 503 (`provisioning`) or 404 + (`terminated`/`error`). +4. Resolve `sandbox_url` + `sandbox_metadata.opencode_session_id`. + +Once the SSE stream has started we cannot change the HTTP status code, +so a `SandboxUnreachableError` raised mid-stream is converted to an +`error` SSE event and the stream closes cleanly. +""" + +import json +from typing import Any, AsyncIterator, Dict, Tuple + +from fastapi import APIRouter, Depends, HTTPException +from fastapi.responses import StreamingResponse + +from litellm.managed_agents.adapters.base import ( + SandboxAdapter, + SandboxUnreachableError, +) +from litellm.managed_agents.adapters.registry import get_adapter +from litellm.managed_agents.db import get_session +from litellm.proxy._types import UserAPIKeyAuth +from litellm.proxy.auth.user_api_key_auth import user_api_key_auth + +router = APIRouter() + +# SSE response headers per contract §6.6. +SSE_HEADERS = { + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", +} + + +# --------------------------------------------------------------------------- +# Pre-forward helper +# --------------------------------------------------------------------------- +# TODO: extract to shared helper — `litellm/managed_agents/endpoints/_helpers.py` +# once the messages endpoint lands. Both endpoints need the same logic. + + +async def _load_ready_session( + session_id: str, + user_api_key_dict: UserAPIKeyAuth, +) -> Dict[str, Any]: + """Load a session row, scope to caller, and validate it is ``ready``. + + Returns the session row dict. + + Raises: + HTTPException(404) if the session does not exist for this caller, + or if the session is in a terminal state (``terminated``/``error``). + HTTPException(503) if the session is still provisioning. + HTTPException(500) if the prisma client is not initialized. + """ + # Lazy import to avoid module-load circular import with proxy_server. + from litellm.proxy.proxy_server import prisma_client + + if prisma_client is None: + raise HTTPException(status_code=500, detail="Database not initialized") + + session_row = await get_session( + prisma_client, + session_id=session_id, + created_by=user_api_key_dict.user_id, + ) + if not session_row: + raise HTTPException(status_code=404, detail=f"Session {session_id} not found") + + status = session_row.get("status") + if status == "provisioning": + raise HTTPException( + status_code=503, + detail="Session not ready", + headers={"Retry-After": "5"}, + ) + if status != "ready": + # terminated, error, or any other non-ready terminal state → + # treat as not-found per contract §7 failure modes. + raise HTTPException(status_code=404, detail=f"Session {session_id} not found") + + return session_row + + +def _resolve_adapter_handle( + session_row: Dict[str, Any], +) -> Tuple[SandboxAdapter, str, str]: + """Resolve adapter + sandbox_url + opencode_session_id from a session row. + + Raises HTTPException(500) if any required field is missing — this + indicates an internally-corrupted row and should never happen for a + session in ``ready`` status. + """ + sandbox_type = session_row.get("sandbox_type") + sandbox_url = session_row.get("sandbox_url") + sandbox_metadata = session_row.get("sandbox_metadata") or {} + if isinstance(sandbox_metadata, str): + # Defensive: some Prisma clients return JSON columns as strings. + try: + sandbox_metadata = json.loads(sandbox_metadata) + except (TypeError, ValueError): + sandbox_metadata = {} + opencode_session_id = ( + sandbox_metadata.get("opencode_session_id") + if isinstance(sandbox_metadata, dict) + else None + ) + + if not sandbox_type or not sandbox_url or not opencode_session_id: + raise HTTPException( + status_code=500, detail="Session row is missing sandbox state" + ) + + try: + adapter = get_adapter(sandbox_type) + except ValueError as e: + raise HTTPException(status_code=422, detail=str(e)) + + return adapter, sandbox_url, opencode_session_id + + +# --------------------------------------------------------------------------- +# SSE formatter +# --------------------------------------------------------------------------- + + +async def _format_sse( + adapter: SandboxAdapter, + sandbox_url: str, + oc_sid: str, + our_session_id: str, +) -> AsyncIterator[bytes]: + """Format adapter event tuples as SSE frames. + + Once the stream has started we can't switch HTTP status codes, so a + `SandboxUnreachableError` becomes an inline `error` event and the + stream closes. Client disconnects are handled by Starlette: when the + consumer cancels the iterator, the underlying adapter's HTTP stream + is aborted via the async context manager. + """ + try: + async for event_type, data in adapter.stream_events( + sandbox_url, oc_sid, our_session_id + ): + yield (f"event: {event_type}\ndata: {json.dumps(data)}\n\n").encode() + except SandboxUnreachableError: + error_payload = json.dumps({"error": "Sandbox unreachable"}) + yield f"event: error\ndata: {error_payload}\n\n".encode() + + +# --------------------------------------------------------------------------- +# Route +# --------------------------------------------------------------------------- + + +@router.get("/v2/sessions/{session_id}/events") +async def stream_events( + session_id: str, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +) -> StreamingResponse: + """SSE stream of normalized events for a session. + + See contract §6.6 — first event is always ``connected`` (synthesized + by the adapter), followed by a stream of ``message.*`` events + translated from the upstream sandbox bus. Pre-forward checks happen + before streaming starts so 404/503 responses use the correct status + code; once streaming begins, errors are delivered as inline ``error`` + events. + """ + session_row = await _load_ready_session(session_id, user_api_key_dict) + adapter, sandbox_url, oc_sid = _resolve_adapter_handle(session_row) + + return StreamingResponse( + _format_sse(adapter, sandbox_url, oc_sid, session_id), + media_type="text/event-stream", + headers=SSE_HEADERS, + )