Streaming Responses

How the Bigdata.com Research Agent and Workflows stream results, what each message type means, and how to build a robust streaming handler.

The Research Agent (/v1/research-agent) and the Workflows execution endpoint (/v1/workflow/execute) return a single, long-lived HTTP response that streams a sequence of typed events using Server-Sent Events (SSE). A submitted workflow delivers the same events, either live from its event stream or replayed from the stored run. This page is the canonical reference for that stream: the event format, every public message type, the typical order of events in a request, and a complete handler you can copy into your codebase.

The SSE format

Each event is a single line beginning with data: , followed by a JSON document terminated by a newline:

data: {"chat_id": "abc123", "message": {"type": "THINKING", "content": "Analyzing..."}}\n
data: {"chat_id": "abc123", "message": {"type": "ANSWER", "content": "NVIDIA's"}}\n
data: {"chat_id": "abc123", "message": {"type": "ANSWER", "content": " gross margin"}}\n
...
data: {"chat_id": "abc123", "message": {"type": "COMPLETE", "consumption": [...]}}\n

Lines that do not start with data: are heartbeats or comments and should be ignored. The stream ends after a COMPLETE (or ERROR) event.

The Research Agent and Workflows use slightly different envelopes:

• Research Agent wraps the event in a message field: {"chat_id": "...", "message": {"type": "...", ...}}
• Workflows wraps it in a delta field: {"request_id": "...", "execution_id": "...", "delta": {"type": "...", ...}}

Examples in this guide use the Research Agent envelope. For Workflows, read the typed event from event["delta"] instead of event["message"].

Where the stream comes from

The same typed events reach you through several routes, and one handler serves all of them:

• Live, on the request that started the run — POST /v1/research-agent or POST /v1/workflow/execute.
• Live, attached to a submitted run — GET /v1/workflow/execute/async/{execution_id}/stream. Every event on this stream also carries an SSE id, so a dropped connection can resume from where it left off with the Last-Event-ID header. See Running a workflow.
• Replayed from a stored run — the events list in GET /v1/workflow/executions/{execution_id}, in the order they were emitted.

Write your dispatch on type once and reuse it for all three. The only routes that differ are the envelope (see the note above) and, when reading a stored run, that the events arrive as a JSON array rather than as SSE lines.

Message types

The public message types are listed below. The type field is the only field guaranteed to be present on every message and is the discriminator your handler should dispatch on.

Type	Purpose	Key fields
PLANNING	Research plan progress	plan.title, plan.steps[] with per-step status
THINKING	Internal reasoning text	content
ACTION	The agent is invoking a tool	tool_name, tool_arguments
AUDIT	Detailed trace of a tool’s execution	audit_traces[]
GROUNDING	Citations linking spans of the answer to sources	references[]
CHART	A chart the agent produced while running code	vega_lite_spec, chart_type, title
ANSWER	A chunk of the final answer text	content
STRUCTURED_OUTPUT	Extracted structured data	content, json_schema
LLM_RETRY	The agent is retrying a transient LLM failure	message
TOOL_ERROR	A specific tool failed (often recoverable)	error, tool_name
ERROR	Unrecoverable error; the stream terminates	error
COMPLETE	End of stream	consumption[], checkpoint_id
STREAM_ROLLBACK	Discard part of the answer already delivered	to_seq

Every message also carries an optional message_id (groups related chunks; see below) and a role (defaults to "assistant"; sub-agents may use other role names).

Typical lifecycle

A request that runs to completion emits events in roughly this order. Some types are optional and depend on the request configuration.

PLANNING            (if a research plan is used or generated)
  -> THINKING       (reasoning between actions)
  -> ACTION         (tool call begins)
  -> AUDIT          (tool result detail)
  -> (loop: more THINKING / ACTION / AUDIT cycles as needed)
GROUNDING           (one or more, referencing spans of the upcoming ANSWER)
CHART               (only if code execution produced a chart; anchored into the ANSWER)
ANSWER              (streamed in chunks; concatenate them in order)
STRUCTURED_OUTPUT   (only if the request included structured_output_schema)
COMPLETE            (carries final consumption and checkpoint_id)

A request that hits a recoverable problem may also emit LLM_RETRY (informational, agent retries automatically) or TOOL_ERROR (per-tool failure; the agent may try a different approach). An unrecoverable failure emits ERROR and terminates the stream without a COMPLETE event. A submitted workflow whose answer restarts emits STREAM_ROLLBACK before the replacement text.

Per-type reference

PLANNING

Emitted when the agent’s research plan changes. The full plan is sent each time; clients should replace any previously displayed plan with the latest version rather than diffing.

{
  "type": "PLANNING",
  "plan": {
    "title": "Credit Analysis Framework",
    "steps": [
      {"description": "Analyze business model", "status": "COMPLETED"},
      {"description": "Review financial statements", "status": "IN_PROGRESS"},
      {"description": "Assess liquidity", "status": "NOT_STARTED"}
    ]
  }
}

Step statuses: NOT_STARTED, IN_PROGRESS, COMPLETED, SKIPPED, FAILED.

THINKING

