Streaming

Aegra streams agent execution to clients using Server-Sent Events (SSE). This gives you real-time token-by-token output, tool call updates, and state changes as they happen.

Quick example

import asyncio
from langgraph_sdk import get_client


async def main():
    client = get_client(url="http://localhost:2026")

    thread = await client.threads.create()

    async for chunk in client.runs.stream(
        thread_id=thread["thread_id"],
        assistant_id="agent",
        input={"messages": [{"type": "human", "content": "Hello!"}]},
        stream_mode=["messages-tuple"],
    ):
        if hasattr(chunk, "data") and chunk.data:
            print(chunk.data)


asyncio.run(main())

Stream modes

Control what data you receive by setting stream_mode on the run. You can pass a single mode as a string or multiple modes as a list.

Mode	What it streams
`values`	Full state snapshot after each node execution
`updates`	Only the state changes (delta) produced by each node
`messages`	LLM tokens and tool calls as `(message, metadata)` tuples with accumulation into `messages/partial` and `messages/complete` events
`messages-tuple`	Raw message tuples without accumulation (JavaScript graph compatibility)
`custom`	User-defined data emitted from inside nodes via `get_stream_writer()`
`events`	Low-level LangGraph `astream_events` for fine-grained tracing
`debug`	Checkpoint and task result events (auto-enabled on all streams)

debug and updates are used internally on every stream for checkpoint tracking and interrupt detection, but their events are only sent to the client if you explicitly request them in stream_mode. When updates is not requested, only interrupt events (human-in-the-loop) are forwarded — remapped as values events for compatibility.

Custom streaming

Send custom data from inside graph nodes using get_stream_writer():

from langgraph.config import get_stream_writer


def my_node(state):
    writer = get_stream_writer()
    writer({"progress": "Fetching data..."})
    # ... do work ...
    writer({"progress": "Processing results..."})
    return {"result": "done"}

Receive custom events by including custom in the stream modes:

async for chunk in client.runs.stream(
    thread_id=thread_id,
    assistant_id="agent",
    input={"messages": [{"type": "human", "content": "Analyze this"}]},
    stream_mode=["custom", "values"],
):
    print(f"Event: {chunk.event}, Data: {chunk.data}")

Using multiple modes

async for chunk in client.runs.stream(
    thread_id=thread_id,
    assistant_id="agent",
    input={"messages": [{"type": "human", "content": "Search for AI news"}]},
    stream_mode=["values", "messages-tuple"],
):
    print(f"Event: {chunk.event}, Data: {chunk.data}")

When multiple modes are active, the event field tells you which mode each chunk comes from.

Event types

During streaming, you’ll receive these event types:

Serialized graph values

Graph state and streamed events are converted to JSON-compatible values before they are sent over SSE or stored in the run output. Dataclasses are serialized recursively by field, and common Python values use these representations:

Python value	JSON representation
`bytes`	Base64-encoded string
`Enum`	Serialized enum value
`datetime` / `date`	ISO 8601 string
`UUID`, `Decimal`, `Path`	String
`deque`, `tuple`, `set`, `frozenset`	JSON array with recursively serialized members
`Exception`	Object containing `type` and `message`

Nested values inside Pydantic or LangChain model dumps are normalized using the same rules. Mapping values are serialized recursively, and non-JSON-native keys such as UUID or Path are converted to JSON property names with the same scalar rules. Key collisions after conversion raise a serialization error.

Event	Description
`metadata`	Run metadata (run_id, attempt number) — sent first
`values`	Full state snapshot (when using `values` mode)
`updates`	State delta from a single node (when using `updates` mode)
`messages/partial`	Partial message chunk (streaming token)
`messages/complete`	Complete message after all tokens received
`messages/metadata`	Message metadata (run_id)
`custom`	User-defined data from `get_stream_writer()`
`events`	LangGraph internal events (when using `events` mode)
`debug`	Debug checkpoint and task result events
`error`	Error during execution
`end`	Stream complete

Streaming endpoints

Create and stream

The most common pattern — create a run and stream its output in one call:

async for chunk in client.runs.stream(
    thread_id=thread_id,
    assistant_id="agent",
    input={"messages": [{"type": "human", "content": "Hello"}]},
):
    print(chunk)

This calls POST /threads/{thread_id}/runs/stream under the hood.

