Streaming (SSE)

Streaming (SSE)

Use POST /v1/query/stream for real-time responses via Server-Sent Events.

The stream emits typed events in order:

EventPayloadWhen
thinking{stage, message}Progress updates during pipeline stages
sql{sql}Generated SQL (before execution)
token{token}One text token of the AI answer
completeFull query resultAll stages finished
error{message}Pipeline failure
done{}Stream closed

Example Stream

event: thinking
data: {"stage": "schema_lookup", "message": "Loading relevant schema..."}

event: thinking
data: {"stage": "sql_generation", "message": "Generating SQL query..."}

event: sql
data: {"sql": "SELECT p.name, SUM(oi.unit_price * oi.qty) AS revenue FROM ..."}

event: thinking
data: {"stage": "executing", "message": "Running query on your database..."}

event: token
data: {"token": "The top"}

event: token
data: {"token": " 5 products"}

event: token
data: {"token": " by revenue last month were"}

event: complete
data: {"id": "qry_...", "answer": "...", "data": {...}, "chart": {...}}

event: done
data: {}

JavaScript Example

async function* streamQuery(question: string, sourceId: string) {
  const response = await fetch("/v1/query/stream", {
    method: "POST",
    headers: {
      Authorization: `Bearer ${API_KEY}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ source_id: sourceId, question }),
  });
 
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
 
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
 
    const lines = buffer.split("\n");
    buffer = lines.pop()!;
 
    let eventType = "";
    for (const line of lines) {
      if (line.startsWith("event: ")) {
        eventType = line.slice(7).trim();
      } else if (line.startsWith("data: ")) {
        const data = JSON.parse(line.slice(6));
        yield { type: eventType, data };
      }
    }
  }
}
 
// Usage
for await (const event of streamQuery("Top 5 products by revenue", "src_abc123")) {
  if (event.type === "token") {
    process.stdout.write(event.data.token);
  } else if (event.type === "sql") {
    console.log("\nGenerated SQL:", event.data.sql);
  } else if (event.type === "complete") {
    console.log("\nChart data:", event.data.chart);
  }
}

Python Example

import httpx
 
with httpx.stream(
    "POST",
    "https://api.retailmind.dev/v1/query/stream",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={"source_id": "src_abc123", "question": "Top 5 products by revenue"},
    timeout=60,
) as response:
    event_type = ""
    for line in response.iter_lines():
        if line.startswith("event: "):
            event_type = line[7:].strip()
        elif line.startswith("data: "):
            import json
            data = json.loads(line[6:])
            if event_type == "token":
                print(data["token"], end="", flush=True)
            elif event_type == "complete":
                print(f"\n\nChart: {data['chart']}")

The thinking events are perfect for showing a progress indicator in your UI while the pipeline runs.

Query APIInsights & Anomalies