Streaming LangChain agent output to a React frontend

Streaming a langchain agent streaming react frontend pipeline is one of those tasks that looks simple in tutorials but breaks in production when tool calls interleave with final output, connections drop, or backpressure builds. This guide walks through a complete, runnable implementation: a FastAPI backend that streams agent tokens and tool events via Server-Sent Events, and a React frontend that renders them incrementally without blocking the UI.

Step 1: Set up the backend dependencies

Create a virtual environment and install the minimal set. We use langchain-core for the callback interface, langchain-openai for the model, and fastapi with uvicorn for the server. sse-starlette handles the SSE response format.

python -m venv .venv
source .venv/bin/activate
pip install "langchain-core>=0.2" "langchain-openai>=0.1" fastapi uvicorn sse-starlette pydantic-settings python-dotenv

If you route through n4n.ai, swap langchain-openai for langchain-openai pointed at the n4n.ai base URL — the streaming interface is identical.

Step 2: Define the streaming callback handler

LangChain’s BaseCallbackHandler lets you intercept tokens, tool starts, tool ends, and errors. We push each event into an asyncio.Queue so the SSE generator can yield them in order.

# backend/streaming.py
import asyncio
import json
from typing import Any, Dict, List, Optional
from uuid import UUID

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult


class SSECallbackHandler(BaseCallbackHandler):
    """Queues streaming events for SSE delivery."""

    def __init__(self, queue: asyncio.Queue):
        self.queue = queue
        self.current_tool: Optional[str] = None

    def _put(self, event: Dict[str, Any]) -> None:
        try:
            self.queue.put_nowait(event)
        except asyncio.QueueFull:
            pass  # Backpressure: drop if client can't keep up

    # LLM token streaming
    async def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        self._put({"type": "token", "data": token})

    # Tool invocation lifecycle
    async def on_tool_start(
        self, serialized: Dict[str, Any], input_str: str, **kwargs: Any
    ) -> None:
        self.current_tool = serialized.get("name", "unknown")
        self._put({
            "type": "tool_start",
            "data": {"name": self.current_tool, "input": input_str}
        })

    async def on_tool_end(self, output: str, **kwargs: Any) -> None:
        self._put({
            "type": "tool_end",
            "data": {"name": self.current_tool, "output": output}
        })
        self.current_tool = None

    async def on_tool_error(self, error: BaseException, **kwargs: Any) -> None:
        self._put({
            "type": "tool_error",
            "data": {"name": self.current_tool, "error": str(error)}
        })
        self.current_tool = None

    # Optional: capture final LLM result for debugging
    async def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        self._put({"type": "llm_end", "data": {}})

The queue decouples agent execution from network I/O. If the client disconnects, the generator stops consuming and the queue fills — we drop events rather than block the agent.

Step 3: Build the agent executor with streaming enabled

Use create_react_agent (or your preferred agent type) and pass the callback handler at invoke time. The key is stream_mode="values" on the executor, which yields intermediate state — but we only need the callback queue for token-level streaming.

# backend/agent.py
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

from streaming import SSECallbackHandler


@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    # Replace with real API call
    return f"The weather in {city} is sunny, 72°F."


@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    # Replace with real search API
    return f"Search results for: {query}"


TOOLS = [get_weather, search_web]

PROMPT = PromptTemplate.from_template("""You are a helpful assistant with access to tools.

{tools}

Use the following format:

Question: {input}
Thought: {agent_scratchpad}""")


def build_agent() -> AgentExecutor:
    llm = ChatOpenAI(
        model="gpt-4o-mini",
        temperature=0,
        streaming=True,  # Critical: enables token callbacks
    )
    agent = create_react_agent(llm, TOOLS, PROMPT)
    return AgentExecutor(
        agent=agent,
        tools=TOOLS,
        verbose=True,
        handle_parsing_errors=True,
        max_iterations=5,
    )

streaming=True on the ChatOpenAI instance is what triggers on_llm_new_token callbacks. Without it, you only get the final message.

Step 4: Create the FastAPI SSE endpoint

The endpoint accepts a question, spins up a callback queue, runs the agent in a background task, and streams events from the queue until the agent finishes or the client disconnects.

