Most LLM APIs emit tokens incrementally, but a naive Flask JSON endpoint buffers the whole response and defeats the point. To ship a responsive UI, you need flask server-sent events llm streaming: a server route that yields chunks as Server-Sent Events while proxying the model’s token stream. This article walks through a working implementation you can run locally and extend to production.
Step 1: Scaffold the Flask app and dependencies
Start with a clean virtual environment and install the only two packages you actually need: Flask for the HTTP layer and Requests for the upstream streaming call.
python -m venv venv
source venv/bin/activate
pip install flask requestsCreate a minimal app.py so you have a known-good baseline before adding streaming logic:
from flask import Flask, Response, request, jsonify
app = Flask(__name__)
@app.route("/health")
def health():
return jsonify(ok=True)
if __name__ == "__main__":
app.run(port=5000, debug=True)Run python app.py and curl localhost:5000/health. If that returns {"ok":true}, your Python environment and Flask import path are correct. Debugging SSE on a broken skeleton wastes hours.
Keep the file layout flat for this tutorial. In a larger service you would split the LLM client into its own module, but a single file makes the data flow obvious.
Step 2: Call a streaming LLM endpoint
OpenAI-compatible chat completions accept "stream": true and respond with text/event-stream. Each line is data: {json} and the stream terminates with data: [DONE]. Use requests.post(..., stream=True) so the body is never fully buffered in memory.
If you point at an OpenAI-compatible gateway such as n4n.ai, the request shape is identical and you get automatic fallback when a provider is rate-limited, but the parsing code below does not change.
import requests
import os
LLM_URL = os.environ.get("LLM_URL", "https://api.openai.com/v1/chat/completions")
LLM_KEY = os.environ.get("LLM_KEY", "sk-placeholder")
def stream_llm(messages):
payload = {
"model": "gpt-4o-mini",
"messages": messages,
"stream": True,
"temperature": 0.7,
}
headers = {
"Authorization": f"Bearer {LLM_KEY}",
"Content-Type": "application/json",
}
with requests.post(LLM_URL, json=payload, headers=headers, stream=True, timeout=30) as r:
r.raise_for_status()
for line in r.iter_lines(decode_unicode=True):
if not line:
continue
if line.startswith("data: "):
data = line[len("data: "):]
if data.strip() == "[DONE]":
break
yield dataA typical chunk looks like this:
{
"id": "chatcmpl-123",
"object": "chat.completion.chunk",
"choices": [
{ "delta": { "content": "The" }, "index": 0, "finish_reason": null }
]
}The delta.content field holds the new text. The first chunk usually carries delta.role; the last carries finish_reason. Your generator should ignore empty content and break on [DONE].
Step 3: Convert token chunks to SSE frames
Server-Sent Events are simple: a sequence of field: value lines followed by a blank line. The browser EventSource API parses these automatically. We wrap each LLM delta as a JSON object inside a data: line so the client gets structured events.
import json
def sse_pack(event, data):
return f"data: {json.dumps({'event': event, 'data': data})}\n\n"
def llm_sse_generator(messages):
for chunk_json in stream_llm(messages):
try:
chunk = json.loads(chunk_json)
except json.JSONDecodeError:
continue
delta = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "")
if delta:
yield sse_pack("token", delta)
yield sse_pack("done", "")We send only the incremental text, not the full accumulated string, to keep frame size small. The done event lets the client know the stream finished without parsing the connection close.
Step 4: Expose the SSE route in Flask
Flask’s Response object accepts any generator and a mimetype. Set text/event-stream and add headers that tell intermediaries not to buffer.
@app.route("/stream", methods=["POST"])
def stream():
body = request.get_json(force=True)
messages = body.get("messages", [])
if not messages:
return jsonify(error="messages required"), 400
def gen():
yield "retry: 1000\n\n"
for frame in llm_sse_generator(messages):
yield frame
return Response(gen(), mimetype="text/event-stream", headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
})The retry: 1000 line is part of the SSE spec; it instructs the browser to wait one second before reconnecting if the socket drops. X-Accel-Buffering: no is an nginx-specific hint we reinforce with proxy config later.
Note that Flask’s default dev server is synchronous. A single slow upstream will block other requests. That is fine for local testing but must be fixed before production (see below).
Step 5: Build a minimal browser client
EventSource only supports GET, so for a POST body we use the Fetch API with a ReadableStream reader. This works in all modern browsers and gives us full control over the SSE frame parsing.
<!doctype html>
<html>
<body>
<div id="out"></div>
<script>
async function callStream() {
const res = await fetch("/stream", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
messages: [{ role: "user", content: "Explain SSE in one sentence." }]
})
});
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const frames = buffer.split("\n\n");
buffer = frames.pop();
for (const frame of frames) {
if (!frame.startsWith("data: ")) continue;
const payload = JSON.parse(frame.slice(6));
if (payload.event === "token") {
document.getElementById("out").textContent += payload.data;
}
}
}
}
callStream();
</script>
</body>
</html>If you prefer a pure GET design, move the prompt into a query parameter and use EventSource("/stream?q=...") instead. Either way, the Flask generator stays the same.
Step 6: Run and verify success
Export a real API key and start the server:
export LLM_KEY=your_real_key
python app.pyVerify with curl using the -N flag to disable curl’s own buffering:
curl -N -X POST http://localhost:5000/stream \
-H "Content-Type: application/json" \
-d '{"messages":[{"role":"user","content":"Hi"}]}'Success looks like multiple data: {...} lines printed one at a time with small delays, not a single blob after a long pause. In the browser, the #out div fills incrementally. If you see buffering, confirm X-Accel-Buffering is set and that you are not behind a default nginx config.
Production considerations
The Flask dev server will not survive concurrent streams. Deploy with Gunicorn using a coroutine worker:
pip install gunicorn gevent
gunicorn -k gevent -w 4 app:app --timeout 60Gevent yields during iter_lines, so a worker handles many idle streams simultaneously. Set --timeout above your longest expected LLM latency.
Behind nginx, add to the location block:
proxy_buffering off;
proxy_set_header X-Accel-Buffering no;
proxy_read_timeout 300s;Without proxy_buffering off, nginx holds the response until its buffer fills or the upstream closes.
Handle client disconnects explicitly. When the browser closes the tab, Flask raises GeneratorExit inside your generator. Wrap the upstream request so it closes cleanly:
def stream_llm(messages):
try:
with requests.post(LLM_URL, json=payload, headers=headers, stream=True, timeout=30) as r:
for line in r.iter_lines(decode_unicode=True):
yield line
except GeneratorExit:
pass # context manager closes the socketAdd CORS headers if your frontend is on a different origin. Use Flask-CORS or manual Access-Control-Allow-Origin on the /stream route.
For observability, log the number of frames sent and the elapsed time per request. If you later add per-token metering or provider routing directives, an inference gateway can forward cache-control hints transparently; your flask server-sent events llm streaming code stays exactly as written because it only speaks SSE and JSON.
That is the complete path: an LLM token stream enters Flask, gets repackaged as SSE frames, and reaches the browser incrementally. The pattern is stable across any OpenAI-compatible backend and holds up under real load with the right worker and proxy configuration.