Streaming

AgentVisor™ supports real-time streaming for LLM responses and agent output.

Overview

Streaming enables:

• Real-time LLM output: See tokens as they're generated
• Progress updates: Monitor long-running operations
• Responsive UX: No waiting for complete responses

LLM Streaming

Standard LangChain streaming works transparently through the proxy:

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-20250514", streaming=True)

# Streaming works automatically through the proxy
for chunk in llm.stream("Tell me a story"):
    print(chunk.content, end="", flush=True)

This is streaming inside your agent process — the tokens never leave the sandbox unless your agent code re-emits them (see Emitting Stream Chunks below). To deliver progress to an external client over HTTP, use the SSE API described next.

API Streaming

AgentVisor exposes three SSE (Server-Sent Events) endpoints. They emit different frame shapes — pick the one that matches your integration:

Endpoint	Chunk data shape	Leading start event	done event data
GET /threads/{thread_id}/runs/{run_id}/stream	raw chunk payload	No	{}
GET /runs/{run_id}/stream (by composite run ID)	{"type": "<chunk type>", "data": {...}}	No	full run object ({} if the final lookup fails)
POST /runs/stream (stateless)	{"type": "<chunk type>", "data": {...}}	Yes	full run object ({} if the final lookup fails)

Every event's SSE event: name is the chunk type — there is no generic event: chunk wrapper. See the Runs API reference for the full per-endpoint spec; this guide focuses on how to consume the frames.

Thread-scoped Streaming

curl -N "http://localhost:8090/threads/$THREAD/runs/$RUN/stream"

Real captured output for a run with "stream": true, where the agent called stream.emit_token() four times and then returned (LangGraph's own stream_mode="updates" auto-emission produced the final updates event):

event: token
data: {"token":"Real"}

event: token
data: {"token":"-time"}

event: token
data: {"token":" updates"}

event: token
data: {"token":" work!"}

event: updates
data: {"respond":{"messages":[{"content":"Real-time updates work!","role":"assistant"}]}}

event: done
data: {}

data is always the chunk's raw payload — never wrapped in a {"type": ..., "data": ...} envelope on this endpoint. The final done event's data is always the literal {}; look up the run separately (GET /threads/{thread_id}/runs/{run_id}) for its final status and output.

Run-by-ID and Stateless Streaming

GET /runs/{run_id}/stream and POST /runs/stream share the same wrapped shape. Real captured output for POST /runs/stream against the same agent:

event: start
data: {"run_id":"242cdf3a-...-d8f24b3a-...","status":"running","thread_id":"242cdf3a-..."}

event: token
data: {"data":{"token":"Real"},"type":"token"}

event: token
data: {"data":{"token":"-time"},"type":"token"}

event: token
data: {"data":{"token":" updates"},"type":"token"}

event: token
data: {"data":{"token":" work!"},"type":"token"}

event: updates
data: {"data":{"respond":{"messages":[{"content":"Real-time updates work!","role":"assistant"}]}},"type":"updates"}

event: done
data: {"run_id":"242cdf3a-...","thread_id":"242cdf3a-...","status":"completed","output":{"respond":{"messages":[{"content":"Real-time updates work!","role":"assistant"}]}},"created_at":"...","updated_at":"..."}

GET /runs/{run_id}/stream (streaming an existing run/thread by composite ID instead of creating a new one) emits identical shapes for <chunk-type> and done, but has no leading start event.

curl -N -X POST "http://localhost:8090/runs/stream" \
    -H "Content-Type: application/json" \
    -d '{"input": {"messages": [{"role": "user", "content": "Hello"}]}}'

note

When "stream": true is set, the run's final output reflects the last streamed chunk (shape depends on the stream mode — e.g. {"<node_name>": {...}} under the default updates mode), not the full graph state you'd get back from a non-streaming run.

Event Types

Event	When	Meaning
start	First event, POST /runs/stream only	Run created; data is {run_id, thread_id, status}
<chunk type>	Zero or more, per emitted chunk	Event name is whatever chunk_type the agent passed — see below
done	Always last (absent on error)	Stream complete
error	Instead of done, if streaming fails	data is {"error": "<message>"}

<chunk type> is not a fixed enum — it's the chunk_type string the agent (or the LangGraph runner's own auto-streaming) passed to emit_chunk. Two independent sources feed the same stream:

• LangGraph's own streaming (enabled by setting "stream": true on run creation) emits one chunk per graph step, with chunk_type set to the LangGraph stream mode: values, updates (the default), messages, or custom.
• Manual chunks emitted from your node code via stream.emit_token(), stream.emit_message(), stream.emit_state(), or stream.emit_chunk(data, chunk_type=...) — common chunk_type values are token, message, state, or custom (the default).