# backend/main.py
import asyncio
import json
from contextlib import asynccontextmanager
from typing import AsyncGenerator

from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from sse_starlette.sse import EventSourceResponse

from agent import build_agent
from streaming import SSECallbackHandler


app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],  # Vite default
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class ChatRequest(BaseModel):
    question: str


agent_executor = build_agent()


async def event_generator(
    request: Request, question: str
) -> AsyncGenerator[dict, None]:
    queue: asyncio.Queue = asyncio.Queue(maxsize=100)
    handler = SSECallbackHandler(queue)

    # Run agent in background
    task = asyncio.create_task(
        agent_executor.ainvoke(
            {"input": question},
            config={"callbacks": [handler]},
        )
    )

    try:
        while True:
            # Check for client disconnect
            if await request.is_disconnected():
                task.cancel()
                break

            try:
                event = await asyncio.wait_for(queue.get(), timeout=0.5)
                yield {"event": event["type"], "data": json.dumps(event["data"])}
            except asyncio.TimeoutError:
                # Heartbeat to keep connection alive
                yield {"event": "heartbeat", "data": "{}"}
                continue

            # Check if agent completed
            if task.done():
                # Drain remaining queue
                while not queue.empty():
                    event = queue.get_nowait()
                    yield {"event": event["type"], "data": json.dumps(event["data"])}
                break

    except asyncio.CancelledError:
        task.cancel()
        raise
    except Exception as e:
        yield {"event": "error", "data": json.dumps({"message": str(e)})}
    finally:
        if not task.done():
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass


@app.post("/api/chat/stream")
async def chat_stream(request: Request, body: ChatRequest):
    if not body.question.strip():
        raise HTTPException(400, "Question cannot be empty")
    return EventSourceResponse(event_generator(request, body.question))


@app.get("/health")
async def health():
    return {"status": "ok"}

Run it:

uvicorn main:app --reload --port 8000

Verify with curl:

curl -N -H "Content-Type: application/json" \
  -d '{"question": "What is the weather in Tokyo?"}' \
  http://localhost:8000/api/chat/stream

You should see a stream of event: token, event: tool_start, event: tool_end, and event: heartbeat lines.

Step 5: Build the React frontend hook

On the client, a custom hook manages the EventSource connection, parses events, and exposes an array of rendered messages. We distinguish between assistant tokens, tool calls, and tool results so the UI can show each appropriately.

// frontend/src/hooks/useAgentStream.ts
import { useState, useCallback, useRef, useEffect } from "react";

export type StreamEvent =
  | { type: "token"; data: string }
  | { type: "tool_start"; data: { name: string; input: string } }
  | { type: "tool_end"; data: { name: string; output: string } }
  | { type: "tool_error"; data: { name: string; error: string } }
  | { type: "error"; data: { message: string } }
  | { type: "heartbeat"; data: Record<string, never> }
  | { type: "llm_end"; data: Record<string, never> };

export type Message =
  | { role: "user"; content: string }
  | { role: "assistant"; content: string }
  | { role: "tool"; name: string; input: string; output?: string; error?: string };

interface UseAgentStreamResult {
  messages: Message[];
  isStreaming: boolean;
  error: string | null;
  sendMessage: (question: string) => Promise<void>;
  clear: () => void;
}

