Streaming & Server-Sent Events

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

The reference material consulted while preparing these pages is listed in the section bibliography.

This page is written against Flask 3.1.3 (released 2026-02-19), Werkzeug 3.1.9 and Python 3.13, checked on 2026-10-07; every Python example was executed with those versions (Flask’s test client for the Flask examples, the Quart test client for the Quart one, a fake Redis for the fan-out example). Flask 3.2 is not released yet and its changelog has nothing that changes streaming. The page follows the Bookshelf application of the previous pages and answers one question from several angles: how does a WSGI application send a response that is not finished when the view function returns? It starts with plain generators and the response-ordering rule that bites everybody once, adds the request context that generators lose, streams templates and large exports, builds a live "new reviews" feed with Server-Sent Events (SSE), explains why every open stream occupies a worker and what to do about it (threads, gevent, Quart, a proxy that does not buffer), and closes with an overview of WebSockets. The four Flask books used as a cross-check for this site hardly touch the subject: they stop at ordinary request and response cycles, and the one chapter that mentions asynchronous Python only names Quart. Everything here comes from the official documentation.

How Streaming Works in WSGI

A WSGI application is a callable that returns an iterable of byte strings, and the server writes every item to the socket as it is produced (PEP 3333). Flask normally hands the server a one-item list holding the whole body, but a Werkzeug response accepts any iterable. When the view returns a generator, Flask wraps it in a Response whose body is that generator, the server sends the status line and the headers, and then pulls the generator one item at a time. Because the length is not known up front, there is no Content-Length; the server frames the body itself (chunked transfer encoding over HTTP/1.1, or connection close).

Three consequences shape the rest of the page.

  • Headers are sent before the first item. Nothing set after the first yield reaches the client, and an exception raised inside the generator can no longer become a 500 page, because the status line is already on the wire.

  • The generator runs after the view has returned. The request and application contexts have been popped by then, unless stream_with_context keeps them alive.

  • A worker is busy until the generator finishes. For a short export this is nothing; for a stream that stays open for minutes it is the single most important operational fact, and the subject of its own section.

sequenceDiagram participant C as Browser participant P as Reverse proxy participant S as WSGI server worker participant F as Flask view participant G as Generator C->>P: GET /reviews/stream P->>S: GET /reviews/stream S->>F: call application F-->>S: Response with generator body S-->>P: status line and headers P-->>C: status line and headers loop until the generator ends or the client leaves S->>G: next item G-->>S: bytes S-->>P: chunk P-->>C: chunk (only if the proxy does not buffer) end C--xP: connection closed S->>G: close() raises GeneratorExit
Figure 1. Who does what while a streamed response is produced

Source: Werkzeug: Request and Response Objects

Streaming with Generators

The Flask pattern page recommends a nested generator function, a response that wraps it, and setting everything about the response before the generator is returned. A view may return the generator directly (Flask converts it, see the table of return values in Responses & JSON), but building the Response explicitly makes the mimetype and the headers visible.

from flask import Flask, Response

app = Flask(__name__)

BOOKS = ["Dune", "Emma", "Ubik"]


@app.get("/books.txt")
def books_text():
    def generate():
        for title in BOOKS:
            yield f"{title}\n"          # every yield is one chunk for the server

    headers = {"Cache-Control": "no-store"}          # set before streaming
    return Response(generate(), mimetype="text/plain", headers=headers)


client = app.test_client()
response = client.get("/books.txt")
assert "Content-Length" not in response.headers
assert list(response.response) == [b"Dune\n", b"Emma\n", b"Ubik\n"]
print(response.headers["Content-Type"])

Source: Flask: Streaming Contents

Things worth knowing before you stream:

  • A generator can yield str (encoded as UTF-8) or bytes; a dict is not a chunk.

  • With Response(generate(), ...) Werkzeug does not add a Content-Length. If you do know the size (for example the size of a file on disk), set it yourself, or better use send_file, which handles ranges and conditional requests (covered in Responses & JSON).

  • Streaming and compression are at odds. A middleware or a proxy that compresses the whole body has to buffer it, or has to flush after every chunk. The official page warns that some WSGI middlewares break streaming (profilers and debug middleware that collect the body are the usual cause); test streaming with the same stack you deploy.

  • The development server handles each connection in its own thread (flask run is threaded by default), so a stream does not block other requests in development. This hides the worker problem that appears with the single-threaded workers of production servers, see below.

Headers Are Final Once Streaming Starts

