Streaming Events from LangGraph: A Practical Guide

LangGraph streaming is how you turn a graph that silently computes for ten seconds into one that shows the user something happening on every step. If you have ever built a LangGraph agent and watched the UI sit frozen until the whole run finishes, the fix is almost always the same: you are calling invoke() when you should be calling stream() or astream_events(). This guide walks through every stream mode LangGraph exposes, shows working code for each one, and ends with a pattern for piping token-level output into a web frontend over Server-Sent Events.

We will build a small multi-node graph (a router node plus a tool-calling node plus a final answer node) and stream it four different ways so you can see exactly what each mode gives you and when to reach for it.

Why streaming matters for graph-based agents

A single LLM call is easy to stream: you get a token, you print it, repeat. A LangGraph agent is harder because a single "run" is actually a sequence of node executions, and each node might itself contain zero, one, or several LLM calls, tool calls, and state updates. "Streaming" in LangGraph therefore has to answer two separate questions:

• Which node is currently running, and what did it just write to state?
• Within a node, what tokens is the LLM emitting right now?

LangGraph separates these concerns cleanly. stream_mode="updates" and stream_mode="values" answer the first question. stream_mode="messages" and astream_events() answer the second. Understanding this split is the single most useful mental model for LangGraph streaming, because most confusion comes from picking the wrong mode for the question you are actually trying to answer.

Setting up the example graph

Here is a minimal graph with three nodes: a classifier, a tool-calling researcher, and a writer that produces the final answer. We will reuse this graph for every streaming example below.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.tools import tool
from langchain_anthropic import ChatAnthropic

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    topic: str

llm = ChatAnthropic(model="claude-sonnet-5", temperature=0)

@tool
def search_docs(query: str) -> str:
    """Look up internal documentation for a given query."""
    return f"Docs snippet for: {query}"

def classify(state: AgentState) -> dict:
    topic = "technical" if "error" in state["messages"][-1].content.lower() else "general"
    return {"topic": topic}

def research(state: AgentState) -> dict:
    result = search_docs.invoke(state["messages"][-1].content)
    return {"messages": [AIMessage(content=f"Found context: {result}")]}