Stream an existing run

If you created a background run, you can stream it later:

GET /threads/{thread_id}/runs/{run_id}/stream

This supports reconnection via the Last-Event-ID header. If the connection drops, the client can reconnect and receive events from where it left off.

Wait for completion

If you don’t need streaming but want to wait for the result:

result = await client.runs.wait(
    thread_id=thread_id,
    assistant_id="agent",
    input={"messages": [{"type": "human", "content": "Hello"}]},
)
print(result)

This calls POST /threads/{thread_id}/runs/wait and returns the final output. If the run fails, the body is an error envelope instead of an output:

{"__error__": {"error": "Error", "message": "KeyError: 'topic'"}}

client.runs.wait() raises on that envelope by default; pass raise_error=False to get it back as a dict. The same envelope is returned when the wait gives up before the run reaches a terminal state (TimeoutError or IncompleteRun), and by client.runs.join(), which returns it rather than raising. An interrupted run is not a failure — it returns its partial output with __interrupt__.

Subgraph streaming

If your graph uses subgraphs, you can stream events from them too:

async for chunk in client.runs.stream(
    thread_id=thread_id,
    assistant_id="agent",
    input={"messages": [{"type": "human", "content": "Hello"}]},
    stream_subgraphs=True,
):
    print(chunk)

Disconnection behavior

By default, when a client disconnects during streaming, the run is cancelled. You can change this:

async for chunk in client.runs.stream(
    thread_id=thread_id,
    assistant_id="agent",
    input={"messages": [{"type": "human", "content": "Hello"}]},
    on_disconnect="continue",  # keep running even if client disconnects
):
    print(chunk)

Value	Behavior
`"cancel"` (default)	Cancel the run when client disconnects
`"continue"`	Run continues in the background; reconnect later to get results

The server sends SSE keepalive comments (: heartbeat) every KEEPALIVE_INTERVAL_SECS (default 5s, truncated to whole seconds, minimum 1s), so idle proxies (Nginx 60s, AWS ALB, Cloudflare) won’t drop long-running silent agents — for example, a graph node holding an upstream WebSocket without emitting events. The wire-format matches LangGraph Platform, so existing SSE clients (the LangGraph SDK, browser EventSource, httpx-sse) silently ignore these lines per the W3C spec. Only a real client disconnect — not an idle proxy timeout — triggers automatic cancellation.

Background runs

For long-running tasks, you can create a run in the background and check on it later:

# Create run (returns immediately)
run = await client.runs.create(
    thread_id=thread_id,
    assistant_id="agent",
    input={"messages": [{"type": "human", "content": "Analyze this dataset"}]},
)

# Check status
run = await client.runs.get(thread_id=thread_id, run_id=run["run_id"])
print(run["status"])  # "pending", "running", "success", "error"

# Wait for completion and get output
output = await client.runs.join(thread_id=thread_id, run_id=run["run_id"])
print(output)

Cancelling runs

Cancel or interrupt a running execution:

# Hard cancel
await client.runs.cancel(thread_id=thread_id, run_id=run["run_id"])

# Cooperative interrupt (if graph supports it)
await client.runs.cancel(
    thread_id=thread_id,
    run_id=run["run_id"],
    action="interrupt",
)

Cancellation is asynchronous while a live worker owns the run. The cancel response may therefore still show pending or running; poll, join, or stream the run until it reaches a terminal status. Use the REST endpoint with wait=1 when the caller needs the request to wait briefly for the worker to settle.

Cancelling several runs

POST /runs/cancel cancels many runs in one call. Select them by thread_id plus run_ids, or by status (pending, running, or all). The SDK exposes it as cancel_many:

# Stop specific runs on a thread
await client.runs.cancel_many(thread_id=thread_id, run_ids=[run_a, run_b], action="interrupt")

# Stop every active run you own
await client.runs.cancel_many(status="all")

Runs that already finished are skipped, and only the caller’s runs are affected. action accepts cancel and interrupt, the same values as the per-run endpoint. The response is 204 No Content.

SSE reconnection

Aegra stores streaming events in a replay buffer (Redis Lists when REDIS_BROKER_ENABLED=true, in-memory list in dev mode) for replay. If your connection drops:

Track the last event ID you received
Reconnect to GET /threads/{thread_id}/runs/{run_id}/stream with Last-Event-ID header
You’ll receive all events from where you left off