The Flask documentation states the rule bluntly: headers cannot change after the streaming response starts. In practice this has four effects.

  • Status code and Content-Type are decided in the view. An error discovered halfway through cannot change the status to 500.

  • Vary: Cookie. Flask adds Vary: Cookie to a response when the session was accessed during the request. If only the generator reads the session, the header is never added. Touch session in the view as well, as the official page recommends, and never modify the session inside the generator: the Set-Cookie header has already gone out, so the change is lost.

  • after_request functions run before the generator does. A function that logs the number of bytes sent or adds a header to the response sees an unfinished stream.

  • teardown_request and teardown_appcontext functions run when the context is popped, normally before the first chunk (see the next section).

Large CSV Exports

A catalogue export is the classic case: millions of rows must not be built in memory, so each row is formatted and sent as soon as it is read. Python’s csv.writer needs a file-like object; the trick is a tiny object whose write returns the string instead of storing it, so that the generator can yield what the writer produced for the current row. The headers, including Content-Disposition, are set up front.

import csv
import secrets
from io import StringIO

from flask import Flask, Response, session

app = Flask(__name__)
app.config["SECRET_KEY"] = secrets.token_hex(32)   # in an application: read it from the environment

ROWS = [("Dune", "Frank Herbert", 1965), ("Emma", "Jane Austen", 1815), ('Say "hi", Book', "A. Writer", 2001)]


class Echo:
    """Write-only buffer: write() hands the text back instead of keeping it."""

    def write(self, value: str) -> str:
        return value


def csv_lines(rows):
    writer = csv.writer(Echo())
    yield writer.writerow(["title", "author", "year"])
    for row in rows:
        yield writer.writerow(row)


@app.get("/books.csv")
def export_books():
    session.get("user_id")          # read the session in the view so Vary: Cookie is set
    return Response(
        csv_lines(ROWS),
        mimetype="text/csv",
        headers={"Content-Disposition": 'attachment; filename="books.csv"'},
    )


response = app.test_client().get("/books.csv")
text = response.get_data(as_text=True)
assert text.splitlines()[0] == "title,author,year"
assert '"Say ""hi"", Book"' in text                  # quoting is done by csv, not by f-strings
assert "Cookie" in response.headers["Vary"]
print(text)

Source: Flask: Streaming Contents

Never build CSV with ",".join(row) (the official page’s own sketch does, to keep it short): a comma or a quote in a title breaks the file, and a cell starting with =, +, - or @ becomes a formula when someone opens the file in a spreadsheet. If the data is not fully trusted, prefix such cells with a quote character or a space before writing them.

NDJSON and Rows from a Database

JSON arrays cannot be streamed by a plain jsonify, but newline-delimited JSON (NDJSON, one object per line, application/x-ndjson) can, and every client can parse it line by line. The database side has the same requirement as the response side: do not load all rows. SQLAlchemy 2.0 streams rows when the statement has the yield_per execution option, which fetches them in batches instead of buffering the full result. A generator that uses db.session needs the application context for as long as it runs, which is the next topic: the example wraps it with stream_with_context.

import json

from flask import Flask, Response, stream_with_context
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


db = SQLAlchemy(model_class=Base)


class Book(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]


app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite://"     # in-memory, for the example
db.init_app(app)

with app.app_context():
    db.create_all()
    db.session.add_all(Book(title=f"Book {n}") for n in range(1, 251))
    db.session.commit()


@app.get("/books.ndjson")
def books_ndjson():
    statement = select(Book).order_by(Book.id).execution_options(yield_per=100)

    @stream_with_context
    def generate():
        for book in db.session.scalars(statement):
            yield json.dumps({"id": book.id, "title": book.title}) + "\n"

    return Response(generate(), mimetype="application/x-ndjson")


lines = app.test_client().get("/books.ndjson").get_data(as_text=True).splitlines()
assert len(lines) == 250 and json.loads(lines[-1]) == {"id": 250, "title": "Book 250"}
print(lines[0], "...", lines[-1])

Source: SQLAlchemy: ORM API Features for Querying (yield_per)

Databases that do server-side cursors (PostgreSQL) really stream; SQLite and some drivers still buffer in the driver. Measure memory on the database you use. The Flask-SQLAlchemy session is scoped to the application context, so a stream that reads from it holds a connection until the stream ends; a long export is a good candidate for a read replica or for a background job that writes a file to object storage (see Background Jobs, Celery & Email).

stream_with_context and the Lost Request

Flask pops the request and application contexts when the view returns, before the server asks the generator for its first item. Code inside the generator that touches request, session, g or current_app therefore fails with a RuntimeError ("Working outside of request context"). The mechanism behind the contexts is covered in Application & Request Lifecycle, which also has a first example; this section adds the details.

