Streaming with SSE

How to read the event stream from /api/query/stream in your own code, in Node, Python, a browser or curl, with the parsing details and the few pitfalls that catch most people the first time.

01What SSE is

Server-Sent Events is a small format for one-way streaming over HTTP. The server keeps one response open and writes events to it as text; the client reads them as they arrive. Each event is a few lines, event: with its type and data: with its content, ended by a blank line.

It suits a research run: the client sends one question and then only listens, while the run reports each phase, each source it finds and reads, its reasoning, and finally the answer.

02The events to handle

Every data: line is one JSON object. The full payloads are in API · Streaming; these are the ones most clients act on.

Decide success on final.status, not on the HTTP status: a bad or revoked key is reported inside an HTTP 200 stream. Ignore event types and fields you don't recognise.

03Browser

Don't call ResearchAnything from the browser with your key: anyone who loads the page can read it. Point the browser at your own backend, which relays the stream (see Pattern C).

The browser's EventSource only makes GET requests and can't send a body, so read a fetch() response yourself:

async function streamQuery(query, onEvent) {
  // Your own backend's relay endpoint, which holds the key.
  const res = await fetch("/stream", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ query }),
  });

  const reader = res.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });

    // Events end with a blank line. The last piece may be incomplete.
    const blocks = buffer.split("\n\n");
    buffer = blocks.pop();

    for (const block of blocks) {
      let event = "message";
      let data = "";
      for (const line of block.split("\n")) {
        if (line.startsWith("event:")) event = line.slice(6).trim();
        else if (line.startsWith("data:")) data += line.slice(5).trim();
      }
      if (data) onEvent(event, JSON.parse(data));
    }
  }
}

const sources = new Map();
streamQuery("Compare lithium-ion and sodium-ion batteries for grid storage.", (event, data) => {
  if (event === "step") showActivity(data.label);
  if (event === "source" && data.stage === "discovered") sources.set(data.id, data);
  if (event === "reasoning" && data.stage === "chunk") appendReasoning(data.step_id, data.chunk);
  if (event === "final") data.status === "ok" ? renderAnswer(data.answer, data.sources) : showErrors(data.errors);
});

04Node

On Node 18 and later, fetch is built in and the response body can be read with for await. On the server you can send the key directly.

async function* streamQuery(query) {
  const res = await fetch("https://researchanything.ai/api/query/stream", {
    method: "POST",
    headers: {
      "Authorization": `Bearer ${process.env.ESSARION_KEY}`,
      "Content-Type": "application/json",
      "Accept": "text/event-stream",
    },
    body: JSON.stringify({ query }),
  });
  if (res.status === 422) throw new Error(JSON.stringify(await res.json()));

  const decoder = new TextDecoder();
  let buffer = "";
  for await (const bytes of res.body) {
    buffer += decoder.decode(bytes, { stream: true });
    const blocks = buffer.split("\n\n");
    buffer = blocks.pop();
    for (const block of blocks) {
      let event = "message";
      let data = "";
      for (const line of block.split("\n")) {
        if (line.startsWith("event:")) event = line.slice(6).trim();
        else if (line.startsWith("data:")) data += line.slice(5).trim();
      }
      if (data) yield { event, data: JSON.parse(data) };
    }
  }
}

for await (const { event, data } of streamQuery("Compare lithium-ion and sodium-ion batteries for grid storage.")) {
  if (event === "step") console.log("·", data.label);
  if (event === "final") {
    console.log(data.status === "ok" ? data.answer : data.errors);
  }
}

05Python

httpx streams a response and splits it into lines for you; you only have to group lines into events.

import asyncio
import json
import os

import httpx

async def stream_query(query):
    headers = {
        "Authorization": f"Bearer {os.environ['ESSARION_KEY']}",
        "Accept": "text/event-stream",
    }
    timeout = httpx.Timeout(10.0, read=None)
    async with httpx.AsyncClient(timeout=timeout) as client:
        async with client.stream(
            "POST",
            "https://researchanything.ai/api/query/stream",
            headers=headers,
            json={"query": query},
        ) as resp:
            event, data = "message", []
            async for line in resp.aiter_lines():
                if line.startswith("event:"):
                    event = line[6:].strip()
                elif line.startswith("data:"):
                    data.append(line[5:].strip())
                elif line == "" and data:
                    yield event, json.loads("".join(data))
                    event, data = "message", []

async def main():
    async for event, data in stream_query("Compare lithium-ion and sodium-ion batteries for grid storage."):
        if event == "step":
            print("·", data["label"])
        elif event == "final":
            print(data["answer"] if data["status"] == "ok" else data["errors"])

asyncio.run(main())

06curl

For a quick look from a shell, -N turns off curl's buffering so events print as they land:

curl -N https://researchanything.ai/api/query/stream \
  -H "Authorization: Bearer $ESSARION_KEY" \
  -H "Content-Type: application/json" \
  -d '{"query": "Compare lithium-ion and sodium-ion batteries for grid storage."}'

To watch only one kind of event, filter the output, for example the phase labels:

curl -sN … | grep --line-buffered -A1 '^event: step'

07Disconnects

Connections drop: networks change, laptops sleep, proxies recycle. Plan for what that means here.

• A dropped stream stops the run. Closing the connection before final ends the run on ResearchAnything's side. The research run it used is not given back.
• There is no resume. A stream cannot be rejoined, and sending the same question again starts a new run, which uses another research run.
• Nothing to fetch afterwards. Your key cannot fetch a past run, so what you did not keep from the stream is gone for your code.

So: store request_id from start, keep the events you care about as they arrive, retry at most once after a drop, and then tell the user it failed.

08Common pitfalls

Proxy buffering

If you relay the stream through your own server and events arrive in large bursts instead of one at a time, something in between is buffering:

• Nginx: set proxy_buffering off on the streaming location, or send the X-Accel-Buffering: no response header.
• Express and Node: call res.flushHeaders() first and res.write() each chunk, rather than collecting the body and sending it at the end.
• Compression: turn off response compression on the streaming route; compressors often hold data back until they have enough to work with.

Request time limits

If your host has a short cap, move the relay to a host that allows long-lived requests, or run the call in a background worker (Pattern B) and let the browser poll your API.

Partial chunks

A single read can end in the middle of an event. Buffer the text, split on the blank line between events, and keep the last, possibly incomplete, piece for the next read, as the examples above do. Skip that and you will see occasional JSON.parse errors on cut-off data.

09Where to go next

• Relay the stream from your own server: Pattern C — streaming relay
• Every event and payload: API · Streaming
• What the Console keeps of a run: API · Run timelines