Streaming Chat APIs

This content was generated with the assistance of AI and should be verified against the official documentation of each tool and framework before being relied on in production.

This section’s bibliography lists the reference material consulted while preparing these pages.

This page was written on 2026-10-01 against FastAPI 0.142.2 (fastapi.sse), langchain 1.4.3 and langgraph 1.2.12 on Python 3.12, Spring AI 2.0.1 on Spring Boot 4.1.1 with Java 21, and Express 5.2.1 on Node.js (see the version baseline). The snippets are illustrative: they were written from the official documentation of each framework and were not compiled or run by this repository. The SSE fields and defaults of FastAPI were checked against its documentation on the date above.

Serving RAG as an API designs the public /v1/ask contract: request and response fields, OpenAPI, idempotency, rate limiting and versioning. This page does not repeat it. It takes the channel adapter view: the HTTP and WebSocket endpoints below are two of the adapters in front of the channel-agnostic core, and the interesting questions are the ones every streaming adapter shares — which transport to pick, how to cancel a generation nobody is reading, how a dropped connection resumes, what happens when the client is slower than the model, what proxies do to the stream and how a browser authenticates a connection that cannot carry an Authorization header.

The core used throughout is the DocsAssistantCore (Python) and AssistantCore (Java) defined on Channel-Agnostic Architecture; this page reuses it unchanged:

ask(conversationId, userId, text) -> stream of { token | citation | done }

In Python it is an async generator of the frozen dataclasses Token(text), Citation(marker, chunk_id, title, url) and Done(conversation_id, answer, citations), built on the persistent LangChain agent of LangChain: Streaming; in Java it is a Flux of the sealed interface CoreEvent with the same three records, built on the Spring AI ChatClient of Spring AI: Streaming & Reactive. Both cores resolve the caller’s ACLs from userId and pass them to the retriever, so no adapter can forget the access-control filter. An adapter never touches the agent or the ChatClient; it only translates those events to bytes on the wire and the wire’s signals (disconnect, cancel, reconnect) back to the core.

SSE, WebSocket or NDJSON

All three carry an ordered sequence of small JSON messages from the server to the client over a long-lived HTTP response or connection. They differ in what else they give you and in what the infrastructure in between tolerates.

Server-sent events (SSE) WebSocket NDJSON over a streaming response

Direction

Server to client; the client talks with ordinary requests

Full duplex on one connection

Server to client; the client talks with ordinary requests

Wire format

text/event-stream: event:, data:, id:, retry: fields and : comment lines, events separated by a blank line

Frames (text or binary) defined by RFC 6455; the message schema is yours

application/x-ndjson: one JSON document per line; no standard media-type registration, so clients agree by convention

Browser API

EventSource (GET only, no custom headers) or fetch with a stream reader

WebSocket (no custom headers on the handshake)

fetch with response.body.getReader(); any HTTP method and headers

Reconnect

Built in: EventSource retries and sends Last-Event-ID

None; your code reconnects and resumes

None; your code reconnects and resumes

Cancel a generation

Close the connection (the server sees a disconnect) or send a separate DELETE/POST

Send a cancel message on the same connection

Abort the fetch, which closes the connection

Proxies and CDNs

Plain HTTP; works wherever response buffering is off; HTTP/2 multiplexing avoids the 6-connections-per-host limit of HTTP/1.1

Needs an Upgrade-aware proxy and idle-timeout tuning; some corporate proxies break it

Plain HTTP, with the same buffering caveats as SSE

Best fit

Chat answers: one question, one streamed answer, citations at the end

Typing indicators, barge-in, several turns in flight, voice-adjacent control messages

Service-to-service streaming and fetch-based clients that need POST with a body