stream_with_context(generator_or_function) works in two forms: as a wrapper around a generator that already exists, and as a decorator on a generator function. In both forms the wrapped generator re-pushes the saved contexts every time the server asks for the next item and pops them again afterwards, so request, session and g work inside it. It must be called while a request is active; outside one it raises a RuntimeError. It has existed since Flask 0.9, and the decorator form is what the current API reference shows.

from flask import Flask, Response, g, request, stream_with_context

app = Flask(__name__)


@app.before_request
def remember_shelf():
    g.shelf = request.args.get("shelf", "all")


@app.get("/broken")
def broken():
    def generate():
        yield f"shelf={request.args.get('shelf')}\n"      # request is gone when this runs

    return Response(generate(), mimetype="text/plain")


@app.get("/wrapped")
def wrapped():
    def generate():
        yield f"shelf={request.args.get('shelf')}\n"
        yield f"g.shelf={g.shelf}\n"

    return Response(stream_with_context(generate()), mimetype="text/plain")


@app.get("/decorated")
def decorated():
    @stream_with_context
    def generate():
        yield f"shelf={request.args.get('shelf')}\n"

    return Response(generate(), mimetype="text/plain")


client = app.test_client()
try:
    client.get("/broken?shelf=sci-fi").get_data()
except RuntimeError as error:
    print("without stream_with_context:", error)

assert client.get("/wrapped?shelf=sci-fi").get_data(as_text=True) == "shelf=sci-fi\ng.shelf=sci-fi\n"
assert client.get("/decorated?shelf=sci-fi").get_data(as_text=True) == "shelf=sci-fi\n"

Source: Flask API: stream_with_context

Points to keep in mind:

  • Prefer reading what you need in the view. The simplest generator takes plain values as arguments (generate(shelf)) and needs no context at all. Use stream_with_context when the generator really needs the request, the session, g or the database session.

  • The context lives as long as the stream. Teardown functions run when the generator finishes or is closed, so a database session or a connection acquired in the context stays checked out for the whole stream.

  • Do not modify the session in the generator, see above.

  • Only generators. Passing a non-iterable that is not callable fails; stream_template and stream_template_string call it for you when a request is active.

  • Threads are different. stream_with_context does not help a worker thread you start yourself; use copy_current_request_context for that, or better pass plain data.

Streaming Templates

Jinja can render a template piece by piece instead of building one string. Flask 2.2 added the two functions that expose it, both returning an iterator of strings that you can return from a view: stream_template(name, **context) renders a template by name (or the first existing name of a list), and stream_template_string(source, **context) renders template source. They use stream_with_context automatically if a request is active, so request and g work in the template. The chunks generally match the statement blocks of the template (loops, ifs, text between them), so the browser can start rendering the page header while the body is still being computed. Combine it with a slow data source (a report, a search over many shelves) and the user sees something immediately.

import time

from flask import Flask, Response, stream_template, stream_template_string
from jinja2 import DictLoader

app = Flask(__name__)
app.jinja_env.loader = DictLoader({
    "timeline.html": (
        "<h1>Timeline for {{ shelf }}</h1>\n"
        "<ul>\n"
        "{% for event in events %}<li>{{ event }}</li>\n{% endfor %}"
        "</ul>\n"
    ),
})


def slow_events():
    for n in range(1, 4):
        time.sleep(0.01)                  # a slow source: each item appears when it is ready
        yield f"review #{n} <posted>"      # autoescaped: the angle brackets are not markup


@app.get("/timeline")
def timeline():
    return Response(stream_template("timeline.html", shelf="sci-fi", events=slow_events()))


@app.get("/inline")
def inline():
    return Response(stream_template_string("Hello {{ name }}!", name="reader"))


client = app.test_client()
chunks = list(client.get("/timeline").response)
print(len(chunks), "chunks;", chunks[0])
assert chunks[0].startswith(b"<h1>Timeline")
assert any(b"review #2 &lt;posted&gt;" in chunk for chunk in chunks)     # still escaped
assert client.get("/inline").get_data(as_text=True) == "Hello reader!"

Source: Flask: Streaming Contents

Two details are easy to miss. First, the template is rendered by the same Jinja environment as render_template, so autoescaping and template inheritance apply as usual; pass the values the template needs explicitly, as the example does. Second, a variable that is a generator (as events above) is consumed while the template is rendered, which is what makes the page progressive; a variable that is a list is already in memory. Template inheritance is described in Jinja Templates.

Errors, Disconnects and Cleanup

