What is AI agent streaming?
AI agent streaming sends responses to the client token-by-token (or chunk-by-chunk) as the LLM generates them, instead of waiting for the full response to complete before sending anything. From the user's perspective, the text appears progressively on screen — the same experience as ChatGPT's typewriter effect — dramatically reducing perceived latency.
Streaming matters because LLMs generate tokens sequentially. A 500-token response at typical generation speed takes 5–10 seconds without streaming. With streaming, the user sees the first tokens within 200–500ms.
How streaming works
LLM generates tokens: "The" → " weather" → " in" → " Tokyo" → "..."
│ │ │ │
▼ ▼ ▼ ▼
SSE stream to client: chunk1 chunk2 chunk3 chunk4
│ │ │ │
▼ ▼ ▼ ▼
UI renders: "The" "The weather" "The weather in" ...
The server sends each token as an SSE event the moment it is generated. The client receives and renders tokens in real time.
SSE vs WebSockets for streaming
| SSE (Server-Sent Events) | WebSockets | |
|---|---|---|
| Direction | Server → Client only | Bidirectional |
| Protocol | HTTP/1.1 | WS upgrade |
| Browser support | Native | Native |
| Reconnection | Automatic | Manual |
| Best for | Token streaming | Bidirectional agent interaction |
AgentFlow supports both: POST /v1/graph/stream for SSE token streaming, and WS /v1/graph/ws for bidirectional WebSocket connections (used for remote tool calls where the client executes tools on behalf of the agent).
Streaming from Python with AgentFlow
AgentFlow's built-in API server exposes SSE streaming with zero configuration:
agentflow api # starts REST + SSE + WebSocket server
The streaming endpoint:
curl -N -X POST http://localhost:8000/v1/graph/stream \
-H "Content-Type: application/json" \
-d '{
"messages": [{"role": "user", "content": [{"type": "text", "text": "Explain neural networks"}]}],
"config": {"thread_id": "demo-1"}
}'
Each SSE event arrives as:
data: {"event": "token", "content": "Neural"}
data: {"event": "token", "content": " networks"}
data: {"event": "token", "content": " are"}
...
data: {"event": "end"}
Consuming the stream from TypeScript
The @10xscale/agentflow-client npm package handles SSE parsing automatically:
import { AgentFlowClient, Message, StreamEventType } from "@10xscale/agentflow-client";
const client = new AgentFlowClient({ baseUrl: "http://localhost:8000" });
for await (const chunk of client.stream(
[Message.text_message("Explain neural networks")],
{ config: { thread_id: "demo-1" } }
)) {
if (chunk.event === StreamEventType.MESSAGE && chunk.message) {
process.stdout.write(chunk.message.text());
}
}
Streaming in a React UI
The client package exposes a single entry point and ships no React bindings, so wire client.stream() into your own state. The generator is an ordinary AsyncGenerator, so a for await inside an effect or an event handler is all you need:
import { useCallback, useState } from "react";
import { AgentFlowClient, Message, StreamEventType } from "@10xscale/agentflow-client";
const client = new AgentFlowClient({ baseUrl: "http://localhost:8000" });
function Chat() {
const [text, setText] = useState("");
const [isStreaming, setIsStreaming] = useState(false);
const send = useCallback(async (prompt: string) => {
setText("");
setIsStreaming(true);
try {
const stream = client.stream([Message.text_message(prompt)], {
config: { thread_id: "user-session-123" },
});
for await (const chunk of stream) {
if (chunk.event === StreamEventType.MESSAGE && chunk.message?.delta) {
setText((previous) => previous + chunk.message!.text());
}
}
} finally {
setIsStreaming(false);
}
}, []);
return (
<div>
<p>{text}</p>
{isStreaming && <span>...</span>}
<button onClick={() => send("Tell me more")}>Send</button>
</div>
);
}
chunk.message.delta === true marks an incremental token; the same message arrives once more with delta === false when it is complete, so filter on delta to avoid printing the answer twice.
Streaming with tool calls
When an agent calls a tool during streaming, the stream pauses while the tool executes, then resumes with the next LLM response. AgentFlow streams tool-call events so the UI can show "calling search_web..." in real time:
{"event": "message", "message": {"role": "assistant", "content": [{"type": "tool_call", "name": "search_web", "args": {"query": "latest AI news"}}]}}
{"event": "message", "message": {"role": "tool", "content": [{"type": "tool_result", "call_id": "...", "output": "..."}]}}
{"event": "message", "message": {"role": "assistant", "delta": true, "content": [{"type": "text", "text": "Based"}]}}
{"event": "message", "message": {"role": "assistant", "delta": true, "content": [{"type": "text", "text": " on"}]}}
The server writes one JSON object per line (NDJSON). Tool activity is not a separate event type: it arrives as ordinary message chunks whose content blocks are tool_call and tool_result, so a UI showing "calling search_web..." watches for those block types.
Streaming in the graph
You can also stream directly from a compiled graph in Python without the API server:
from agentflow.core.state import Message
async def stream_agent():
async for chunk in app.astream(
{"messages": [Message.text_message("Tell me about Python")]},
config={"thread_id": "demo"},
):
if chunk.event == "message" and chunk.message:
print(chunk.message.text(), end="", flush=True)
astream yields StreamChunk objects, not dicts. To push progress out of a tool while it runs, take a StreamEmitter parameter in the tool; see StreamEmitter for the injection rules and from agentflow.core.state.stream_emitter import StreamEmitter as the import path.