For a RAG chat window the default is SSE: it is plain HTTP, goes through the same authentication, rate limiting and logging as every other route, and reconnects by itself. The browser’s EventSource only issues GET requests and cannot set headers, which pushes real applications to one of two shapes: fetch plus a stream parser (a POST with the question in the body, as in the widget of Serving RAG as an API), or the two-step shape shown in Reconnecting with Last-Event-ID (POST to start a turn, GET to stream its events). Choose WebSocket when the client genuinely needs to talk while the server streams; see also the .NET equivalent in SignalR, which negotiates WebSocket, SSE and long polling for you, and the Quarkus take in Quarkus WebSockets. Choose NDJSON for machine clients that want an uncomplicated line protocol. GraphQL subscriptions (GraphQL Subscriptions) and brokers (Messaging Protocols) are the heavier alternatives when the stream must fan out to many subscribers rather than to the single asker.

The shape of one turn

sequenceDiagram autonumber participant B as Browser participant P as Reverse proxy participant A as HTTP adapter participant C as Channel-agnostic core participant M as LLM and retriever B->>P: POST /v1/turns (Authorization: Bearer ...) P->>A: forward (buffering off) A->>A: authenticate, derive userId, check conversation owner A-->>B: 202 {turnId} B->>P: GET /v1/turns/{turnId}/events P->>A: forward A->>C: ask(conversationId, userId, text) C->>M: retrieve, generate loop each token M-->>C: chunk C-->>A: token A-->>B: id: 7 / event: token / data: {"text": "..."} end Note over B,A: connection drops B->>P: GET /events (Last-Event-ID: 7) P->>A: forward A-->>B: replay events after 7, then live events C-->>A: citation, done A-->>B: event: citation (one per source), event: done B->>A: close (or DELETE /v1/turns/{turnId}) A->>C: cancel C->>M: abort generation

FastAPI: EventSourceResponse over LangChain astream

FastAPI 0.142.2 ships SSE support in fastapi.sse: a route declared with response_class=EventSourceResponse that is an async generator yielding ServerSentEvent objects. The class has data (JSON-encoded), raw_data (sent as-is), event, id, retry and comment fields. FastAPI sends a keep-alive comment every 15 seconds when nothing else is flowing and sets Cache-Control: no-cache and X-Accel-Buffering: no on the response by itself. The core is the DocsAssistantCore of Channel-Agnostic Architecture, built once at start-up and stored on app.state.core: its ask already turns the LangChain astream output of the agent into the three channel-neutral events, runs the agent with context=caller (the Caller resolved through caller_for(user_id), which carries the ACLs of Access Control & Privacy) and keys memory with the thread_id convention of Conversational RAG & Memory. The adapter only has to choose event names.

The HTTP adapter is then a thin generator. Everything that can fail with a status code — the token, the conversation ownership — belongs in dependencies, which FastAPI resolves before the first byte; an exception inside the generator can only become an error event on a response that already said 200.

docsassistant/http_adapter.py — the SSE adapter (Python, FastAPI 0.142.2)
from collections.abc import AsyncIterable
from typing import Annotated

from fastapi import Depends, FastAPI, Request
from fastapi.sse import EventSourceResponse, ServerSentEvent

from docsassistant.api import AskRequest, current_caller, owned_conversation   # from Serving RAG as an API
from docsassistant.core import Citation, Done, Token       # DocsAssistantCore events, see Channel-Agnostic Architecture
from docsassistant.security import Caller

app = FastAPI()


@app.post("/v1/chat/stream", response_class=EventSourceResponse)
async def chat_stream(req: AskRequest, request: Request,
                      caller: Annotated[Caller, Depends(current_caller)],
                      conversation_id: Annotated[str, Depends(owned_conversation)]) -> AsyncIterable[ServerSentEvent]:
    try:
        async for ev in request.app.state.core.ask(conversation_id, caller.user_id, req.question):
            match ev:
                case Token(text):
                    yield ServerSentEvent(event="token", data={"text": text})
                case Citation(marker, _chunk_id, title, url):
                    yield ServerSentEvent(event="citation", data={"marker": marker, "title": title, "url": url})
                case Done(cid):
                    yield ServerSentEvent(event="done", data={"conversationId": cid})
    except Exception as exc:
        yield ServerSentEvent(event="error", data={"type": type(exc).__name__})

Source: FastAPI — Server-Sent Events.

The adapter keeps no state of its own, which is what lets the same core.ask feed a WebSocket, a Slack handler or an XMPP bot. How the full /v1/ask event vocabulary (status, citations, grounding) maps onto these three is a contract decision covered in Serving RAG as an API.

