TL;DR:

  • Streaming tokens and agent events over SSE is simpler than WebSockets for most agent use cases — use WebSockets only when you need bidirectional communication.
  • Send intermediate events (tool calls, tool results, thinking steps) not just final tokens — this is what separates a good agent UX from a bad one.
  • Buffer and debounce token updates client-side to avoid excessive DOM thrashing on fast completions.

Waiting 30 seconds for a silent loading spinner to resolve into a wall of text is a poor user experience for AI agents. Users tolerate latency much better when they can see the agent working: which tools it is calling, what it found, how its thinking is progressing. Streaming is how you make that happen, and the implementation choices you make early determine how much you can show.

The Two Transport Options

Server-Sent Events (SSE) is a simple HTTP-based protocol for pushing a stream of events from server to client. The server holds the connection open and writes newline-delimited event data. The browser’s EventSource API handles reconnection automatically. For AI agent streaming, SSE covers most use cases and is easier to implement, deploy, and debug than WebSockets.

WebSockets provide a full-duplex connection: both client and server can send messages at any time. This matters for AI agents only when users need to interrupt or redirect the agent mid-run. Sending a “stop” signal, updating instructions while the agent runs, or injecting new user input mid-task all require WebSockets (or a separate HTTP endpoint for control messages alongside SSE for output).

For a simple question-answer agent, use SSE. For a coding agent where the user might want to redirect mid-task, WebSockets or a hybrid approach makes sense.

What to Stream

Most streaming tutorials show only token streaming: the LLM’s output word by word. For agents, that is the floor, not the ceiling. Users benefit from seeing the full agent execution trace:

  • Thinking / planning step: when the model reasons before acting (if your model supports visible reasoning)
  • Tool call initiated: which tool, with what arguments
  • Tool result received: the data the tool returned, summarised or full
  • Model response tokens: the final response as it generates

Streaming all of these gives users a live view of agent work. A user who sees “Searching your calendar… found 3 conflicts… checking availability for next week…” tolerates the latency because they understand what is happening. A user who stares at a spinner for 20 seconds does not.

Server Implementation: Python with FastAPI and SSE

Using Anthropic’s Python SDK with streaming enabled:

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

app = FastAPI()
client = anthropic.Anthropic()

async def run_agent_stream(user_message: str):
    tools = [
        {
            "name": "search_docs",
            "description": "Search internal documentation",
            "input_schema": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"]
            }
        }
    ]

    with client.messages.stream(
        model="claude-opus-4-7",
        max_tokens=4096,
        tools=tools,
        messages=[{"role": "user", "content": user_message}]
    ) as stream:
        for event in stream:
            if hasattr(event, 'type'):
                if event.type == 'content_block_start':
                    if hasattr(event.content_block, 'type'):
                        if event.content_block.type == 'tool_use':
                            yield f"data: {json.dumps({'type': 'tool_call', 'tool': event.content_block.name})}\n\n"
                elif event.type == 'content_block_delta':
                    if hasattr(event.delta, 'text'):
                        yield f"data: {json.dumps({'type': 'token', 'text': event.delta.text})}\n\n"
                elif event.type == 'message_stop':
                    yield f"data: {json.dumps({'type': 'done'})}\n\n"

@app.get("/stream")
async def stream_endpoint(message: str):
    return StreamingResponse(
        run_agent_stream(message),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
    )

The X-Accel-Buffering: no header is important when running behind Nginx — without it, Nginx buffers the SSE stream and users see nothing until the buffer fills.

Client Implementation: Consuming the SSE Stream

function connectToAgentStream(message: string) {
    // EventSource doesn't support POST, so use fetch with streaming
    fetch(`/stream?message=${encodeURIComponent(message)}`)
        .then(response => {
            const reader = response.body!.getReader();
            const decoder = new TextDecoder();
            let buffer = '';

            function processChunk({ done, value }: ReadableStreamReadResult<Uint8Array>) {
                if (done) return;

                buffer += decoder.decode(value, { stream: true });
                const lines = buffer.split('\n');
                buffer = lines.pop() || '';

                for (const line of lines) {
                    if (line.startsWith('data: ')) {
                        const data = JSON.parse(line.slice(6));
                        handleAgentEvent(data);
                    }
                }

                reader.read().then(processChunk);
            }

            reader.read().then(processChunk);
        });
}

function handleAgentEvent(event: { type: string; text?: string; tool?: string }) {
    if (event.type === 'tool_call') {
        appendToUI(`Calling tool: ${event.tool}...`);
    } else if (event.type === 'token') {
        appendTokenToResponse(event.text!);
    } else if (event.type === 'done') {
        markResponseComplete();
    }
}

Debouncing DOM Updates

For fast-generating models, individual token events arrive faster than the browser renders. Batching token updates into 50ms windows prevents layout thrashing:

let pendingTokens = '';
let flushTimer: ReturnType<typeof setTimeout> | null = null;

function appendTokenToResponse(token: string) {
    pendingTokens += token;
    if (!flushTimer) {
        flushTimer = setTimeout(() => {
            document.getElementById('response')!.textContent += pendingTokens;
            pendingTokens = '';
            flushTimer = null;
        }, 50);
    }
}

This keeps the UI responsive without showing choppy token-by-token updates.

Framework Options

Vercel AI SDK handles SSE and streaming automatically for React and Next.js apps. The useChat hook manages stream state, with built-in support for tool call events. If you are building on Next.js, this is the lowest-effort path.

LangChain has streaming callbacks that fire on token generation and chain events. Combined with FastAPI, this covers Python agent stacks.

LangGraph supports streaming via graph.stream() which yields state updates at each node transition — useful for showing users which step of a multi-step agent workflow is currently executing.

Multi-Turn Agents and Interruption

For long-running agents with tool use loops, store the stream ID server-side and expose a separate /interrupt endpoint. When the client sends an interrupt, set a flag that terminates the next iteration of the tool loop. WebSockets simplify this because the interrupt can flow over the same connection, but it is achievable with HTTP control endpoints alongside SSE if you prefer to avoid WebSocket complexity.

Streaming is not a polish feature for AI agents. It is the mechanism by which users understand and trust what the agent is doing on their behalf.

References