> ## Documentation Index > Fetch the complete documentation index at: https://docs.causeloop.ai/llms.txt > Use this file to discover all available pages before exploring further. # Real-time WebSocket > Subscribe to live issue, pattern, prediction, and job events from your workspace via the authenticated WebSocket stream. Causeloop exposes a real-time event stream over WebSocket. Connect once and receive pushed frames as things happen in your workspace — no polling required. ## Connection URL ``` wss://api.causeloop.ai/v1/stream?token=&workspace= ``` | Environment | URL | | ----------------- | ---------------------------------- | | Production | `wss://api.causeloop.ai/v1/stream` | | Local development | `ws://localhost:4000/v1/stream` | ## Authentication Browsers cannot set the `Authorization` header on a WebSocket handshake. Instead, pass your Causeloop Bearer JWT as a query parameter: ``` ?token= ``` Also pass the `workspace` parameter — the server validates it matches the workspace claim in the token and rejects cross-tenant subscriptions: ``` ?workspace= ``` Keep WebSocket URLs out of logs. The `token` query parameter is a full Bearer JWT. In production, construct the connection URL in memory and never expose it in URLs, analytics, or log pipelines. ## Connection lifecycle Connect to the WebSocket endpoint with `token` and `workspace` query params. Immediately after accepting, the server sends a `connected` frame confirming your workspace and user: ```json theme={null} { "type": "connected", "payload": { "workspace_id": "ws_01abc123", "user_id": "usr_01abc456" }, "ts": "2024-01-15T10:00:00.000Z" } ``` The server pushes domain event frames as they occur. See [Event types](#event-types) below. Every 5 seconds the server sends a heartbeat frame. Use this to detect stale connections: ```json theme={null} { "type": "heartbeat", "payload": { "ok": true }, "ts": "2024-01-15T10:00:05.000Z" } ``` ## Frame format All frames share a common envelope: ```json theme={null} { "type": "", "payload": { ... }, "ts": "2024-01-15T10:00:00.000Z" } ``` The event type identifier (see table below). Event-specific data. Shape varies by event type. UTC timestamp of when the event was emitted by the server. ## Event types | Type | Trigger | | ------------------------------- | ------------------------------------------------------------------ | | `connected` | Sent once immediately after the connection is accepted. | | `heartbeat` | Sent every 5 seconds to keep the connection alive. | | `issue.created` | A new issue was ingested from a connector or created via API. | | `issue.status.changed` | An issue's status changed (e.g. `new` → `analyzing` → `resolved`). | | `pattern.detected` | A new pattern was identified or an existing pattern updated. | | `prediction.alert.fired` | A prediction alert rule triggered. | | `recommendation.status.changed` | A recommendation was accepted, implemented, or dismissed. | | `notification.created` | A workspace notification was created for the connected user. | | `job.updated` | A background report or export job changed status or progress. | ### issue.created payload ```json theme={null} { "id": "iss_01abc123", "title": "Payment gateway timeout on checkout", "status": "new", "severity": "high", "risk_score": 74, "source": "pagerduty", "created_at": "2024-01-15T10:00:00.000Z" } ``` ### issue.status.changed payload ```json theme={null} { "id": "iss_01abc123", "previous_status": "new", "status": "analyzing", "changed_at": "2024-01-15T10:00:05.000Z" } ``` ### pattern.detected payload ```json theme={null} { "id": "pat_01xyz789", "name": "Payment gateway timeout under load", "risk_score": 82, "status": "emerging", "issue_count": 7, "detected_at": "2024-01-15T10:00:10.000Z" } ``` ### job.updated payload ```json theme={null} { "job_id": "rpt_job_01abc789", "status": "running", "progress": 0.65, "result_url": null, "error": null, "created_at": "2024-01-15T09:55:00.000Z", "finished_at": null } ``` When `status` is `"succeeded"`, `result_url` is populated with the download path. ## Close codes | Code | Meaning | | ------ | -------------------------------------------------------------------------------------- | | `4401` | Missing, invalid, or expired token. Re-authenticate and reconnect. | | `4003` | Workspace mismatch — the `workspace` param does not match the token's workspace claim. | | `1001` | Server going away — graceful drain (e.g. deploy). Reconnect immediately. | | `1011` | Unexpected server error. Reconnect with backoff. | ## Example client ```javascript JavaScript theme={null} function connectCauseloopStream({ accessToken, workspaceId, onEvent }) { const url = new URL('wss://api.causeloop.ai/v1/stream'); url.searchParams.set('token', accessToken); url.searchParams.set('workspace', workspaceId); const ws = new WebSocket(url.toString()); let reconnectDelay = 1000; ws.addEventListener('open', () => { console.log('[WS] connected'); reconnectDelay = 1000; // reset backoff on successful open }); ws.addEventListener('message', (event) => { const frame = JSON.parse(event.data); if (frame.type === 'heartbeat') return; // internal keepalive, ignore if (frame.type === 'connected') { console.log('[WS] authenticated as workspace', frame.payload.workspace_id); return; } onEvent(frame); }); ws.addEventListener('close', ({ code, reason }) => { console.warn(`[WS] closed: ${code} ${reason}`); if (code === 4401) { // Token expired — refresh before reconnecting console.error('[WS] invalid token, re-authenticate before reconnecting'); return; } if (code === 4003) { console.error('[WS] workspace mismatch, check your workspace_id'); return; } // Reconnect with exponential backoff for all other close codes const delay = reconnectDelay + Math.random() * 500; console.log(`[WS] reconnecting in ${Math.round(delay)}ms`); setTimeout(() => connectCauseloopStream({ accessToken, workspaceId, onEvent }), delay); reconnectDelay = Math.min(reconnectDelay * 2, 30_000); }); ws.addEventListener('error', (err) => { console.error('[WS] error', err); }); return ws; } // Usage const ws = connectCauseloopStream({ accessToken: '', workspaceId: 'ws_01abc123', onEvent: (frame) => { switch (frame.type) { case 'issue.created': console.log('New issue:', frame.payload.title); break; case 'pattern.detected': console.log('Pattern:', frame.payload.name, '— risk:', frame.payload.risk_score); break; case 'job.updated': console.log('Job progress:', frame.payload.progress, frame.payload.status); break; } }, }); ``` ```python Python theme={null} import asyncio import json import websockets async def stream_events(access_token: str, workspace_id: str): url = ( f"wss://api.causeloop.ai/v1/stream" f"?token={access_token}&workspace={workspace_id}" ) reconnect_delay = 1.0 while True: try: async with websockets.connect(url) as ws: print('[WS] connected') reconnect_delay = 1.0 async for message in ws: frame = json.loads(message) event_type = frame.get('type') if event_type in ('heartbeat', 'connected'): continue # Handle domain events print(f"[{event_type}]", json.dumps(frame['payload'], indent=2)) except websockets.exceptions.ConnectionClosedError as e: if e.code in (4401, 4003): print(f'[WS] fatal close {e.code} — not reconnecting') break print(f'[WS] closed ({e.code}), reconnecting in {reconnect_delay:.1f}s') except Exception as e: print(f'[WS] error: {e}, reconnecting in {reconnect_delay:.1f}s') await asyncio.sleep(reconnect_delay) reconnect_delay = min(reconnect_delay * 2, 30) asyncio.run(stream_events('', 'ws_01abc123')) ``` ## Event bus integration Server-side routers can push frames directly into the stream using the `publish` function (for workspace-wide events) or `publish_user` (for user-scoped events). This allows any API mutation — creating an issue, updating a recommendation — to fan out a real-time event to all connected clients in that workspace without polling. WebSocket messages do not count against the REST [rate limit](/api-reference/rate-limits). For real-time use cases such as live dashboards, notification badges, and job progress bars, prefer the WebSocket stream over polling REST endpoints.