FastAPI WebSockets

A WebSocket adapter wraps the same core.ask. What it adds is a control channel: the client can send cancel, or a new ask that supersedes the running one, without opening a second connection. The complete, tested-in-process variant with a first-message token handshake is in the WebSocket section of Serving RAG as an API; the version below shows only the adapter-specific part, the translation of core events and the lifecycle of the generation task.

docsassistant/ws_adapter.py — a WebSocket adapter over the same core (Python, FastAPI 0.142.2)
import asyncio

from fastapi import WebSocket, WebSocketDisconnect

from docsassistant.core import Citation, DocsAssistantCore, Done, Token


async def run_turn(ws: WebSocket, core: DocsAssistantCore, conversation_id: str, user_id: str, question: str) -> None:
    async for ev in core.ask(conversation_id, user_id, question):
        match ev:
            case Token(text):
                await ws.send_json({"type": "token", "text": text})
            case Citation(marker, _chunk_id, title, url):
                await ws.send_json({"type": "citation", "marker": marker, "title": title, "url": url})
            case Done(cid):
                await ws.send_json({"type": "done", "conversationId": cid})


async def serve(ws: WebSocket, core: DocsAssistantCore, user_id: str, conversation_id: str) -> None:
    turn: asyncio.Task | None = None
    try:
        while True:
            msg = await ws.receive_json()
            if msg["type"] == "ask":
                if turn and not turn.done():
                    turn.cancel()                                   # a new question supersedes the running one
                turn = asyncio.create_task(run_turn(ws, core, conversation_id, user_id, msg["question"]))
            elif msg["type"] == "cancel" and turn:
                turn.cancel()
                await ws.send_json({"type": "cancelled"})
    except WebSocketDisconnect:
        if turn:
            turn.cancel()

Source: FastAPI — WebSockets.

Browsers cannot set an Authorization header on the WebSocket handshake, so the token travels in the first message, in a Sec-WebSocket-Protocol subprotocol entry, or as a short-lived single-use ticket in the URL (see Authenticating EventSource connections).

Spring WebFlux: Flux<ServerSentEvent> from ChatClient.stream()

In Spring AI 2.0.1, chatClient.prompt()…​stream() returns a Flux of text deltas (content()) or of whole responses with their metadata (chatClientResponse()). The Java core is the AssistantCore of Channel-Agnostic Architecture: it streams chatClientResponse() once, applies the ACL filter as the VectorStoreDocumentRetriever.FILTER_EXPRESSION advisor parameter built from an AclResolver, and appends Citation events and a terminal Done read from the RAG advisor’s retrieved documents (see Spring AI: Advisors and ChatModel & ChatClient). The adapter maps its CoreEvent records to ServerSentEvent objects. WebFlux supports a Flux<ServerSentEvent<T>> return value with produces = text/event-stream directly, and cancels the subscription when the client disconnects. The Spring MVC variant of the same endpoint is on Serving RAG as an API; this page uses WebFlux because a reactive pipeline is the natural fit for the cancellation and backpressure sections below (see Spring Boot: Reactive Programming).

ChatController.java — the SSE adapter (Java 21, Spring Boot 4.1.1 WebFlux)
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.security.core.annotation.AuthenticationPrincipal;
import org.springframework.security.oauth2.jwt.Jwt;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.Map;

@RestController
@RequestMapping("/v1/chat")
public class ChatController {

    private final AssistantCore core;                    // see Channel-Agnostic Architecture

    public ChatController(AssistantCore core) {
        this.core = core;
    }