export function useAgentStream(apiBase: string = "/api"): UseAgentStreamResult {
  const [messages, setMessages] = useState<Message[]>([]);
  const [isStreaming, setIsStreaming] = useState(false);
  const [error, setError] = useState<string | null>(null);
  const eventSourceRef = useRef<EventSource | null>(null);
  const currentToolRef = useRef<{ name: string; input: string } | null>(null);
  const assistantBufferRef = useRef<string>("");

  const clear = useCallback(() => {
    setMessages([]);
    setError(null);
    assistantBufferRef.current = "";
    currentToolRef.current = null;
  }, []);

  const appendAssistantToken = useCallback((token: string) => {
    assistantBufferRef.current += token;
    setMessages((prev) => {
      const last = prev[prev.length - 1];
      if (last && last.role === "assistant") {
        return [...prev.slice(0, -1), { ...last, content: assistantBufferRef.current }];
      }
      return [...prev, { role: "assistant", content: assistantBufferRef.current }];
    });
  }, []);

  const handleEvent = useCallback((event: MessageEvent) => {
    try {
      const parsed: StreamEvent = JSON.parse(event.data);
      switch (parsed.type) {
        case "token":
          appendAssistantToken(parsed.data);
          break;
        case "tool_start":
          currentToolRef.current = { name: parsed.data.name, input: parsed.data.input };
          setMessages((prev) => [
            ...prev,
            { role: "tool", name: parsed.data.name, input: parsed.data.input },
          ]);
          break;
        case "tool_end":
          if (currentToolRef.current?.name === parsed.data.name) {
            setMessages((prev) =>
              prev.map((msg, i) =>
                i === prev.length - 1 && msg.role === "tool" && msg.name === parsed.data.name
                  ? { ...msg, output: parsed.data.output }
                  : msg
              )
            );
            currentToolRef.current = null;
          }
          break;
        case "tool_error":
          if (currentToolRef.current?.name === parsed.data.name) {
            setMessages((prev) =>
              prev.map((msg, i) =>
                i === prev.length - 1 && msg.role === "tool" && msg.name === parsed.data.name
                  ? { ...msg, error: parsed.data.error }
                  : msg
              )
            );
            currentToolRef.current = null;
          }
          break;
        case "error":
          setError(parsed.data.message);
          setIsStreaming(false);
          break;
        case "llm_end":
          // Agent finished; finalize
          setIsStreaming(false);
          assistantBufferRef.current = "";
          break;
        case "heartbeat":
          // No-op, keeps connection alive
          break;
      }
    } catch (e) {
      console.error("Failed to parse SSE event:", e);
    }
  }, [appendAssistantToken]);

  const sendMessage = useCallback(
    async (question: string) => {
      if (isStreaming) return;
      setError(null);
      setIsStreaming(true);
      setMessages((prev) => [...prev, { role: "user", content: question }]);
      assistantBufferRef.current = "";

      const es = new EventSource(`${apiBase}/chat/stream`, {
        // EventSource doesn't support POST body; we use a workaround below
      });
      eventSourceRef.current = es;

      // EventSource is GET-only. For POST with body, use fetch + ReadableStream instead.
      // This is a known limitation; see the fetch-based implementation below.
    },
    [apiBase, isStreaming]
  );

  // Cleanup on unmount
  useEffect(() => {
    return () => {
      eventSourceRef.current?.close();
    };
  }, []);

  return { messages, isStreaming, error, sendMessage, clear };
}

Important: EventSource only supports GET. For a POST body (the question), you have two options:

1. Short questions: Encode in query string (/chat/stream?question=...) — simple but length-limited.
2. Production: Use fetch with ReadableStream (shown next).

Step 6: Replace EventSource with fetch + ReadableStream

This version works for any payload size and gives you full control over reconnection logic.

// frontend/src/hooks/useAgentStream.ts (replace sendMessage)
const sendMessage = useCallback(
  async (question: string) => {
    if (isStreaming) return;
    setError(null);
    setIsStreaming(true);
    setMessages((prev) => [...prev, { role: "user", content: question }]);
    assistantBufferRef.current = "";

    try {
      const response = await fetch(`${apiBase}/chat/stream`, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ question }),
      });

      if (!response.ok) {
        throw new Error(`HTTP ${response.status}`);
      }

      const reader = response.body?.getReader();
      if (!reader) throw new Error("No response body");

      const decoder = new TextDecoder();
      let buffer = "";

      while (true) {
        const { done, value } = 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.trim()) continue;
          // SSE format: "event: type\ndata: {...}"
          const eventMatch = line.match(/^event: (\w+)/m);
          const dataMatch = line.match(/^data: (.+)/m);
          if (eventMatch && dataMatch) {
            const event: StreamEvent = {
              type: eventMatch[1] as StreamEvent["type"],
              data: JSON.parse(dataMatch[1]),
            };
            // Reuse handleEvent logic
            const fakeEvent = { data: JSON.stringify(event) } as MessageEvent;
            handleEvent(fakeEvent);
          }
        }
      }

      setIsStreaming(false);
      assistantBufferRef.current = "";
    } catch (e) {
      setError(e instanceof Error ? e.message : "Stream failed");
      setIsStreaming(false);
    }
  },
  [apiBase, isStreaming, handleEvent]
);