def write_answer(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

graph = StateGraph(AgentState)
graph.add_node("classify", classify)
graph.add_node("research", research)
graph.add_node("write_answer", write_answer)

graph.add_edge(START, "classify")
graph.add_edge("classify", "research")
graph.add_edge("research", "write_answer")
graph.add_edge("write_answer", END)

app = graph.compile()

Nothing fancy: classify tags the request, research calls a tool, write_answer calls the LLM. Now let us stream it.

stream_mode="values": the full state after every node

This is the simplest mode. After every node finishes, LangGraph hands you the entire current state. It is verbose but it is the easiest one to reason about when you are debugging.

inputs = {"messages": [HumanMessage(content="Why am I getting a timeout error?")]}

for state in app.stream(inputs, stream_mode="values"):
    print("---")
    print("topic:", state.get("topic"))
    print("last message:", state["messages"][-1].content[:80])

Every iteration prints the whole state as it stands, including keys that a previous node set and this node did not touch. Use values when you are building an admin panel or a debug console where you want to see the accumulated picture, not just the delta.

stream_mode="updates": only what changed

updates gives you a dict keyed by node name, containing only what that node returned. This is the mode you want for building a "thinking..." trace in a UI, because it maps cleanly onto "node X just ran and produced Y."

for update in app.stream(inputs, stream_mode="updates"):
    for node_name, node_output in update.items():
        print(f"[{node_name}] wrote: {node_output}")

Output looks like:

[classify] wrote: {'topic': 'technical'}
[research] wrote: {'messages': [AIMessage(content='Found context: ...')]}
[write_answer] wrote: {'messages': [AIMessage(content='A timeout error usually means...')]}

This is the workhorse mode for logging and for driving a step-by-step progress indicator ("Classifying request", "Searching docs", "Writing answer") without shipping the entire state payload on every tick.

stream_mode="messages": token-level streaming from inside nodes

This is what most people actually mean when they say "I want streaming." messages mode streams individual LLM tokens as they are produced inside any node, tagged with metadata about which node and which LLM call they came from.

for chunk, metadata in app.stream(inputs, stream_mode="messages"):
    if chunk.content:
        print(chunk.content, end="", flush=True)
    node = metadata.get("langgraph_node")
    if node:
        pass  # useful if you want to route tokens by node in a multi-agent UI

The metadata dict tells you which node (langgraph_node) and which run the chunk belongs to. This matters once you have more than one node calling an LLM: without the metadata, you cannot tell whether a token belongs to a classifier's structured-output call or the final answer, and you would end up printing internal reasoning to the user by accident. Always filter on metadata["langgraph_node"] when a graph has multiple LLM-calling nodes and you only want to surface one of them.

TARGET_NODE = "write_answer"

for chunk, metadata in app.stream(inputs, stream_mode="messages"):
    if metadata.get("langgraph_node") == TARGET_NODE and chunk.content:
        print(chunk.content, end="", flush=True)

Combining multiple stream modes

You can request more than one mode at once by passing a list, which is handy when your UI needs both a progress trace and token output simultaneously.

for mode, payload in app.stream(inputs, stream_mode=["updates", "messages"]):
    if mode == "updates":
        for node_name, node_output in payload.items():
            print(f"\n[STEP] {node_name}")
    elif mode == "messages":
        chunk, metadata = payload
        if chunk.content:
            print(chunk.content, end="", flush=True)

When you pass a list, each yielded item is a (mode, payload) tuple instead of just the payload, so you branch on mode to know how to unpack it. This single loop gives you both a step tracker and a token stream from one call, which is exactly what a chat UI with a "thinking" indicator needs.

astream_events: the full event firehose

stream() covers most cases, but astream_events() gives you a lower-level, more granular event stream: node starts, node ends, chat model starts, chat model streams, tool starts, tool ends, and chain starts and ends, each as a distinct event with a name, an event type, and data. Reach for this when you need to react to specific lifecycle moments (like "show a spinner the instant a tool call begins") rather than just node boundaries.

import asyncio

async def run_with_events():
    async for event in app.astream_events(inputs, version="v2"):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            token = event["data"]["chunk"].content
            if token:
                print(token, end="", flush=True)
        elif kind == "on_tool_start":
            print(f"\n[tool started] {event['name']} input={event['data'].get('input')}")
        elif kind == "on_tool_end":
            print(f"[tool finished] {event['name']}")
        elif kind == "on_chain_start" and event["name"] == "write_answer":
            print("\n[node started] write_answer")

asyncio.run(run_with_events())

astream_events is async-only and requires version="v2" on current LangGraph releases. It is more verbose to filter (you are pattern-matching on event names and types instead of getting clean node keys) but it is the only mode that exposes tool start and tool end as first-class events, which matters if you want a UI that shows "Searching docs..." while a tool call is in flight and then swaps to the actual answer once the tool returns.

Streaming custom data from inside a node

Sometimes a node does work that is not an LLM call or a tool call, and you still want to report progress: a loop over documents, a multi-step calculation, a batch embedding job. LangGraph supports this with stream_mode="custom" combined with a get_stream_writer() call inside the node.

from langgraph.config import get_stream_writer

def research(state: AgentState) -> dict:
    writer = get_stream_writer()
    queries = ["error codes", "timeout config", "retry policy"]
    results = []
    for i, q in enumerate(queries):
        writer({"progress": f"searching {q}", "step": i + 1, "total": len(queries)})
        results.append(search_docs.invoke(q))
    return {"messages": [AIMessage(content=f"Found: {results}")]}

for chunk in app.stream(inputs, stream_mode="custom"):
    print("progress event:", chunk)

This is the escape hatch for anything that is not naturally an LLM token: file processing, database scans, or a long-running tool that reports its own progress. get_stream_writer() works whether the node is sync or async, and the writer call is a no-op if nobody is listening on custom mode, so it is safe to leave in production code even when a caller uses invoke() instead of stream().

Wiring LangGraph streaming into a FastAPI backend with SSE

The most common real-world use case is piping graph tokens to a browser. Server-Sent Events (SSE) are the simplest transport for this because they are just a long-lived HTTP response with a specific text format, no websocket handshake required.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

api = FastAPI()

async def event_generator(user_input: str):
    inputs = {"messages": [HumanMessage(content=user_input)]}
    async for chunk, metadata in app.astream(inputs, stream_mode="messages"):
        if metadata.get("langgraph_node") == "write_answer" and chunk.content:
            payload = json.dumps({"token": chunk.content})
            yield f"data: {payload}\n\n"
    yield "data: [DONE]\n\n"

@api.post("/chat")
async def chat(user_input: str):
    return StreamingResponse(
        event_generator(user_input),
        media_type="text/event-stream",
    )

Note the switch from app.stream() to app.astream(), which is the async equivalent and the one you want inside an async FastAPI route handler. The text/event-stream media type and the data: ...\n\n framing are what let a browser's EventSource API (or a fetch call reading the response body as a stream) render tokens as they arrive instead of waiting for the full response.

On the frontend, a minimal consumer looks like this:

const response = await fetch("/chat", {
  method: "POST",
  body: JSON.stringify({ user_input: "Why am I getting a timeout error?" }),
});
const reader = response.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 });
  const lines = buffer.split("\n\n");
  buffer = lines.pop();
  for (const line of lines) {
    if (!line.startsWith("data: ")) continue;
    const data = line.slice(6);
    if (data === "[DONE]") continue;
    const { token } = JSON.parse(data);
    document.getElementById("output").textContent += token;
  }
}