    @PostMapping(path = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<Object>> stream(@RequestBody AskRequest req, @AuthenticationPrincipal Jwt jwt) {
        Flux<ServerSentEvent<Object>> events = core.ask(req.conversationId(), jwt.getSubject(), req.question())
                .map(ev -> switch (ev) {
                    case CoreEvent.Token t -> sse("token", Map.of("text", t.text()));
                    case CoreEvent.Citation c -> sse("citation", Map.of("marker", c.marker(), "title", c.title(),
                            "url", c.url()));
                    case CoreEvent.Done d -> sse("done", Map.of("conversationId", d.conversationId()));
                })
                .onErrorResume(e -> Flux.just(sse("error", Map.of("type", e.getClass().getSimpleName()))))
                .publish()
                .refCount(2);                                   // one model call shared by both subscribers below
        Flux<ServerSentEvent<Object>> keepAlive = Flux.interval(Duration.ofSeconds(15))
                .map(i -> ServerSentEvent.<Object>builder().comment("keep-alive").build());
        return events.mergeWith(keepAlive.takeUntilOther(events.ignoreElements()));
    }

    private static ServerSentEvent<Object> sse(String name, Object data) {
        return ServerSentEvent.<Object>builder(data).event(name).build();
    }
}

Source: Spring Framework — WebFlux and Spring Framework — WebFlux return values.

The keep-alive merge stops when the event flux completes or errors. events is subscribed to twice, once by mergeWith and once by ignoreElements(); without publish().refCount(2) each subscription would start its own model call. refCount(2) connects once both subscribers are present and cancels the core when both are gone, so a client disconnect still stops the generation. Spring does not add the 15-second ping that FastAPI sends by default, so the application has to. Everything outside the stream — the JWT, the CSRF decision, method security — is ordinary Spring Security, covered with the rest of the endpoint design in Spring Boot: REST APIs and Spring AI: Security & Guardrails.

Spring WebSocket and STOMP

A STOMP adapter is worth it when the client already speaks STOMP (for example through a broker relay used by other parts of the application), or when you want destination-based routing and user-targeted messages for free. The handler below starts the core flux for a user and sends each event to that user’s queue; it keeps the Disposable so that a cancel message or a disconnect can stop the generation.

ChatStompController.java — a STOMP adapter (Java 21, Spring Boot 4.1.1)
import org.springframework.context.event.EventListener;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.stereotype.Controller;
import org.springframework.web.socket.messaging.SessionDisconnectEvent;
import reactor.core.Disposable;

import java.security.Principal;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Controller
public class ChatStompController {

    private final AssistantCore core;
    private final SimpMessagingTemplate template;
    private final Map<String, Disposable> running = new ConcurrentHashMap<>();   // keyed by WebSocket session id

    public ChatStompController(AssistantCore core, SimpMessagingTemplate template) {
        this.core = core;
        this.template = template;
    }

    @MessageMapping("/ask")
    public void ask(AskRequest req, Principal principal, SimpMessageHeaderAccessor headers) {
        String session = headers.getSessionId();
        cancel(session);                                                           // supersede a running turn
        Disposable d = core.ask(req.conversationId(), principal.getName(), req.question())
                .subscribe(ev -> template.convertAndSendToUser(principal.getName(), "/queue/answer", ev));
        running.put(session, d);
    }

    @MessageMapping("/cancel")
    public void cancelMessage(SimpMessageHeaderAccessor headers) {
        cancel(headers.getSessionId());
    }

    @EventListener
    public void onDisconnect(SessionDisconnectEvent event) {
        cancel(event.getSessionId());
    }