Events are retained for 10 minutes after the run completes, in both dev and production.

In production deployments, SSE events are delivered via Redis pub/sub from workers — the client’s SSE connection and the worker executing the run can be on different instances. See the worker architecture guide for details on how this works.

Agent Protocol v2 event streaming

The latest LangGraph JS and Python SDKs (including @langchain/langgraph-sdk and the Vue/React useStream() composables) speak a newer streaming protocol with a dedicated event envelope and a content-block message model. Aegra serves this protocol natively.

v2 streaming is on by default — it’s a new endpoint set the SDK targets and has no v1 to break; the legacy runs/stream endpoints are unchanged. FF_V2_EVENT_STREAMING is a kill switch: set it to false to disable v2 serving (requests return 503 with an enable hint) and roll back without a redeploy. The endpoints also require a langgraph / langchain-core new enough to emit native v3 events; the server returns 503 with an upgrade hint if the runtime is too old.

Using the SDK

The stock LangGraph SDK drives both endpoints for you. Streaming is thread-scoped: open the stream, start a run, and consume the events:

from langgraph_sdk import get_client

client = get_client(url="http://localhost:8000")

async with client.threads.stream(assistant_id="agent") as ts:
    await ts.run.start(input={"messages": [{"role": "user", "content": "hi"}]})
    async for event in ts.events:
        print(event["method"], event["params"]["data"])

The same flow backs the Vue/React useStream() composables.

Endpoints

Endpoint	Purpose
`POST /threads/{thread_id}/commands`	Run a command: `run.start` (start a run), `input.respond` (resume an interrupt). Returns a JSON response envelope.
`POST /threads/{thread_id}/stream/events`	Open a channel-filtered SSE stream of the thread’s run events.

The stream is scoped to the thread, not a run — you open it, then issue run.start, and the events of whatever run executes flow through. The client mints the thread id, and run.start creates the thread if it doesn’t exist yet.

Starting a run

curl -X POST http://localhost:8000/threads/$THREAD/commands \
  -H "Content-Type: application/json" \
  -d '{
        "id": 1,
        "method": "run.start",
        "params": {
          "assistant_id": "agent",
          "input": {"messages": [{"role": "user", "content": "hi"}]},
          "context": {"tenant_id": "acme"}
        }
      }'
# → {"type": "success", "id": 1, "result": {"run_id": "..."}}

context is forwarded to both the graph factory through ServerRuntime[T] and graph nodes through Runtime[T]. It is an Aegra extension: the Agent Protocol run.start params are assistant_id, input, config, metadata, and langsmith_tracer. For a payload that also works against LangGraph Platform, put the same values under config.configurable; when context is absent the server copies configurable into it. To fork or time travel, set config.configurable.checkpoint_id to a checkpoint of the thread. The run replays from that checkpoint, and input is optional in that case. The JS SDK forkFrom option (the regenerate button in useStream and assistant-ui) sends this field. A value that is not a UUID string returns invalid_argument. langsmith_tracer is accepted and ignored. Aegra traces through OpenTelemetry, see Observability. Params outside the documented set are logged at warning level with their key names and otherwise ignored.

Streaming events

The SSE filter is a POST body listing the channels you want — no run id. Pass since (the last seq you saw) to resume after a dropped connection.

curl -N -X POST http://localhost:8000/threads/$THREAD/stream/events \
  -H "Content-Type: application/json" \
  -d '{"channels": ["messages", "values", "lifecycle"]}'

The client reads each frame’s data: line — a protocol event envelope. seq is the cursor you echo back as since; event_id dedups across reconnects; params.data is the payload and params.namespace the subgraph path:

data: {"type":"event","seq":1,"event_id":"...:1","method":"messages","params":{"data":{"event":"message-start","role":"ai","id":"msg_1"},"namespace":[]}}

data: {"type":"event","seq":2,"event_id":"...:2","method":"messages","params":{"data":{"event":"content-block-delta","index":0,"delta":{"type":"text-delta","text":"Hello"}},"namespace":[]}}

data: {"type":"event","seq":3,"event_id":"...:3","method":"lifecycle","params":{"data":{"event":"completed"},"namespace":[]}}

Channels