This pattern, an async generator on the backend paired with a manual ReadableStream reader on the frontend, works with any framework that can return a streaming HTTP response, not just FastAPI. The same event_generator shape ports directly to Flask with stream_with_context, or to a Node backend calling out to a Python LangGraph service.

Filtering tool call chunks from final answer chunks

A common bug: when a node uses bind_tools(), the LLM's tool-call arguments also stream through as message chunks, and if you are not careful you will print raw JSON tool arguments to the user before the real answer arrives. Check chunk.tool_call_chunks and skip those.

for chunk, metadata in app.stream(inputs, stream_mode="messages"):
    if chunk.tool_call_chunks:
        continue  # this is a tool call being assembled, not user-facing text
    if chunk.content:
        print(chunk.content, end="", flush=True)

This one check saves you from the classic bug where a tool-calling agent's UI briefly flashes {"query": "timeout err before settling into the real response.

Choosing the right stream mode

• Use values when you want the full state snapshot after each node, good for debugging and admin views.
• Use updates when you want a clean per-node delta, good for a step-by-step progress trace.
• Use messages when you want LLM tokens as they generate, good for chat UIs, always filter by langgraph_node.
• Use astream_events when you need tool start and tool end as distinct events, or when you need the most granular lifecycle hooks available.
• Use custom with get_stream_writer() when a node does non-LLM work and you still want to report progress.

Most production chat apps end up combining exactly two of these: messages filtered to the final answer node for the token stream, and updates (or a subset of astream_events) for a lightweight "here is what the agent is doing right now" indicator above the chat bubble.

FAQ

What is the difference between stream() and astream_events() in LangGraph? stream() (and its async twin astream()) yields payloads shaped by the stream_mode you pick: full state, deltas, or message chunks. astream_events() yields a flat sequence of lifecycle events (node start, node end, chat model stream, tool start, tool end) each tagged with an event name and type, giving you finer control at the cost of more manual filtering.

Why am I seeing tool call JSON mixed into my streamed tokens? Your node is using bind_tools() and the LLM is streaming tool-call argument chunks alongside normal text chunks. Check chunk.tool_call_chunks and skip anything non-empty before printing content to the user.

Can I stream from a node that does not call an LLM? Yes. Call get_stream_writer() inside the node and invoke it with any JSON-serializable payload, then consume it with stream_mode="custom" on the caller side. This is the right approach for progress updates during loops, batch jobs, or long-running tool calls that are not themselves LLM calls.

Does stream_mode="messages" work with multiple LLM-calling nodes in the same graph? Yes, and this is exactly when you need the metadata["langgraph_node"] field. Without filtering on it, tokens from every LLM-calling node in the graph (a classifier, a summarizer, a final answer generator) all interleave in the same stream, which usually is not what you want to show a user.

Is astream_events synchronous or asynchronous only? Asynchronous only. If your code is fully synchronous, use stream() with stream_mode=["updates", "messages"] instead, which covers almost everything astream_events does for typical chat-agent use cases without requiring async/await.

How do I stream over a websocket instead of SSE? The producer side does not change: keep using app.astream(inputs, stream_mode="messages") inside an async generator. Instead of yielding data: ...\n\n strings for an SSE response, call await websocket.send_json({"token": chunk.content}) inside the loop. The graph-side streaming logic is transport-agnostic; only the delivery mechanism changes.

Why does my stream stop halfway through with no error? This usually happens when a node raises an exception partway through and the graph has no error handling around it. Wrap node bodies in try/except and either re-raise with context or write an error message into state so the stream terminates cleanly instead of silently cutting off. It can also happen if you are iterating a sync stream() generator inside an async context that gets cancelled early, in which case switching to astream() with proper await semantics fixes it.