    private void cancel(String session) {
        Disposable d = running.remove(session);
        if (d != null) {
            d.dispose();
        }
    }
}

Source: Spring Framework — STOMP.

convertAndSendToUser needs the user destination prefix and a Principal set during the handshake, which is where the bearer token or ticket is validated (a ChannelInterceptor on the inbound channel can validate a token sent in the STOMP CONNECT headers). Each token becomes one STOMP frame, so the per-frame overhead is larger than with SSE; batching a few deltas per frame is a reasonable optimisation. Sending from a subscriber callback ignores backpressure on the outbound channel; see Backpressure.

A Node and Express variant

Many teams put a small Node service in front of the Python or Java core, for example to share the SSE endpoint with a Next.js front end or to terminate browser concerns (cookies, CORS) in one place. The sketch below relays the core’s event stream and propagates the browser’s disconnect upstream through an AbortController. It assumes the core already emits SSE, so the gateway adds authentication and nothing else.

gateway/server.js — an Express SSE relay with disconnect propagation (Node.js, Express 5.2.1)
import express from "express";
import { Readable } from "node:stream";

const app = express();
app.use(express.json());

const CORE_URL = process.env.DOCSASSISTANT_CORE_URL;          // e.g. http://core:8000/v1/chat/stream
const CORE_TOKEN = process.env.DOCSASSISTANT_CORE_TOKEN;      // service credential, never sent to the browser

app.post("/chat", async (req, res) => {
  const abort = new AbortController();
  res.on("close", () => abort.abort());                       // the browser went away: stop the upstream request

  const upstream = await fetch(CORE_URL, {
    method: "POST",
    headers: { "Content-Type": "application/json", Authorization: `Bearer ${CORE_TOKEN}` },
    body: JSON.stringify(req.body),
    signal: abort.signal,
  });

  res.status(upstream.status);
  res.set({
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache, no-transform",
    "X-Accel-Buffering": "no",
  });
  res.flushHeaders();
  Readable.fromWeb(upstream.body).pipe(res);                  // pipe() propagates backpressure from the socket
});

app.listen(3000);

Source: Express 5.x API reference and Node.js — Stream.

The comment on pipe is the point of using it: a slow reader pauses the upstream body instead of letting Node buffer the whole answer in memory. A real gateway would also validate the browser’s session, map it to the userId the core expects, and handle an aborted request (fetch rejects with an AbortError that must not be logged as a failure).

Cancellation and client disconnects

A generation costs tokens and holds a retrieval and database connection until it finishes, so an abandoned request must stop it. Each transport signals abandonment differently:

Transport What stops the generation

SSE with FastAPI

When the client closes the connection, the server cancels the response task; the CancelledError is raised at the generator’s current await, which propagates into the async for over astream and stops the LLM call. Keep cleanup in try/finally and never swallow CancelledError. A route can also poll await request.is_disconnected().

SSE with WebFlux

The disconnect cancels the subscription, and Reactor propagates cancel upstream to the Spring AI Flux, which closes the HTTP call to the model provider. Operators such as doOnCancel are for logging and metrics.

WebSocket

The server receives a close or a cancel message and cancels the task (Python) or disposes the subscription (Java), as in the snippets above.

Express gateway

res.on("close") aborts the upstream fetch; the core then sees its own client disconnect.

What a disconnect should do to the conversation is a product decision. With a checkpointer the user message is already stored when generation is cancelled, so the next turn sees a question without an answer; decide whether to store the partial answer, drop the dangling question or mark it as interrupted. Explicit user cancellation ("Stop generating") deserves a distinct signal from a network drop: the first ends the turn, the second should allow a resume. That distinction is the next section.

Reconnecting with Last-Event-ID

When an EventSource loses its connection it reconnects after the retry interval and sends the id of the last event it received in a Last-Event-ID header. FastAPI exposes that header like any other, and Spring through @RequestHeader("Last-Event-ID"). Resuming a model generation is a different thing from resuming a file download: the stream is not recomputable from an offset, so the server must have kept the events already produced. The usual design is to decouple the generation from the connection:

  1. POST /v1/turns authenticates the caller, starts the core in a background task and returns 202 with a turnId.

  2. The task appends every core event to a per-turn log (an in-process list for one instance; a Redis Stream or a database table when several instances serve the same user).

  3. GET /v1/turns/{turnId}/events replays the log from Last-Event-ID + 1 and then follows it live.

The two-step shape also repairs the EventSource limitation: the question travels in a POST body, and the GET that EventSource can issue carries no secrets, only the unguessable turnId and the credentials discussed in the next sections.

docsassistant/turns.py — resumable event stream over an in-process log (Python, FastAPI 0.142.2)
import asyncio
from collections.abc import AsyncIterable
from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.sse import EventSourceResponse, ServerSentEvent

from docsassistant.api import current_caller                 # from Serving RAG as an API
from docsassistant.security import Caller

app = FastAPI()
turns: dict[str, "Turn"] = {}                    # illustrative: use Redis Streams or a table for several instances


class Turn:
    def __init__(self, user_id: str) -> None:
        self.user_id = user_id
        self.events: list[ServerSentEvent] = []
        self.finished = False
        self.changed = asyncio.Condition()
        self.task: asyncio.Task | None = None