The agent’s intermediate reasoning between tool calls. content may be a partial chunk; concatenate consecutive chunks that share the same message_id to assemble a single reasoning block.

{
  "type": "THINKING",
  "message_id": "think-1",
  "content": "I need to retrieve the latest 10-K filing before..."
}

THINKING is informational. Many integrations suppress it from the user-facing UI and show only ANSWER content.

ACTION

Notification that the agent is calling a tool. Useful for showing per-tool progress in the UI (e.g., “Searching for earnings transcripts…”).

{
  "type": "ACTION",
  "tool_name": "search",
  "tool_arguments": {"query": "NVIDIA Q2 FY26 earnings"}
}

The corresponding result arrives later as an AUDIT message.

AUDIT

Detailed traces of tool executions. Each entry in audit_traces carries an audit_type discriminator and a tool_id that matches references in subsequent GROUNDING messages.

{
  "type": "AUDIT",
  "audit_traces": [
    {
      "audit_type": "SearchAuditV1",
      "tool_id": "audit-42",
      "query": {"text": "NVIDIA Q2 FY26 revenue"},
      "results": [
        {"type": "CQS", "values": [{"id": "doc-1", "hd": "NVIDIA 10-Q", "...": "..."}]}
      ]
    }
  ]
}

Audit trace types include SearchAuditV1, MarkdownAuditV1, StructuredReportAuditV1, SubAgentStartedAuditV1, and SubAgentCompletedAuditV1. The full schema for each is defined in the OpenAPI specs.

GROUNDING

Citations linking spans of the upcoming ANSWER text to their sources. Each reference carries start and end offsets, in Unicode code points, into the cumulative answer text. See the Grounding and citations guide for the full citation-rendering pattern, including the buffering requirement.

{
  "type": "GROUNDING",
  "references": [
    {
      "start": 120,
      "end": 175,
      "tool_name": "search",
      "audit_id": "audit-42",
      "source": {
        "type": "BIGDATA",
        "id": "doc-1",
        "hd": "NVIDIA 10-Q Filing",
        "src_name": "SEC EDGAR",
        "ts": "2026-04-15T00:00:00Z"
      }
    }
  ]
}

CHART

The agent sends this when it produces a chart while running code. See Code execution and charts for how to turn it on. vega_lite_spec is a standard Vega-Lite specification. You can render it with any Vega-Lite renderer. A chart also has start and end offsets. They point to a span of the cumulative ANSWER text, so you can place the chart where the answer refers to it.

{
  "type": "CHART",
  "chart_id": "chart_ea2d8588ce41",
  "title": "Tesla Quarterly Revenue",
  "caption": "Tesla Quarterly Revenue",
  "chart_type": "bar",
  "data_points": 6,
  "start": 0,
  "end": 17,
  "vega_lite_spec": {
    "$schema": "https://vega.github.io/schema/vega-lite/v6.json",
    "mark": {"type": "bar", "tooltip": true},
    "encoding": {
      "x": {"field": "fiscal_quarter", "type": "nominal"},
      "y": {"field": "revenue_billions", "type": "quantitative"}
    },
    "data": {"values": [{"fiscal_quarter": "2025-Q3", "revenue_billions": 28.095}]}
  }
}

The vega_lite_spec above is trimmed to one data point for space. A real spec carries the full series.

ANSWER

A chunk of the final answer text. Multiple ANSWER events fire as the model generates output. Concatenate content values in arrival order to reassemble the full answer. Grounding offsets index into this cumulative concatenated text.

{
  "type": "ANSWER",
  "message_id": "ans-1",
  "content": " The Data Center segment contributed $41.1 billion in"
}

STRUCTURED_OUTPUT

Only emitted when the request included a structured_output_schema. Fires once, after the final ANSWER chunk, with the extracted JSON. See the request schema for constraints on the input schema (top-level must be an object or list-of-objects).

{
  "type": "STRUCTURED_OUTPUT",
  "json_schema": {"type": "object", "properties": {"...": "..."}},
  "content": {"company": "NVIDIA", "revenue_q2_fy26": 46.74}
}

LLM_RETRY

Informational: the agent is retrying a transient LLM failure (rate limit, timeout, etc.). No client action is required; the agent will continue automatically. Log if useful for observability.

{
  "type": "LLM_RETRY",
  "message": "Retrying after upstream rate limit"
}

TOOL_ERROR

A specific tool failed. The agent may recover by calling a different tool or synthesizing without that source. The stream continues. See Error handling for guidance on what to surface to users.

{
  "type": "TOOL_ERROR",
  "tool_name": "search",
  "error": "Upstream search service timed out"
}

ERROR

Unrecoverable. The stream terminates without a COMPLETE event. Surface this to the user.

{
  "type": "ERROR",
  "error": "Request failed: invalid checkpoint id"
}

COMPLETE

End of stream. Carries the final token consumption breakdown and, for Research Agent requests, a checkpoint_id that can be passed as from_checkpoint_id in a follow-up request. See Conversation continuity.