Because the response has started, error handling for a stream is different from the rest of the application (the general mechanism is in Error Handling). Plan for three situations.

  • The generator raises. The status line is gone, so Flask cannot send its 500 page; the server logs the traceback and closes the connection, and the client sees a truncated body. Catch exceptions inside the generator and stream a final error record (for NDJSON, a line such as {"error": "export failed"}), or let the connection die and make clients verify completeness (a trailer line, a row count).

  • The client leaves. The server notices when it fails to write, stops pulling, and closes the iterable. Closing a generator raises GeneratorExit at the paused yield, so a finally block (or a context manager) in the generator is the right place to release subscriptions, cursors and temporary files. Werkzeug’s Response.close() forwards the call to the body.

  • Nobody ever reads. A generator that is created but never iterated runs no code, so it also never runs its finally. Create expensive resources inside the generator body, not in the view.

from flask import Flask, Response

app = Flask(__name__)
events = []


@app.get("/count")
def count():
    def generate():
        events.append("opened")
        try:
            for n in range(1000):
                yield f"{n}\n"
        except Exception:
            events.append("failed")
            yield '{"error": "stream failed"}\n'
        finally:
            events.append("closed")          # runs on normal end, error and client disconnect

    return Response(generate(), mimetype="text/plain")


client = app.test_client()
response = client.get("/count")
first = next(iter(response.response))        # the client reads one chunk ...
assert first == b"0\n"
response.close()                              # ... and disconnects
assert events == ["opened", "closed"]
print(events)

Source: Werkzeug: Request and Response Objects

Server-Sent Events

Server-Sent Events are a one-way channel from server to browser over a normal HTTP response that never ends. The browser side is the EventSource object, which opens a GET request, parses the stream, dispatches events and reconnects by itself. The server side is any streaming response with Content-Type: text/event-stream, which is why Flask needs no extension for it. Compared with WebSockets (see the last section) SSE is simpler: plain HTTP, automatic reconnection with resume, works with ordinary authentication cookies and most proxies, and is enough whenever the client does not need to send a stream of its own (live feeds, progress bars, notifications, token streams from an LLM).

The Wire Format

The body is UTF-8 text. A message is a block of field: value lines ended by an empty line.

Field Meaning

data

The payload. Several data lines of one message are joined with a newline by the browser. A newline inside your payload must therefore be sent as separate data: lines; a JSON document without line breaks avoids the issue.

event

The event name. A message without it is dispatched as a generic message event (onmessage); with it, to listeners registered with addEventListener(name, ...).

id

Sets the stream’s last event id. After a reconnect, the browser sends it back in a Last-Event-ID request header.

retry

Reconnection delay, in milliseconds (an integer).

a line starting with :

A comment, ignored by the browser. Used as a keep-alive so that proxies and the browser do not time out an idle stream.

The response headers are Content-Type: text/event-stream and Cache-Control: no-cache. The Quart documentation also sets Transfer-Encoding: chunked, which Flask leaves to the WSGI server. With nginx in front, add X-Accel-Buffering: no (see the proxy section). The specification is the WHATWG HTML standard.

A Live "New Reviews" Feed

The Bookshelf feed has three parts: a broker that remembers recent events and fans out to subscribers, a helper that encodes an event, and a view that holds the stream open. The broker keeps one bounded queue per subscriber; a subscriber that stops reading and fills its queue is dropped rather than slowing everybody down. The view subscribes before it returns the response (so a failure to subscribe is still an ordinary error response), and the generator uses only plain values, so it needs no stream_with_context. It sends a comment when nothing has happened for a while, and unsubscribes in finally, so a closed connection releases the queue.

import json
import queue
import threading
from collections import deque

from flask import Flask, Response, request


def format_sse(data=None, *, event=None, id=None, retry=None) -> str:
    """Encode one Server-Sent Event; data is serialized as one-line JSON."""
    lines = []
    if retry is not None:
        lines.append(f"retry: {int(retry)}")
    if event:
        lines.append(f"event: {event}")
    if id is not None:
        lines.append(f"id: {id}")
    if data is not None:
        lines.append(f"data: {json.dumps(data)}")      # json.dumps never contains a raw newline
    return "\n".join(lines) + "\n\n"


