Streaming a LangGraph Agent as OpenAI-Compatible SSE (with a Thinking Panel)

A developer built an adapter that converts LangGraph agent event streams into OpenAI-compatible Server-Sent Events, enabling tools like Open WebUI to display a 'thinking' panel showing the agent's tool calls in real time. The 90-line solution handles the strict OpenAI chunk format, including role, content, finish reason, and the required [DONE] sentinel, while also wrapping tool activity in <think> tags for a collapsible reasoning display.

In Part 1 https://dev.to/javaking1129/running-a-langgraph-react-agent-in-production-openai-compatible-api-multi-model-gateway--emi I built a LangGraph ReAct agent behind an OpenAI-compatible API and waved at one line: return StreamingResponse graph to openai sse graph, inputs, model name, config=config , media type="text/event-stream" That graph to openai sse is where the real work hides. An OpenAI client like Open WebUI doesn't want "a LangGraph run" — it wants a very specific stream of chat.completion.chunk JSON objects over Server-Sent Events, terminated by a DONE sentinel. LangGraph, meanwhile, emits its own rich event stream. This post is the adapter between the two — about 90 lines that also give you a free "thinking" panel showing the agent's tool calls as they happen. What the client expects — each token arrives as an SSE line: data: {json}\n\n , where the JSON is an OpenAI chunk: python app/api/openai compat.py def make chunk delta, model name, completion id, finish reason=None : return { "id": completion id, "chatcmpl-..." "object": "chat.completion.chunk", "created": int time.time , "model": model name, "choices": {"index": 0, "delta": delta, "finish reason": finish reason} , } The stream has a strict shape: delta = {"role": "assistant"} , delta = {"content": "..."} — one per token, finish reason = "stop" , data: DONE \n\n .Miss the DONE and the client spins forever. Skip the role chunk and some clients drop the first token. The contract is small but unforgiving. What LangGraph emits — astream events is a single async stream of typed events for everything happening inside the graph: model tokens, tool calls, node transitions. We subscribe once and translate each event we care about into chunks. python app/api/streaming.py async def graph to openai sse graph, inputs, model name, config=None : completion id = new completion id yield sse make chunk {"role": "assistant"}, model name, completion id 1 role def emit text : return sse make chunk {"content": text}, model name, completion id async for event in graph.astream events inputs, config=config, version="v2" : kind = event.get "event" if kind == "on chat model stream": chunk = event "data" "chunk" if isinstance chunk, AIMessageChunk and isinstance chunk.content, str : yield emit chunk.content 2 tokens yield sse make chunk {}, model name, completion id, finish reason="stop" 3 stop yield b"data: DONE \n\n" 4 done Three things to notice: version="v2" metadata.langgraph node and data.chunk keys don't silently move under you. on chat model stream data.chunk is an AIMessageChunk — but only when the LLM is actually streaming. Guarding with isinstance ... avoids crashing on the non-streaming events that also flow through. completion id for the whole response. sse is just the wire framing — and note ensure ascii=False , which matters the moment your tokens are Korean, Japanese, or emoji: python def sse payload : return f"data: {json.dumps payload, ensure ascii=False }\n\n".encode "utf-8" Streaming the final answer is table stakes. The interesting part of a ReAct agent is what it did before answering — which document it searched, what came back. Open WebUI renders any text wrapped in <think ...</think as a collapsible reasoning panel. So we narrate tool activity into that panel. First, label the nodes worth announcing: NODE LABELS = { "tools": "🔍 Searching the docs…", } Then open a <think block, and on the relevant events, emit human-readable progress instead of raw tokens: show thinking = bool NODE LABELS think open = False prev node = None if show thinking: yield emit "<think \n" think open = True async for event in graph.astream events inputs, config=config, version="v2" : kind = event.get "event" node = event.get "metadata" or {} .get "langgraph node", "" node entry → status line if node and node = prev node and node in NODE LABELS: yield emit f"\n{NODE LABELS node }\n" prev node = node if kind == "on tool start": yield emit f" • {event.get 'name', 'tool' } running…" continue if kind == "on tool end": output = event.get "data", {} .get "output" text = output.content if hasattr output, "content" else str output snippet = " ".join str text .split :90 collapse whitespace, clip yield emit f" ✓ {snippet}… \n" if snippet else " ✓\n" continue ... on chat model stream handled as before The on tool end output is a ToolMessage , so its text lives on .content — hence the hasattr output, "content" check before falling back to str . Collapsing whitespace and clipping to ~90 chars keeps the panel readable instead of dumping a wall of retrieved text. Closing the panel has to happen no matter how the stream ends — success, exception, or early return — so it goes in a finally : finally: if think open: yield sse make chunk {"content": "\n</think \n"}, model name, completion id The result in the UI: a collapsible "🔍 Searching the docs… ✓" panel, then the streamed answer below it. The user sees the agent reach for RAG in real time. 1. Errors belong in the stream, not in a 500. Once you've started streaming, the HTTP status is already 200 and headers are flushed — you can't switch to an error response. So catch inside the generator and emit the error as content: except Exception as exc: log.exception "stream failed" yield sse make chunk {"content": f"\n error {exc}"}, model name, completion id The user sees error ... in the chat instead of a frozen, half-rendered message. 2. Not every model streams. Some gateways/models return a single batched response with no on chat model stream events at all. If you only ever forwarded tokens, those models would yield an empty answer. Track whether any token was seen, and if not, fall back to a plain ainvoke : if not saw token: result = await graph.ainvoke inputs, config=config final = extract final text result.get "messages", yield emit final extract final text walks the message log backwards for the last non-empty AIMessage — handling both plain-string content and the list-of-blocks shape some providers return. This one guard is the difference between "streaming works on my dev model" and "works on every model behind the gateway." graph.astream events version="v2" │ ├─ on chat model stream → emit {"content": token} ├─ node entry → emit "🔍 status line" ┐ ├─ on tool start → emit "• tool running…" ├─ inside <think …</think ├─ on tool end → emit "✓ snippet…" ┘ └─ exception → emit " error …" ▼ first chunk {role} → …content chunks… → {finish reason: stop} → data: DONE The payoff from Part 1 compounds here: because the boundary is just OpenAI SSE, this thinking-panel UX shows up in any OpenAI-compatible client with zero client code. You wrote a translator, and every frontend in that ecosystem speaks it for free. Next up: persisting conversation threads with a checkpointer so the agent remembers across requests — and what that does to the streaming loop. Built with LangGraph, LangChain, and FastAPI. Part 2 of a series on running LangGraph in production — Part 1 here.