{
  "type": "COMPLETE",
  "checkpoint_id": "ckpt-789",
  "consumption": [
    {"type": "base", "input_tokens": 12450, "output_tokens": 3120, "cached_tokens": 800},
    {"type": "pro", "input_tokens": 0, "output_tokens": 0, "cached_tokens": 0}
  ]
}

STREAM_ROLLBACK

Emitted when a run that had begun answering starts that answer again, so the text already sent is superseded. Discard every event whose SSE id is greater than to_seq, then render what follows. A to_seq of 0 discards the whole answer. Applying the same rollback twice is harmless.

{"type": "STREAM_ROLLBACK", "to_seq": 412}

This event is only delivered on a submitted workflow’s event stream (GET /v1/workflow/execute/async/{execution_id}/stream), where every event carries an SSE id to compare against. It never appears on the synchronous endpoints, and a stored run’s events list never replays it — by the time a run is stored, the superseded text is already gone.

Handle it if you render answer text as it arrives. If you buffer the whole run and only display the result at the end, keep the events in a list keyed by their SSE id, drop the ones above to_seq, and you are done.

message_id and chunk grouping

The optional message_id field groups chunks that belong to a single logical unit. The most common case is ANSWER: a long answer arrives as many small ANSWER events, all sharing the same message_id. Clients can use the id to detect a new answer turn versus a continuation of the current one. For most use cases, simply concatenating consecutive ANSWER chunks in arrival order works.

Canonical Python streaming handler

The handler below dispatches on type, reassembles the answer text, collects grounding references, and prints a useful per-event log. It is the reference implementation the rest of the docs build on.

import json
import requests


def stream_research_agent(api_key: str, message: str, research_effort: str = "standard"):
    """Stream a Research Agent request and return (answer_text, references, consumption)."""
    endpoint = "https://agents.bigdata.com/v1/research-agent"
    headers = {"X-API-KEY": api_key, "Content-Type": "application/json"}
    payload = {"message": message, "research_effort": research_effort}

    answer_text = ""
    references = []
    consumption = []

    with requests.post(endpoint, headers=headers, json=payload, stream=True, timeout=300) as r:
        r.raise_for_status()
        for raw_line in r.iter_lines(decode_unicode=True):
            if not raw_line or not raw_line.startswith("data: "):
                continue

            try:
                event = json.loads(raw_line[6:])
            except json.JSONDecodeError:
                # Malformed line; skip rather than abort the stream.
                continue

            msg = event.get("message", {})
            msg_type = msg.get("type")

            if msg_type == "PLANNING":
                plan = msg.get("plan", {})
                print(f"[plan] {plan.get('title')}: {len(plan.get('steps', []))} steps")

            elif msg_type == "THINKING":
                # Usually suppressed from end-user UIs.
                pass

            elif msg_type == "ACTION":
                print(f"[action] {msg.get('tool_name')}")

            elif msg_type == "AUDIT":
                for trace in msg.get("audit_traces", []):
                    print(f"[audit] {trace.get('audit_type')} {trace.get('tool_id')}")

            elif msg_type == "GROUNDING":
                references.extend(msg.get("references", []))

            elif msg_type == "CHART":
                # Render msg["vega_lite_spec"] with any Vega-Lite renderer.
                print(f"[chart] {msg.get('title')} ({msg.get('chart_type')})")

            elif msg_type == "ANSWER":
                chunk = msg.get("content", "")
                answer_text += chunk
                print(chunk, end="", flush=True)

            elif msg_type == "STRUCTURED_OUTPUT":
                print(f"\n[structured] {json.dumps(msg.get('content'))}")

            elif msg_type == "LLM_RETRY":
                # Informational; agent retries automatically.
                print(f"[retry] {msg.get('message')}")

            elif msg_type == "TOOL_ERROR":
                # Recoverable; log but continue consuming the stream.
                print(f"[tool-error] {msg.get('tool_name')}: {msg.get('error')}")

            elif msg_type == "ERROR":
                # Unrecoverable; the stream terminates after this event.
                raise RuntimeError(f"Research agent error: {msg.get('error')}")

            elif msg_type == "COMPLETE":
                consumption = msg.get("consumption", [])
                print("\n[complete]")

            else:
                # Forward-compatible: ignore unknown message types.
                pass

    return answer_text, references, consumption

For Workflows, the same handler works after two changes: post to https://agents.bigdata.com/v1/workflow/execute with a workflow payload, and read the typed event from event["delta"] instead of event["message"]. It also works unchanged against a submitted run’s event stream and against the events list of a stored run.

Forward compatibility

New message types may be added in future API versions. Clients should treat unknown type values as informational and continue processing the stream rather than raising. The else branch in the handler above demonstrates this pattern.

Next steps

Grounding and citations

How GROUNDING references map to answer text spans, and how to render inline citations.

Error handling

The difference between ERROR, TOOL_ERROR, and LLM_RETRY, plus HTTP-level failure modes.

Conversation continuity

Resuming Research Agent chats with checkpoints and Workflows runs with execution IDs.

Research Agent quickstart

Build your first Research Agent request end-to-end.

Running a workflow

Attach to a submitted run’s event stream and resume it after a dropped connection.