MQTT with Flask: flask-mqtt, paho-mqtt and a Live Dashboard

Q: Why does my Flask app receive every MQTT message twice?

Usually two MQTT clients are running: the debug reloader runs your code in two processes, or Gunicorn started several workers. Disable the reloader (use_reloader=False) or start MQTT only in the serving process, and give each process a unique client ID.

Q: Should I use flask-mqtt or paho-mqtt directly?

flask-mqtt is a thin wrapper that reads settings from app.config and offers decorators. It is convenient for small apps, but it is lightly maintained, so check its compatibility with paho-mqtt 2.x. Using paho-mqtt directly gives you full control and the current callback API.

Q: Can I use Flask-SQLAlchemy inside an MQTT callback?

Yes, but MQTT callbacks run in paho's network thread, outside any request, so there is no application context. Wrap the database code in with app.app_context(): and keep the work short or hand it to a queue.

Flask has no event loop of its own and handles each request in a worker thread, so MQTT in Flask means running an MQTT client next to the web server: a background network thread receives messages, and request handlers publish through the same client. You can do this with the flask-mqtt extension or with paho-mqtt 2.x directly. Both work; both share the same threading and multi-process pitfalls, which are the real subject of this guide.

We will set up each option, fix the "every message arrives twice" problem, and finish with a small live dashboard that streams MQTT data to the browser with Server-Sent Events.

Option 1: flask-mqtt

flask-mqtt wraps paho-mqtt and reads its settings from app.config:

Install

pip install flask flask-mqtt

app_flask_mqtt.py

import os
from flask import Flask
from flask_mqtt import Mqtt

app = Flask(__name__)
app.config["MQTT_BROKER_URL"] = "broker.connectmq.com"
app.config["MQTT_BROKER_PORT"] = 1883
app.config["MQTT_USERNAME"] = os.getenv("MQTT_USER", "")
app.config["MQTT_PASSWORD"] = os.getenv("MQTT_PASS", "")
app.config["MQTT_KEEPALIVE"] = 30
app.config["MQTT_TLS_ENABLED"] = False
mqtt = Mqtt(app)

@mqtt.on_connect()
def handle_connect(client, userdata, flags, rc):
    mqtt.subscribe("testmqtt/flask/#")

@mqtt.on_message()
def handle_message(client, userdata, message):
    print(message.topic, message.payload.decode())

@app.post("/ping")
def ping():
    mqtt.publish("testmqtt/flask/ping", "pong")
    return {"ok": True}

It is quick to set up, but keep two things in mind. The extension is lightly maintained, and older releases were written for the paho-mqtt 1.x callback API — if you hit callback-signature errors after paho-mqtt 2.0 was installed, check for a newer flask-mqtt release or pin paho-mqtt<2. And it connects when Mqtt(app) runs, so every process that imports the app opens a connection (more on that below).

Option 2: paho-mqtt 2.x directly

Using paho directly takes a few more lines and gives you the current API, automatic reconnects and full control. Put the client in its own module:

mqtt_service.py

import os
import queue
import threading
import uuid
import paho.mqtt.client as mqtt

TELEMETRY = "testmqtt/flask/+/telemetry"


class MqttService:
    def __init__(self, host="broker.connectmq.com", port=1883):
        self.host, self.port = host, port
        self.latest: dict[str, str] = {}
        self.listeners: list[queue.Queue] = []
        self.lock = threading.Lock()
        self.client = mqtt.Client(
            mqtt.CallbackAPIVersion.VERSION2,
            client_id=f"testmqtt-flask-{uuid.uuid4().hex[:8]}",  # unique per process
        )
        if os.getenv("MQTT_USER"):
            self.client.username_pw_set(os.environ["MQTT_USER"], os.getenv("MQTT_PASS"))
        self.client.reconnect_delay_set(min_delay=1, max_delay=30)
        self.client.on_connect = self._on_connect
        self.client.on_message = self._on_message

    def start(self):
        self.client.connect_async(self.host, self.port, keepalive=30)
        self.client.loop_start()          # network thread, reconnects automatically

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        if reason_code.is_failure:
            print("MQTT connect failed:", reason_code)
            return
        client.subscribe(TELEMETRY, qos=1)  # runs again after every reconnect

    def _on_message(self, client, userdata, msg):
        item = {"topic": msg.topic, "payload": msg.payload.decode(errors="replace")}
        with self.lock:
            self.latest[msg.topic] = item["payload"]
            listeners = list(self.listeners)
        for q in listeners:
            try:
                q.put_nowait(item)
            except queue.Full:
                pass                        # slow browser: drop, never block the network thread

    def join(self) -> queue.Queue:
        q: queue.Queue = queue.Queue(maxsize=100)
        with self.lock:
            self.listeners.append(q)
        return q

    def leave(self, q: queue.Queue):
        with self.lock:
            self.listeners.remove(q)

    def publish(self, topic: str, payload: str, qos: int = 1) -> bool:
        if not self.client.is_connected():
            return False
        return self.client.publish(topic, payload, qos=qos).rc == mqtt.MQTT_ERR_SUCCESS

connect_async() plus loop_start() means Flask starts even if the broker is unreachable, and paho keeps retrying in the background with the backoff set by reconnect_delay_set(). Subscribing inside on_connect restores the subscription after every reconnect. For the basics of the paho API, see MQTT in Python.

Threading rules

Three threads touch MQTT data in this setup, so a few rules keep it safe:

Rule	Why
Keep `on_message` short	It runs on paho's single network thread. A slow handler delays every message and can make keep-alive pings late.
Protect shared state with a lock	Request threads read `latest` while the network thread writes it.
`publish()` from request handlers is fine	paho's `publish()` is thread-safe; check `is_connected()` first and answer `503` otherwise.
No app context in callbacks	`current_app`, `g` and Flask-SQLAlchemy sessions need `with app.app_context():` inside the callback.
Never block on `wait_for_publish()` without a timeout	A disconnected broker would hang the request thread forever.

Heavy work — database writes, HTTP calls to other services — belongs in a queue consumed by a worker thread or a task queue such as Celery or RQ, not in on_message.

The "every message twice" problem

The most common Flask MQTT bug: messages are processed twice, or the log shows the client connecting and disconnecting in a loop. Causes and fixes:

Debug reloader. app.run(debug=True) runs your module in a watcher process and a serving process, so you get two MQTT clients. Use use_reloader=False, or start MQTT only in the serving process: if os.environ.get("WERKZEUG_RUN_MAIN") == "true" or not app.debug:.
Several Gunicorn workers. -w 4 means four clients. That is fine for a dashboard (every worker needs all messages for its own browsers), but anything that must happen once — a database insert, an alert — should use a shared subscription so the broker hands each message to only one worker. See shared subscriptions.
Fixed client ID. If all those processes use the same client ID, the broker disconnects the old connection every time a new one arrives, and they kick each other off forever. Random suffixes avoid that — see duplicate client ID.
gunicorn --preload. The app is imported in the master before forking, but the paho thread does not survive fork(), so workers have a client object with no network loop. Do not preload, or start the client in a post_fork hook.

A simple live dashboard

Now wire the service into Flask: a JSON snapshot, a command endpoint, an SSE stream and a tiny HTML page.

app.py

import json
import queue
import re
from flask import Flask, Response, abort, jsonify, request
from mqtt_service import MqttService

app = Flask(__name__)
mqtt_service = MqttService()
mqtt_service.start()
DEVICE_ID = re.compile(r"^[A-Za-z0-9_-]{1,64}$")

PAGE = """<!doctype html><title>MQTT dashboard</title>
<h1>Live telemetry</h1><table id="t"></table>
<script>
  const rows = {};
  new EventSource("/stream").onmessage = (e) => {
    const { topic, payload } = JSON.parse(e.data);
    let row = rows[topic];
    if (!row) {
      row = rows[topic] = document.getElementById("t").insertRow();
      row.insertCell().textContent = topic;
      row.insertCell();
    }
    row.cells[1].textContent = payload;
  };
</script>"""


@app.get("/")
def index():
    return PAGE


@app.get("/api/latest")
def latest():
    with mqtt_service.lock:
        return jsonify(mqtt_service.latest)


@app.post("/api/devices/<device_id>/command")
def command(device_id):
    if not DEVICE_ID.match(device_id):
        abort(400)                          # never put raw input into a topic
    body = json.dumps(request.get_json(force=True))
    if not mqtt_service.publish(f"testmqtt/flask/{device_id}/command", body):
        abort(503)
    return jsonify(queued=True), 202


@app.get("/stream")
def stream():
    q = mqtt_service.join()

    def events():
        try:
            while True:
                try:
                    yield f"data: {json.dumps(q.get(timeout=15))}\n\n"
                except queue.Empty:
                    yield ": keep-alive\n\n"  # keeps proxies from closing the stream
        finally:
            mqtt_service.leave(q)

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


if __name__ == "__main__":
    app.run(debug=True, use_reloader=False, threaded=True)

Each open dashboard holds one SSE connection, and with the threaded server that is one thread per browser — fine for an internal dashboard with a handful of viewers. In production run gunicorn -w 1 --threads 16 app:app (or more workers combined with the rules above). For many concurrent viewers, an async framework fits better; see MQTT with FastAPI, or let browsers connect to the broker directly as shown in the live dashboard tutorial.

Test it

Run python app.py and open http://localhost:5000.
In the online MQTT client, connect to broker.connectmq.com and publish {"temp": 23.1} to testmqtt/flask/greenhouse/telemetry. A row should appear in the dashboard immediately.
Subscribe to testmqtt/flask/# in the online client, then send a command:

HTTP → MQTT

curl -X POST localhost:5000/api/devices/greenhouse/command \
  -H 'content-type: application/json' -d '{"fan": "on"}'

Check curl localhost:5000/api/latest returns the last value per topic.
Stop the broker or cut the network: the command endpoint should answer 503, and the dashboard should resume on its own once paho reconnects.

Send test telemetry to your Flask dashboard

If your project is on Django rather than Flask, the same patterns apply with a management command as the long-running subscriber — see MQTT with Django.

MQTT in Flask works well once you treat it as what it is: a long-lived client in a background thread, one per process. Start it once, keep callbacks short and thread-safe, give every process a unique client ID, and test the full loop — device to browser and browser to device — before adding more workers.

Frequently asked questions

Why does my Flask app receive every MQTT message twice?

Usually two MQTT clients are running: the debug reloader runs your code in two processes, or Gunicorn started several workers. Disable the reloader (use_reloader=False) or start MQTT only in the serving process, and give each process a unique client ID.

Should I use flask-mqtt or paho-mqtt directly?

flask-mqtt is a thin wrapper that reads settings from app.config and offers decorators. It is convenient for small apps, but it is lightly maintained, so check its compatibility with paho-mqtt 2.x. Using paho-mqtt directly gives you full control and the current callback API.

Can I use Flask-SQLAlchemy inside an MQTT callback?

Yes, but MQTT callbacks run in paho's network thread, outside any request, so there is no application context. Wrap the database code in with app.app_context(): and keep the work short or hand it to a queue.

MQTT with Flask (flask-mqtt and paho)