Both interleave on the same stream in emission order, as shown in the captured output above (four token events from manual calls inside the node, followed by one updates event from LangGraph's own per-step streaming).

Client Examples

These examples target the thread-scoped endpoint (GET /threads/{id}/runs/{runId}/stream), where data is the raw chunk payload and the chunk type is only available from the SSE event: line — not from a field inside the JSON body.

Python

import json
import requests

def stream_run(thread_id, run_id):
    url = f"http://localhost:8090/threads/{thread_id}/runs/{run_id}/stream"
    event_type = None

    with requests.get(url, stream=True) as response:
        response.raise_for_status()
        for line in response.iter_lines(decode_unicode=True):
            if line == "":
                event_type = None  # blank line ends the frame
                continue
            if line.startswith("event: "):
                event_type = line[len("event: "):]
            elif line.startswith("data: "):
                data = json.loads(line[len("data: "):])
                if event_type == "done":
                    print("\n--- done ---")
                    return
                if event_type == "error":
                    raise RuntimeError(data["error"])
                if event_type == "token":
                    print(data["token"], end="", flush=True)
                else:
                    print(f"\n[{event_type}] {data}")

JavaScript

async function streamRun(threadId, runId) {
  const url = `http://localhost:8090/threads/${threadId}/runs/${runId}/stream`;
  const response = await fetch(url);
  const reader = response.body.getReader();
  const decoder = new TextDecoder();

  let buffer = '';
  let eventType = null;

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

    let newlineIndex;
    while ((newlineIndex = buffer.indexOf('\n')) >= 0) {
      const line = buffer.slice(0, newlineIndex).replace(/\r$/, '');
      buffer = buffer.slice(newlineIndex + 1);

      if (line === '') {
        eventType = null; // blank line ends the frame
        continue;
      }
      if (line.startsWith('event: ')) {
        eventType = line.slice('event: '.length);
      } else if (line.startsWith('data: ')) {
        const data = JSON.parse(line.slice('data: '.length));
        if (eventType === 'done') {
          console.log('\n--- done ---');
          return;
        }
        if (eventType === 'error') {
          throw new Error(data.error);
        }
        if (eventType === 'token') {
          process.stdout.write(data.token);
        } else {
          console.log(`\n[${eventType}]`, data);
        }
      }
    }
  }
}

Emitting Stream Chunks

From your agent code, emit chunks with stream.emit_chunk() (or the convenience helpers emit_token, emit_message, emit_state) — the chunk_type argument becomes the SSE event: name, and data is delivered verbatim (never wrap "type" inside data yourself):

from agentvisor import stream

def my_node(state: AgentState) -> dict:
    # Convenience helpers for common chunk types
    stream.emit_token("Hello")
    stream.emit_message("assistant", "Working on it...")
    stream.emit_state({"step": 1, "total": 3})

    # Generic form — any chunk_type string is allowed; defaults to "custom"
    stream.emit_chunk({"tool": "search", "status": "started"}, chunk_type="tool_call")

    result = process_step()

    return {"result": result}

Streaming with LangGraph

LangGraph's own graph.stream() output is forwarded automatically — you don't need to call it yourself. Set "stream": true on the run-creation request body, and the guest runtime runs your graph with graph.stream(input, stream_mode="updates") by default, emitting one chunk per graph step with chunk_type set to the stream mode.

To use a different LangGraph stream mode (values, messages, or custom), set AGENTVISOR_STREAM_MODE via framework.env in mav-agent-config.yaml:

# mav-agent-config.yaml
framework:
  provider: "langgraph"
  env:
    AGENTVISOR_STREAM_MODE: "messages"  # values | updates (default) | messages | custom

messages mode is useful for token-by-token LLM output — LangGraph emits one chunk per token as your graph's LLM node streams, without any manual stream.emit_token() calls.

SSE Format

AgentVisor uses standard Server-Sent Events framing — each frame is an event: line, a data: line, and a blank line:

event: <event-type>
data: <json-payload>

event: <event-type>
data: <json-payload>

Configuration

Timeouts

For long-running streams:

# Increase proxy timeout for LLM requests
export AGENTVISOR_PROXY_REQUEST_TIMEOUT=120s

Error Handling

The stream automatically closes if the client disconnects, and a genuine streaming failure (rather than a normal end-of-stream) arrives as an event: error frame instead of event: done — see Event Types. Handle transient HTTP-level disconnects with a retry loop:

import requests

def stream_with_retry(url, max_retries=3):
    for attempt in range(max_retries):
        try:
            with requests.get(url, stream=True, timeout=120) as response:
                response.raise_for_status()
                for line in response.iter_lines(decode_unicode=True):
                    yield line
                return
        except requests.exceptions.ChunkedEncodingError:
            if attempt < max_retries - 1:
                continue
            raise
        except requests.exceptions.Timeout:
            if attempt < max_retries - 1:
                continue
            raise

Best Practices

Emit Progress Updates

For long operations, emit progress so clients know the agent is working:

def long_running_node(state: AgentState) -> dict:
    items = state["items"]
    results = []

    for i, item in enumerate(items):
        stream.emit_chunk({
            "current": i + 1,
            "total": len(items),
            "message": f"Processing {item}..."
        }, chunk_type="progress")
        result = process(item)
        results.append(result)

    return {"results": results}

Handle Client Disconnects

If the SSE client disconnects, the host simply stops forwarding chunks — your agent keeps running unaffected, and emit_* calls keep succeeding (they write to the run's buffered chunk history, not directly to the HTTP response). No special handling is needed in node code for this case; see the retry guide for the separate (and unrelated) matter of transient gRPC failures on emit_* calls themselves:

def my_node(state: AgentState) -> dict:
    result = expensive_operation()
    stream.emit_chunk({"data": result}, chunk_type="result")
    return {"result": result}

Use Appropriate Chunk Sizes

Don't emit too frequently (network overhead) or too infrequently (poor UX):

# Good: batch small updates
buffer = []
for item in items:
    buffer.append(process(item))
    if len(buffer) >= 10:
        stream.emit_chunk({"items": buffer}, chunk_type="batch")
        buffer = []
if buffer:
    stream.emit_chunk({"items": buffer}, chunk_type="batch")

# Bad: emit every single item
for item in items:
    stream.emit_chunk({"data": process(item)}, chunk_type="item")  # Too chatty