This parser handles the SSE framing (event: / data: pairs separated by blank lines). It’s more verbose than EventSource but works with POST and lets you implement retry/backoff.

Step 7: Render the message list

A minimal component that distinguishes message types visually. Tool calls show as collapsible panels; assistant tokens append in place.

// frontend/src/components/ChatView.tsx
import { useState } from "react";
import { Message } from "../hooks/useAgentStream";

interface ChatViewProps {
  messages: Message[];
  isStreaming: boolean;
  onSend: (question: string) => Promise<void>;
  disabled?: boolean;
}

export function ChatView({ messages, isStreaming, onSend, disabled }: ChatViewProps) {
  const [input, setInput] = useState("");

  const handleSubmit = async (e: React.FormEvent) => {
    e.preventDefault();
    if (!input.trim() || isStreaming || disabled) return;
    const q = input;
    setInput("");
    await onSend(q);
  };

  return (
    <div style={{ display: "flex", flexDirection: "column", height: "100vh" }}>
      <div style={{ flex: 1, overflow: "auto", padding: "1rem", display: "flex", flexDirection: "column", gap: "0.75rem" }}>
        {messages.map((msg, i) => (
          <MessageBubble key={i} message={msg} />
        ))}
        {isStreaming && <div className="typing-indicator">▌</div>}
      </div>
      <form onSubmit={handleSubmit} style={{ padding: "1rem", borderTop: "1px solid #eee" }}>
        <input
          value={input}
          onChange={(e) => setInput(e.target.value)}
          placeholder={isStreaming ? "Waiting for response..." : "Ask a question..."}
          disabled={isStreaming || disabled}
          style={{ width: "100%", padding: "0.5rem", fontSize: "1rem" }}
        />
      </form>
    </div>
  );
}

function MessageBubble({ message }: { message: Message }) {
  const isTool = message.role === "tool";
  const isAssistant = message.role === "assistant";
  const isUser = message.role === "user";

  if (isTool) {
    const [expanded, setExpanded] = useState(false);
    return (
      <details open={expanded} style={{ border: "1px solid #ddd", borderRadius: "4px", padding: "0.5rem" }}>
        <summary style={{ cursor: "pointer", fontWeight: 600 }}>
          🔧 Tool: {message.name}
        </summary>
        <pre style={{ margin: "0.5rem 0", whiteSpace: "pre-wrap" }}>{message.input}</pre>
        {message.output && (
          <div style={{ color: "green" }}>
            <strong>Output:</strong>
            <pre style={{ margin: "0.5rem 0", whiteSpace: "pre-wrap" }}>{message.output}</pre>
          </div>
        )}
        {message.error && (
          <div style={{ color: "red" }}>
            <strong>Error:</strong> {message.error}
          </div>
        )}
      </details>
    );
  }

  const align = isUser ? "flex-end" : "flex-start";
  const bg = isUser ? "#007bff" : "#f1f3f4";
  const color = isUser ? "white" : "black";

  return (
    <div style={{ display: "flex", justifyContent: align }}>
      <div
        style={{
          maxWidth: "80%",
          padding: "0.75rem 1rem",
          borderRadius: "12px",
          background: bg,
          color,
          whiteSpace: "pre-wrap",
        }}
      >
        {message.content}
      </div>
    </div>
  );
}

Wire it up in App.tsx:

// frontend/src/App.tsx
import { useAgentStream } from "./hooks/useAgentStream";
import { ChatView } from "./components/ChatView";

function App() {
  const { messages, isStreaming, error, sendMessage, clear } = useAgentStream(
    import.meta.env.VITE_API_BASE || "http://localhost:8000"
  );

  return (
    <div style={{ fontFamily: "system-ui, sans-serif" }}>
      <header style={{ padding: "1rem", borderBottom: "1px solid #eee" }}>
        <h1 style={{ margin: 0, fontSize: "1.25rem" }}>LangChain Agent Stream</h1>
        {error && <div style={{ color: "red", marginTop: "0.5rem" }}>{error}</div>}
      </header>
      <ChatView
        messages={messages}
        isStreaming={isStreaming}
        onSend={sendMessage}
        disabled={isStreaming}
      />
    </div>
  );
}

