Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.aegra.dev/llms.txt

Use this file to discover all available pages before exploring further.

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.
ModeWhat it streams
valuesFull state snapshot after each node execution
updatesOnly the state changes (delta) produced by each node
messagesLLM tokens and tool calls as (message, metadata) tuples with accumulation into messages/partial and messages/complete events
messages-tupleRaw message tuples without accumulation (JavaScript graph compatibility)
customUser-defined data emitted from inside nodes via get_stream_writer()
eventsLow-level LangGraph astream_events for fine-grained tracing
debugCheckpoint 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:
EventDescription
metadataRun metadata (run_id, attempt number) — sent first
valuesFull state snapshot (when using values mode)
updatesState delta from a single node (when using updates mode)
messages/partialPartial message chunk (streaming token)
messages/completeComplete message after all tokens received
messages/metadataMessage metadata (run_id)
customUser-defined data from get_stream_writer()
eventsLangGraph internal events (when using events mode)
debugDebug checkpoint and task result events
errorError during execution
endStream 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.

​
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)
ValueBehavior
"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",
)

​
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:
  1. Track the last event ID you received
  2. Reconnect to GET /threads/{thread_id}/runs/{run_id}/stream with Last-Event-ID header
  3. You’ll receive all events from where you left off
Events are retained for 1 hour after the run completes.
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.