QuickstartΒΆ

Here’s a quick working example of how Flask-SSE works.

Warning

Server-sent events do not work with Flask’s built-in development server, because it handles HTTP requests one at a time. The SSE stream is intended to be an infinite stream of events, so it will never complete. If you try to run this code on with the built-in development server, the server will be unable to take any other requests once you connect to this stream. Instead, you must use a web server with asychronous workers. Gunicorn can work with gevent to use asychronous workers: see gunicorn’s design documentation.

You will also need a Redis server running locally for this example to work.

Make a virtual environment and install Flask-SSE, gunicorn, and gevent. You will also need to make sure you have a Redis server running locally.

$ pyvenv sse
$ pip install flask-sse gunicorn gevent

Make a file on your computer called sse.py, with the following content:

from flask import Flask, render_template
from flask_sse import sse

app = Flask(__name__)
app.config["REDIS_URL"] = "redis://localhost"
app.register_blueprint(sse, url_prefix='/stream')

@app.route('/')
def index():
    return render_template("index.html")

@app.route('/hello')
def publish_hello():
    sse.publish({"message": "Hello!"}, type='greeting')
    return "Message sent!"

If you are using a Redis server that has a password use:

app.config["REDIS_URL"] = "redis://:password@localhost"

Make a templates folder next to sse.py, and create a file named index.html in that folder, with the following content:

<!DOCTYPE html>
<html>
<head>
  <title>Flask-SSE Quickstart</title>
</head>
<body>
  <h1>Flask-SSE Quickstart</h1>
  <script>
    var source = new EventSource("{{ url_for('sse.stream') }}");
    source.addEventListener('greeting', function(event) {
        var data = JSON.parse(event.data);
        alert("The server says " + data.message);
    }, false);
    source.addEventListener('error', function(event) {
        alert("Failed to connect to event stream. Is Redis running?");
    }, false);
  </script>
</body>
</html>

Run your code using gunicorn’s gevent workers:

$ gunicorn sse:app --worker-class gevent --bind 127.0.0.1:8000

Open your web browser, and visit 127.0.0.1:8000. Your web browser will automatically connect to the server-sent event stream. Open another tab, and visit 127.0.0.1:8000/hello. You should get a Javascript alert in the first tab when you do so.