Real-time WebSocket

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=<jwt>&workspace=<workspace_id>

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=<access_token>

Also pass the workspace parameter — the server validates it matches the workspace claim in the token and rejects cross-tenant subscriptions:

?workspace=<workspace_id>

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

Open the connection

Connect to the WebSocket endpoint with token and workspace query params.

Receive the connected frame

Immediately after accepting, the server sends a connected frame confirming your workspace and user:

{
  "type": "connected",
  "payload": {
    "workspace_id": "ws_01abc123",
    "user_id": "usr_01abc456"
  },
  "ts": "2024-01-15T10:00:00.000Z"
}

Receive event frames

The server pushes domain event frames as they occur. See Event types below.

Heartbeat

Every 5 seconds the server sends a heartbeat frame. Use this to detect stale connections:

{
  "type": "heartbeat",
  "payload": { "ok": true },
  "ts": "2024-01-15T10:00:05.000Z"
}

Frame format

All frames share a common envelope:

{
  "type": "<event_type>",
  "payload": { ... },
  "ts": "2024-01-15T10:00:00.000Z"
}

string

required

The event type identifier (see table below).

object

required

Event-specific data. Shape varies by event type.

string (ISO 8601)

required

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

{
  "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

{
  "id": "iss_01abc123",
  "previous_status": "new",
  "status": "analyzing",
  "changed_at": "2024-01-15T10:00:05.000Z"
}

pattern.detected payload

{
  "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

{
  "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

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: '<your_access_token>',
  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;
    }
  },
});

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('<access_token>', '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. For real-time use cases such as live dashboards, notification badges, and job progress bars, prefer the WebSocket stream over polling REST endpoints.

Errors Get Jwks

​Connection URL

​Authentication

​Connection lifecycle

​Frame format

​Event types

​issue.created payload

​issue.status.changed payload

​pattern.detected payload

​job.updated payload

​Close codes

​Example client

​Event bus integration

Connection URL

Authentication

Connection lifecycle

Frame format

Event types

issue.created payload

issue.status.changed payload

pattern.detected payload

job.updated payload

Close codes

Example client

Event bus integration