export default App;

Run the frontend:

cd frontend
npm create vite@latest . -- --template react-ts
npm install
npm run dev

Visit http://localhost:5173, type “What’s the weather in Tokyo?”, and watch tokens appear character-by-character while the tool call renders as a collapsible panel.

Step 8: Handle reconnection and backpressure

Production traffic needs resilience. Add a retry wrapper around fetch with exponential backoff, and respect the Retry-After header if your gateway returns 429.

// frontend/src/lib/streamFetch.ts
async function* streamWithRetry(
  url: string,
  options: RequestInit,
  maxRetries = 3
): AsyncGenerator<Uint8Array, void, unknown> {
  let attempt = 0;
  while (true) {
    try {
      const response = await fetch(url, options);
      if (response.status === 429) {
        const retryAfter = response.headers.get("Retry-After");
        const delay = retryAfter ? parseInt(retryAfter, 10) * 1000 : Math.min(1000 * 2 ** attempt, 30000);
        await new Promise((r) => setTimeout(r, delay));
        attempt++;
        if (attempt > maxRetries) throw new Error("Max retries exceeded");
        continue;
      }
      if (!response.ok || !response.body) {
        throw new Error(`HTTP ${response.status}`);
      }
      const reader = response.body.getReader();
      while (true) {
        const { done, value } = await reader.read();
        if (done) return;
        yield value;
      }
    } catch (e) {
      if (attempt >= maxRetries) throw e;
      attempt++;
      await new Promise((r) => setTimeout(r, 500 * attempt));
    }
  }
}

Replace the fetch call in useAgentStream with this generator. The hook stays unchanged — only the transport layer gets smarter.

Step 9: Verify end-to-end

Run both servers and test these scenarios:

1. Simple question — “What is 2+2?” — tokens stream, no tool calls.
2. Tool invocation — “Weather in London” — tool_start renders, tool_end populates output, final answer streams.
3. Multi-step — “Compare weather in Paris and Berlin” — two tool calls sequence correctly.
4. Network interrupt — Kill the backend mid-stream; frontend shows error toast, retry button works.
5. Long response — Ask for a detailed explanation; verify no UI freeze (tokens append via requestAnimationFrame batching if needed).

Check the browser Network tab: the /chat/stream request should stay open with Content-Type: text/event-stream and show incremental frames.

Step 10: Observability hooks (optional but recommended)

Add structured logging to the callback handler so you can trace latency per token, tool duration, and queue depth.

# backend/streaming.py (add to SSECallbackHandler)
import time
import structlog

logger = structlog.get_logger()

class SSECallbackHandler(BaseCallbackHandler):
    def __init__(self, queue: asyncio.Queue, request_id: str):
        self.queue = queue
        self.request_id = request_id
        self.token_count = 0
        self.start_time = time.perf_counter()

    def _put(self, event: Dict[str, Any]) -> None:
        event["request_id"] = self.request_id
        try:
            self.queue.put_nowait(event)
        except asyncio.QueueFull:
            logger.warning("sse_queue_full", request_id=self.request_id)

    async def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        self.token_count += 1
        if self.token_count % 50 == 0:
            logger.info("stream_progress", request_id=self.request_id, tokens=self.token_count)
        self._put({"type": "token", "data": token})

    async def on_tool_end(self, output: str, **kwargs: Any) -> None:
        duration = time.perf_counter() - self.start_time
        logger.info("tool_completed", request_id=self.request_id, tool=self.current_tool, duration_ms=duration*1000)
        self._put({"type": "tool_end", "data": {"name": self.current_tool, "output": output}})
        self.current_tool = None

Pass request_id from the endpoint (generate via uuid4()). This gives you per-request timelines in your log aggregator.

You now have a complete langchain agent streaming react frontend stack: FastAPI + SSE on the backend, a resilient fetch-based hook on the frontend, and clear separation between token streaming, tool lifecycle, and UI rendering. The same pattern works for any LangChain-compatible agent — just swap the executor and tools.