    async def append(self, name: str, data: dict) -> None:
        async with self.changed:
            self.events.append(ServerSentEvent(event=name, data=data, id=str(len(self.events)), retry=3000))
            self.changed.notify_all()

    async def finish(self) -> None:
        async with self.changed:
            self.finished = True
            self.changed.notify_all()


async def open_turn(turn_id: str, caller: Annotated[Caller, Depends(current_caller)],
                    last_event_id: Annotated[int | None, Header()] = None) -> tuple[Turn, int]:
    turn = turns.get(turn_id)
    if turn is None or turn.user_id != caller.user_id:
        raise HTTPException(status_code=404)                  # same answer for foreign and unknown turns
    position = 0 if last_event_id is None else last_event_id + 1
    if turn.finished and position >= len(turn.events):
        raise HTTPException(status_code=204)                  # nothing left: EventSource stops reconnecting
    return turn, position


@app.get("/v1/turns/{turn_id}/events", response_class=EventSourceResponse)
async def turn_events(opened: Annotated[tuple[Turn, int], Depends(open_turn)]) -> AsyncIterable[ServerSentEvent]:
    turn, position = opened
    while True:
        async with turn.changed:
            await turn.changed.wait_for(lambda: position < len(turn.events) or turn.finished)
            batch, finished = turn.events[position:], turn.finished
        for ev in batch:
            yield ev
        position += len(batch)
        if finished and position >= len(turn.events):
            return

Source: FastAPI — Server-Sent Events and WHATWG HTML — Server-sent events.

The ownership check and the "already finished" check live in the open_turn dependency, for the same reason as in the first FastAPI adapter: once the generator has yielded, the response has said 200 and no status code can follow. A completed stream ends normally, so EventSource reconnects with the last id; the dependency then answers 204 No Content, which per the HTML standard makes EventSource stop reconnecting (a client can also call close() when it receives done).

Because the event stream no longer owns the generation, a disconnect must not cancel it; instead, cancel when the last subscriber has been gone for a grace period, or on an explicit DELETE /v1/turns/{turnId}. The log needs a time to live (minutes, not days), and event identifiers need only be unique and ordered within one turn. Spring offers the same shape with a Sinks.Many replay sink per turn; Sinks.many().replay().all() gives replay for late subscribers and live delivery afterwards, with the same lifetime caveats.

Backpressure

A model produces tokens at a rate of tens per second; a phone on a poor network may consume less. Without a policy, a slow reader makes the server buffer the difference. If \(r_m\) is the model’s token rate, \(r_c\) the client’s effective consumption rate and \(T\) the duration of the answer, the unread backlog grows as

\[ B(T) = \max\bigl(0,\,(r_m - r_c)\,T\bigr) \]

tokens per connection, and the memory held across \(N\) concurrent slow readers is \(N \cdot B(T)\) times the size of an event. Practical policies, from cheapest to most careful:

  • Coalesce. Send a batch of tokens every 50 to 100 milliseconds instead of one event per token; this reduces per-event overhead, which is large for SSE and STOMP framing. In Reactor, bufferTimeout or window do this; in Python, accumulate into a buffer and flush on a timer.

  • Let the transport push back. Reactor honours demand end to end, and WebFlux requests items only as the socket drains. Do not add onBackpressureBuffer() without a bound; use onBackpressureBuffer(maxSize) with an overflow strategy, or onBackpressureLatest() for state-like events (a progress indicator) where only the newest value matters.

  • Bound the per-turn log. For text tokens, the full answer is the state: send the client an occasional snapshot and drop the intermediate deltas for readers that fell behind.