class ReviewBroker:
    """In-process fan-out with a short replay history. One instance per process."""

    def __init__(self, history: int = 100, queue_size: int = 100):
        self._lock = threading.Lock()
        self._subscribers: set[queue.Queue] = set()
        self._history: deque = deque(maxlen=history)
        self._queue_size = queue_size
        self._next_id = 1

    def publish(self, payload: dict) -> int:
        with self._lock:
            event_id = self._next_id
            self._next_id += 1
            item = (event_id, payload)
            self._history.append(item)
            for subscriber in list(self._subscribers):
                try:
                    subscriber.put_nowait(item)
                except queue.Full:                   # a slow consumer must not block the others
                    self._subscribers.discard(subscriber)
            return event_id

    def subscribe(self, last_id: int | None = None):
        subscriber: queue.Queue = queue.Queue(maxsize=self._queue_size)
        with self._lock:
            backlog = [item for item in self._history if last_id is not None and item[0] > last_id]
            self._subscribers.add(subscriber)
        return subscriber, backlog

    def unsubscribe(self, subscriber: queue.Queue) -> None:
        with self._lock:
            self._subscribers.discard(subscriber)

    @property
    def count(self) -> int:
        with self._lock:
            return len(self._subscribers)


app = Flask(__name__)
app.config["SSE_KEEPALIVE_SECONDS"] = 15
broker = ReviewBroker()


