Streaming & Async Patterns

Why streaming?

Some API calls return immediately (creating a project takes milliseconds). Others — like asking the AI a complex question — can take several seconds. Streaming lets you show results as they’re generated instead of waiting for the complete response.

Chat streaming

The Chat endpoint (POST /api/v2/chat) supports Server-Sent Events (SSE) for real-time streaming. Set stream: true in your request:

curl -X POST https://api.pyramid-ai.com/api/v2/chat \
  -H "Authorization: Bearer pai_live_YOUR_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "project_id": "...",
    "messages": [{"role": "user", "content": "What are the safety requirements?"}],
    "stream": true
  }'

The response arrives as a stream of events:

data: {"type": "text", "content": "Based on"}
data: {"type": "text", "content": " the project"}
data: {"type": "text", "content": " safety plan,"}
data: {"type": "text", "content": " all workers"}
...
data: {"type": "done", "sources": [...], "usage": {...}}

Each event contains a small piece of the response. Your UI can display each piece as it arrives — creating a typing effect.

When to use streaming

Handling SSE in JavaScript

const response = await fetch('/api/v2/chat', {
  method: 'POST',
  headers: {
    'Authorization': 'Bearer pai_live_YOUR_KEY',
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    project_id: '...',
    messages: [{ role: 'user', content: 'What are the safety requirements?' }],
    stream: true,
  }),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  const text = decoder.decode(value);
  // Parse SSE events and update UI
  console.log(text);
}

Asynchronous operations

Some operations run in the background because they take too long for a synchronous response:

Polling pattern

For background operations, poll the status endpoint until the job completes:

import time
import requests
def wait_for_completion(url, headers, max_wait=300):
    """Poll until status is terminal (completed/failed)."""
    start = time.time()
    while time.time() - start < max_wait:
        response = requests.get(url, headers=headers)
        data = response.json()["data"]
        if data["status"] in ("completed", "failed", "partial"):
            return data
        time.sleep(5)  # Check every 5 seconds
    raise TimeoutError("Operation did not complete in time")

Recommended polling intervals

Synchronous vs asynchronous at a glance