  • Drop the reader. If the backlog passes a limit, close the connection with an error event; the client resumes with Last-Event-ID from the log, and the model keeps generating undisturbed.

In Python, await ws.send_json(…​) and an SSE generator both suspend when the socket’s write buffer is full, so a slow reader naturally slows the generator and, through the async for, the model call; that is back pressure by construction, and the reason an unbounded asyncio.Queue between the core and the adapter is the thing to avoid.

Proxy buffering and timeouts

The most common reason a streamed answer arrives "all at once" in production is a proxy that buffers the response. Nginx buffers proxied responses by default; it honours an X-Accel-Buffering: no response header and the equivalent proxy_buffering off; directive per location. FastAPI’s EventSourceResponse sets that header itself; a Spring or Express adapter has to set it (the Express sketch above does), or the proxy configuration has to turn buffering off.

location /v1/chat/ {
    proxy_pass http://docsassistant_core;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    proxy_buffering off;
    proxy_cache off;
    proxy_read_timeout 300s;
}

Source: nginx — ngx_http_proxy_module.

Related points, each of which has caused a missing-stream incident somewhere:

  • Compression middleware (gzip, brotli) can hold bytes until a block is full. Exclude text/event-stream from it, or flush after each event.

  • Cache-Control: no-cache, no-transform prevents intermediaries from caching or rewriting the stream.

  • Idle timeouts of load balancers, API gateways and mobile networks are often 30 to 120 seconds: a comment line every 15 seconds (: keep-alive) is enough to keep the connection open while the model or a tool is thinking.

  • With HTTP/1.1, browsers allow about six connections per host, and each open EventSource consumes one; HTTP/2 lifts that limit, so terminate TLS with HTTP/2 in front of the adapter.

  • Cloud front ends have their own switches (response streaming, idle timeouts, WebSocket support); the Azure side is covered in Azure Container Apps and Azure API Management, which needs streaming explicitly enabled on the operation.

Authenticating EventSource connections

EventSource and WebSocket cannot attach an Authorization header, which leaves four options:

Option How Trade-offs

fetch plus a stream parser

fetch(url, {method: "POST", headers: {Authorization: …​}}) and read response.body; libraries such as @microsoft/fetch-event-source wrap this and add retry and Last-Event-ID

Keeps standard bearer tokens and POST bodies; you own reconnection logic (or the library does)

Cookie session

new EventSource(url, {withCredentials: true}) with a SameSite session cookie set by your login flow

Simple for same-site apps; needs CSRF thought for any state-changing request and CORS with credentials for cross-origin

Short-lived ticket

An authenticated POST /v1/tickets returns an opaque, single-use value valid for about 30 seconds; the browser opens /v1/turns/{id}/events?ticket=…​

Works with native EventSource and WebSocket; the ticket is still visible in URLs and logs, so it must be single-use and bound to the user and the turn

Token in a query string

?access_token=…​ with the real access token

Avoid: tokens leak into access logs, browser history and Referer headers

Whichever option is used, the adapter derives userId from the verified credential, never from a request field, because that value keys conversation memory and the access-control filter of the retriever (see Access Control & Privacy). OAuth 2.0 mechanics — which grant for a browser client, how to keep tokens out of JavaScript, how long access tokens should live — are in OAuth for Browser-Based Apps, Authorization Code & PKCE and OAuth Security Best Practices. A token that expires in the middle of a long stream does not interrupt it; the check happens at connect time, so reconnection is the moment to re-authenticate and a short access-token lifetime combined with Last-Event-ID resume is acceptable.

Choosing, in short

  • One question, one streamed answer, a browser or app client: SSE, with the two-step POST then GET shape when the stream must be resumable, otherwise fetch streaming over POST.

  • The client talks while the server streams (cancel, supersede, typing indicators): WebSocket, or STOMP if the application already runs a STOMP broker.

  • Service-to-service or script clients: NDJSON or SSE, whichever the client library parses with less code.

  • Always: a transport-independent core, keep-alives, a disconnect policy, buffering off in every proxy on the path and the user identity taken from the verified credential.

The next pages in this sub-section keep the transport and standardise the payload: Chat UI Protocols covers the AI SDK UI Message Stream and AG-UI, and Embedding a Web Chat Widget consumes the SSE endpoint from a page without a framework. The adapter pattern itself is on Channel-Agnostic Architecture, and operational concerns (rate limits per channel, observability, escalation) on Operating Chat Channels.

References