@app.get("/reviews/stream")
def review_stream():
    last_id = request.headers.get("Last-Event-ID", type=int)     # sent by EventSource on reconnect
    keepalive = app.config["SSE_KEEPALIVE_SECONDS"]
    subscriber, backlog = broker.subscribe(last_id)

    def generate():
        try:
            yield format_sse(retry=3000)                         # tell the browser how long to wait
            for event_id, payload in backlog:                    # replay what the client missed
                yield format_sse(payload, event="review", id=event_id)
            while True:
                try:
                    event_id, payload = subscriber.get(timeout=keepalive)
                except queue.Empty:
                    yield ": keep-alive\n\n"                     # comment line, ignored by the browser
                    continue
                yield format_sse(payload, event="review", id=event_id)
        finally:
            broker.unsubscribe(subscriber)

    return Response(
        generate(),
        mimetype="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@app.post("/books/<int:book_id>/reviews")
def add_review(book_id: int):
    payload = {"book_id": book_id, "stars": request.get_json()["stars"]}
    return {"id": broker.publish(payload)}, 201


app.config["SSE_KEEPALIVE_SECONDS"] = 0.05      # short, only so that the example finishes quickly
client = app.test_client()

response = client.get("/reviews/stream")
stream = response.iter_encoded()
assert response.headers["Content-Type"].startswith("text/event-stream")
assert next(stream) == b"retry: 3000\n\n"

assert client.post("/books/7/reviews", json={"stars": 5}).status_code == 201
assert next(stream) == b'event: review\nid: 1\ndata: {"book_id": 7, "stars": 5}\n\n'
assert next(stream) == b": keep-alive\n\n"        # nothing happened for 50 ms
assert broker.count == 1

response.close()                                   # the client goes away
assert broker.count == 0

client.post("/books/8/reviews", json={"stars": 4})
resumed = client.get("/reviews/stream", headers={"Last-Event-ID": "1"})
chunks = resumed.iter_encoded()
next(chunks)                                        # retry
print(next(chunks).decode().strip())               # only the event missed while disconnected
resumed.close()

Source: WHATWG HTML Standard: Server-sent events

The matching browser code is short. EventSource reconnects after a network error or a server restart, waiting the retry interval, and sends the Last-Event-ID header automatically, which is why the view above can replay.

const source = new EventSource("/reviews/stream");

source.addEventListener("review", (event) => {
  const review = JSON.parse(event.data);
  console.log(`new review for book ${review.book_id}: ${review.stars} stars`);
});

source.onerror = () => {
  // The browser retries on its own; close() stops it for good.
};

Source: MDN: Using server-sent events

flowchart LR P["POST /books/7/reviews"] --> B["ReviewBroker.publish"] B --> H["history deque, for replay"] B --> Q1["queue of subscriber 1"] B --> Q2["queue of subscriber 2"] B --> Q3["queue of subscriber 3"] Q1 --> G1["generator in worker thread 1"] Q2 --> G2["generator in worker thread 2"] Q3 --> G3["generator in worker thread 3"] G1 --> C1["browser 1"] G2 --> C2["browser 2"] G3 --> C3["browser 3"]
Figure 2. Fan-out from a review being posted to every open stream

Source: Flask: Streaming Contents

SSE Pitfalls and Security

  • EventSource can only issue GET requests and cannot set headers. An Authorization header is impossible; rely on the session cookie (add withCredentials: true to the constructor options for a cross-origin stream, and configure CORS accordingly, see SPAs & CORS) or, if you must use a token, a short-lived one-time token in the URL, which ends up in logs and therefore must expire within seconds. A fetch call that reads the response body as a stream is the alternative when headers are required, at the price of writing the reconnect logic yourself.

  • Authorize every stream, and filter per user. The feed above is public; a feed of private shelves must check the user at connection time and publish only events the user may see. A stream that stays open outlives the permission that was checked when it started: close it on logout or after a token lifetime.

  • Limit the number of connections. Each open stream is a held worker, a queue and a socket. Cap per user or per address (see Flask-Limiter in Extensions & Ecosystem), bound the queues as the broker does, and set a maximum stream lifetime after which you end the response and let the browser reconnect.

  • Browsers cap connections. Over HTTP/1.1, a browser allows six connections per domain across all tabs, and an EventSource holds one; HTTP/2 defaults to many more streams. Share one stream per page rather than one per widget.

  • Do not stream secrets and do not put session identifiers or tokens in event payloads.

  • Escape on the client. Event data is text: insert it with textContent, not innerHTML.

  • Behind a proxy, buffering and idle timeouts turn a stream into a stalled request, see below.

  • A status of 204 from the endpoint tells EventSource not to reconnect; any other non-200 or a wrong content type fails the connection.

Why a Stream Ties Up a Worker

A WSGI worker handles a request from the first byte to the last. While a generator waits on queue.get, that worker (or thread) can do nothing else. With a long-lived SSE stream, concurrency is bounded by the number of workers or threads, not by CPU. The table shows what each common setup does with 1,000 idle subscribers.

Setup What happens When it is acceptable

Werkzeug development server

One thread per connection (flask run is threaded by default).

Development only; it is not a production server.

Gunicorn sync worker (the default)

One request at a time per worker: with four workers, the fifth open stream waits, and ordinary requests queue behind them.

Never for SSE.

Gunicorn gthread (--workers 2 --threads 100)

A thread per stream: the Gunicorn design page lists it among the worker types that keep connections, but a thread costs memory and the pool is finite.

Tens to a few hundred concurrent streams.

Gunicorn gevent worker

Greenlets: a blocked wait switches to another one, so one process serves thousands of connections, according to the Gunicorn design page (long polling and streaming are its recommended case). Needs gevent-aware code: queues from gevent.queue or monkey-patched queue, and I/O libraries that cooperate.

Thousands of idle streams with mostly sleeping connections.

An ASGI framework such as Quart on Hypercorn or Uvicorn

An asyncio task per stream on one event loop, with async def generators and await for the waits.

The workload is mostly open connections. Needs code written for asyncio.

 
# a thread per stream
gunicorn --worker-class gthread --workers 2 --threads 100 "bookshelf:create_app()"

# greenlets: thousands of connections per process (gevent must be installed)
gunicorn --worker-class gevent --workers 2 --worker-connections 1000 "bookshelf:create_app()"

Source: Flask: Gunicorn

Further points:

  • Async views in Flask do not fix this. An async def view runs on a short-lived event loop per request in a worker thread, so it is still one worker per request. See the Async, gevent & Quart page of this section for the explanation and the details of gevent’s monkey patching.

  • Idle timeouts. Every layer between the browser and Flask has its own idle timeout (Gunicorn’s worker --timeout, the proxy’s read timeout, a load balancer’s idle timeout), and the shortest one wins. Send a keep-alive comment more often than the shortest, and check the Gunicorn worker documentation for how your worker class counts a stream that is quiet but alive.

  • Scale out with a shared bus. The in-process broker above lives in one process. With several workers or hosts, a review posted to worker A never reaches subscribers connected to worker B. The usual answer is Redis publish/subscribe (or a message broker, see Messaging): the POST handler publishes to a channel, each stream generator subscribes to it. Redis itself is introduced in Redis.

  • Eventlet is not an option for new projects: the Flask documentation notes that it is no longer maintained, so gevent is the greenlet choice.

import threading
import time

import fakeredis                     # a local fake: use redis.Redis.from_url(os.environ["REDIS_URL"]) in production
from flask import Flask, Response

app = Flask(__name__)
bus = fakeredis.FakeRedis()


@app.get("/feed")
def feed():
    def generate():
        pubsub = bus.pubsub()
        pubsub.subscribe("reviews")
        try:
            yield ": connected\n\n"
            while True:
                message = pubsub.get_message(ignore_subscribe_messages=True, timeout=1.0)
                if message is None:
                    yield ": keep-alive\n\n"
                    continue
                yield f"event: review\ndata: {message['data'].decode()}\n\n"
        finally:
            pubsub.close()

    return Response(generate(), mimetype="text/event-stream", headers={"X-Accel-Buffering": "no"})


response = app.test_client().get("/feed")
stream = response.iter_encoded()
assert next(stream) == b": connected\n\n"
threading.Timer(0.1, lambda: bus.publish("reviews", '{"stars": 5}')).start()      # another worker publishes
for chunk in stream:
    if b"event: review" in chunk:
        print(chunk.decode().strip())
        break
response.close()
time.sleep(0.2)

Source: redis-py: Advanced features (Publish / Subscribe)

Proxies, Buffering and Timeouts

A streaming response that works with flask run can appear to hang behind a reverse proxy. The proxy is a second buffering layer: nginx, for example, reads the upstream response into buffers (proxy_buffering is on by default) and sends it to the client when the buffers fill or the response ends, so small SSE events arrive in bursts or not at all. The remedies are:

  • send the X-Accel-Buffering: no response header from the view, as the examples do; nginx honours it per response unless proxy_ignore_headers disables it;

  • or turn proxy_buffering off for the location;

  • raise proxy_read_timeout (60 seconds by default) above your keep-alive interval, or send comments more often than that: the timeout is between two successive reads, so a keep-alive every 15 seconds keeps the connection open;

  • do not compress text/event-stream responses at the proxy unless the proxy flushes after each event.

location /reviews/stream {
    proxy_pass http://bookshelf_upstream;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    proxy_buffering off;
    proxy_read_timeout 1h;
}

Source: nginx: ngx_http_proxy_module

The general nginx deployment of a Flask application is described in Flask: nginx; CDN and load-balancer products have their own buffering and idle-timeout settings, so test the real chain with curl -N (which disables curl’s own buffering).

curl -N -H "Accept: text/event-stream" http://localhost:8000/reviews/stream

Source: Flask: Gunicorn

Many Clients: Quart

When most of the workload is open connections, the framework most similar to Flask is Quart, an ASGI reimplementation of the Flask API that shares its Sans-IO classes with Flask and needs Python 3.13 or newer in its current release (0.23.1 at the time of writing). Views are async def, the request body is read with await request.get_json(), and a streaming response uses an asynchronous generator, so a thousand waiting streams cost a thousand small tasks instead of a thousand threads. The server is an ASGI server such as Hypercorn. Two differences matter for SSE: Quart times out long responses by default, to protect against slow-response denial of service, and the how-to asks you to disable that for the stream’s response only (response.timeout = None); and the code inside the stream must not block (use await asyncio.sleep and asynchronous clients, never time.sleep or a blocking Redis client). The migration recipe from a Flask application is in the Async, gevent & Quart page of this section.

import asyncio
import json

from quart import Quart, make_response

app = Quart(__name__)


@app.get("/reviews/stream")
async def review_stream():
    async def events():
        for n in range(1, 4):                       # a real feed would wait on an asyncio.Queue
            await asyncio.sleep(0.01)
            yield f"event: review\nid: {n}\ndata: {json.dumps({'stars': n})}\n\n".encode()

    response = await make_response(
        events(),
        {"Content-Type": "text/event-stream", "Cache-Control": "no-cache"},
    )
    response.timeout = None                         # long-lived on purpose, for this response only
    return response


async def main():
    response = await app.test_client().get("/reviews/stream")
    body = await response.get_data(as_text=True)
    assert body.count("event: review") == 3
    print(body.splitlines()[0], "...", len(body.splitlines()), "lines")


asyncio.run(main())

Source: Quart: Server Sent Events

WebSockets: Overview

SSE covers server-to-browser pushes. When both sides send messages freely (chat, collaborative editing, games), the protocol is WebSocket (RFC 6455): an HTTP request is upgraded to a persistent two-way connection. WSGI has no model for that, so Flask itself does not offer WebSockets; there are two common extensions, and both are covered only at overview level here, because real-time communication is a topic of its own (chat and videoconferencing are tracked in issue #264 of this site’s repository). Frameworks with native WebSocket handlers are documented in the Tornado WebSockets and FastAPI WebSockets pages, and Tornado’s streaming, long polling and SSE page shows the same streaming techniques on an asyncio-native server.

flask-sock Flask-SocketIO

What it is

A thin WebSocket route decorator on top of the simple-websocket library.

The Socket.IO protocol (events, rooms, acknowledgements, fallbacks) over WebSocket, built on python-socketio.

API

Sock(app); @sock.route("/path") on a function that receives ws; ws.send(), ws.receive(), ws.close().

SocketIO(app); @socketio.on("event") handlers; emit(); rooms; socketio.run(app).

Client

The browser’s native WebSocket.

The Socket.IO client library (a different protocol: a plain WebSocket client cannot talk to it).

Servers

The Werkzeug development server, Gunicorn with threads, gevent or eventlet workers.

Async modes threading, eventlet and gevent; Gunicorn is the documented production server.

Scaling

Same worker-per-connection limits as SSE; one thread or greenlet per socket.

A message queue (such as Redis) is needed to coordinate several workers; set cors_allowed_origins explicitly.

Use it when

You need raw WebSocket messages and want the smallest dependency.

You want rooms, broadcast, reconnect and fallback handled for you.

 
from flask import Flask
from flask_sock import Sock
from flask_socketio import SocketIO, emit

app = Flask(__name__)

sock = Sock(app)


@sock.route("/ws/echo")
def echo(ws):
    while True:
        message = ws.receive()          # blocks this worker for the life of the connection
        ws.send(message)


socketio = SocketIO(app)                 # set cors_allowed_origins explicitly when the page is on another origin


@socketio.on("new_review")
def handle_new_review(data):
    emit("review", {"stars": data["stars"]}, broadcast=True)


assert "/ws/echo" in [rule.rule for rule in app.url_map.iter_rules()]

client = socketio.test_client(app)
client.emit("new_review", {"stars": 5})
received = client.get_received()
print(received[0]["name"], received[0]["args"])

Source: Flask-SocketIO documentation

Both extensions need a server that supports them, and neither works with flask run plus a plain single-threaded WSGI worker. Their documentation covers deployment in detail: flask-sock: Web servers and Flask-SocketIO: Deployment. Authenticate in the connection handshake (a cookie works, as for SSE), check the Origin header (cross-site WebSocket hijacking is the real-time version of CSRF), and apply message size and rate limits. A choice guide:

Need Long polling SSE WebSocket

Server pushes to browser

yes

yes

yes

Browser sends frequently on the same channel

no (separate requests)

no (separate requests)

yes

Works through any HTTP proxy

yes

usually, with buffering off

needs upgrade support

Automatic reconnect and resume

build it

built in (Last-Event-ID)

build it, or Socket.IO

Binary messages

yes

no (text only)

yes

Effort in Flask

low

low

extension and a suitable server

Summary of Pitfalls

Symptom Cause and fix

RuntimeError: Working outside of request context in a generator

The contexts are popped before the generator runs. Wrap it with stream_with_context, or pass plain values.

Session changes made while streaming are lost

Set-Cookie has already been sent. Change the session before the stream starts.

Events arrive in bursts or only when the connection closes

A proxy or middleware buffers. Send X-Accel-Buffering: no, disable proxy_buffering, and check compression and profiler middleware.

The application freezes with a few clients

Each stream holds a sync worker. Use threads, gevent workers or Quart, and cap the number of streams.

The browser reconnects every 60 seconds

An idle timeout in the proxy or load balancer. Send a keep-alive comment more often than the timeout.

Subscribers on other workers see nothing

The broker is in-process. Use a shared bus such as Redis publish/subscribe.

Corrupted CSV, or a formula executes in a spreadsheet

CSV was built by joining strings. Use csv.writer and neutralize cells that start with a formula character.

Truncated download without an error page

An exception inside the generator after the headers were sent. Catch it inside, and stream an error record or a trailer line.

Connections or cursors never released

Cleanup placed after the loop. Put it in finally, which also runs when the client disconnects.

Recent Changes and Where the Books Are Outdated

  • Flask 2.2 (2022) added stream_template and stream_template_string; before it, tutorials streamed a template with the Jinja environment and stream_with_context by hand. Flask 3.1 keeps both functions unchanged.

  • The generator form return generate(), {"Content-Type": "text/csv"} (a generator as the first item of the return tuple) is the form in the current official pattern page; older articles wrap everything in Response(...), which still works and is used here for clarity.

  • stream_with_context as a decorator is what the current API reference shows; the wrapper form from old tutorials still works.

  • Flask 3.2 (unreleased) merges the application and request contexts; nothing in the examples depends on the split, and the generator behaviour described here is expected to stay.

  • Books. The four Flask books cross-checked for this section contain no chapter on streaming or SSE; one only mentions async and Quart in passing and another runs background work in threads. Do not replace streaming with a thread that writes to a global list, and do not follow the usual blog recipe of one while True loop per client on flask run. Eventlet-based real-time examples in older books are superseded by gevent, because Flask’s docs say eventlet is no longer maintained.

References