values, updates, messages, tools, lifecycle, input, checkpoints, tasks, and custom (plus custom:<name>). Message streams arrive as content-block events (message-start → content-block-delta → message-finish); lifecycle reports started / completed / failed / interrupted.

Human-in-the-loop interrupts

When you subscribe to the input channel, a graph interrupt arrives as an input.requested event. Its params.data includes the interrupt id and two identical content fields:

{
  "type": "event",
  "seq": 4,
  "event_id": "...:4",
  "method": "input.requested",
  "params": {
    "data": {
      "interrupt_id": "interrupt-id",
      "payload": {"question": "Approve?"},
      "value": {"question": "Approve?"}
    },
    "namespace": []
  }
}

payload is the Agent Protocol field consumed by the JavaScript SDK and Vue/React useStream() integrations. value mirrors it for current Python SDKs, including ts.interrupts[].value. If the source interrupt has no value, both content fields are omitted; explicit null, empty, and falsy values are preserved under both fields. Interrupt IDs are deduplicated within each run so the same interrupt is not emitted twice when it appears in both updates and values. A resumed run may reuse an earlier run’s ID; its input.requested event is emitted again because it belongs to a new run.

Responding to an interrupt

input.respond resumes the interrupted run. Only response is required; interrupt_id targets one interrupt when several are pending.

curl -X POST http://localhost:8000/threads/$THREAD/commands   -H "Content-Type: application/json"   -d '{
        "id": 2,
        "method": "input.respond",
        "params": {
          "interrupt_id": "interrupt-id",
          "namespace": [],
          "response": {"action": "approve"},
          "update": {"reviewed_by": "alice"},
          "goto": "finalize"
        }
      }'

Param	Effect
`response`	The value returned from `interrupt()` inside the graph.
`interrupt_id`	Targets one pending interrupt. Omit it only when a single interrupt is pending.
`responses`	Batch form: a list of `{interrupt_id, response}` entries resumed in one run. Replaces `response` and `interrupt_id`.
`update`	State update applied in the same superstep as the resume, as `Command(update=...)`. Must be an object.
`goto`	Node name, `{node}` or `{node, input}` object, or a list of those, applied as `Command(goto=...)`. `input` is optional.
`config`, `metadata`	Per-run overrides, same as `run.start`.
`context`	Aegra extension, same semantics as on `run.start`.
`namespace`	Accepted for protocol compatibility. Interrupt ids already encode the subgraph namespace, so the server does not read it.

update and goto fold into the same Command as the resume, so the resumed run writes a single checkpoint.

This first release covers the HTTP SSE + commands path the JS/Python SDKs use, verified end-to-end against the stock langgraph-sdk. The WebSocket transport and the agent.getTree / state.fork / subscription.* commands are not yet implemented; SSE filtering via the POST body covers the useStream() path without them.

Getting started

Core concepts

Data & storage

Configuration

Contributing

Reference

Quick example

Stream modes

Custom streaming

Using multiple modes

Event types

Serialized graph values

Streaming endpoints

Create and stream

Stream an existing run

Wait for completion

Subgraph streaming

Disconnection behavior

Background runs

Cancelling runs

Cancelling several runs

SSE reconnection

Agent Protocol v2 event streaming

Using the SDK

Endpoints

Starting a run

Streaming events

Channels

Human-in-the-loop interrupts

Responding to an interrupt

​Quick example

​Stream modes

​Custom streaming

​Using multiple modes

​Event types

​Serialized graph values

​Streaming endpoints

​Create and stream

​Stream an existing run

​Wait for completion

​Subgraph streaming

​Disconnection behavior

​Background runs

​Cancelling runs

​Cancelling several runs

​SSE reconnection

​Agent Protocol v2 event streaming

​Using the SDK

​Endpoints

​Starting a run

​Streaming events

​Channels

​Human-in-the-loop interrupts

​Responding to an interrupt

Quick example

Stream modes

Custom streaming

Using multiple modes

Event types

Serialized graph values

Streaming endpoints

Create and stream

Stream an existing run

Wait for completion

Subgraph streaming

Disconnection behavior

Background runs

Cancelling runs

Cancelling several runs

SSE reconnection

Agent Protocol v2 event streaming

Using the SDK

Endpoints

Starting a run

Streaming events

Channels

Human-in-the-loop interrupts

Responding to an interrupt