MQTT avec Flask (flask-mqtt et paho)

Flask n'a pas d'event loop et traite chaque requête dans un thread : faire du MQTT dans Flask revient donc à faire tourner un client MQTT à côté du serveur web. Un thread réseau en arrière-plan reçoit les messages, et les handlers de requêtes publient via le même client. Vous pouvez le faire avec l'extension flask-mqtt ou directement avec paho-mqtt 2.x. Les deux fonctionnent, et les deux partagent les mêmes pièges liés aux threads et aux processus multiples — le vrai sujet de ce guide.

Nous allons configurer chaque option, régler le problème du « chaque message arrive deux fois » et terminer par un petit tableau de bord en direct qui envoie les données MQTT au navigateur via Server-Sent Events.

Option 1 : flask-mqtt

flask-mqtt enveloppe paho-mqtt et lit sa configuration dans app.config :

pip install flask flask-mqtt

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}

C'est rapide à mettre en place, mais gardez deux points en tête. L'extension est peu maintenue, et ses anciennes versions ont été écrites pour l'API de callbacks de paho-mqtt 1.x : si vous obtenez des erreurs de signature de callback après l'installation de paho-mqtt 2.0, cherchez une version plus récente de flask-mqtt ou figez paho-mqtt<2. Par ailleurs, elle se connecte dès l'exécution de Mqtt(app) : chaque processus qui importe l'application ouvre donc une connexion (voir plus bas).

Option 2 : paho-mqtt 2.x directement

Utiliser paho directement demande quelques lignes de plus et vous donne l'API actuelle, les reconnexions automatiques et un contrôle total. Placez le client dans son propre module :

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 par processus
        )
        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()          # thread réseau, se reconnecte tout seul

    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)  # rejoué après chaque reconnexion

    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                        # navigateur lent : on jette, sans jamais bloquer le thread réseau

    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

Avec connect_async() et loop_start(), Flask démarre même si le broker est injoignable, et paho continue de réessayer en arrière-plan avec le backoff défini par reconnect_delay_set(). S'abonner dans on_connect rétablit l'abonnement après chaque reconnexion. Pour les bases de l'API paho, voir MQTT en Python.

Règles de threads

Dans ce montage, trois threads touchent aux données MQTT ; quelques règles suffisent à rester sûr :

Le travail lourd — écritures en base, appels HTTP vers d'autres services — doit passer par une file consommée par un thread worker ou un gestionnaire de tâches comme Celery ou RQ, pas par on_message.

Le problème du « chaque message deux fois »

Le bug MQTT le plus courant avec Flask : des messages traités deux fois, ou un log qui montre le client se connecter et se déconnecter en boucle. Causes et solutions :

• Reloader de debug. app.run(debug=True) exécute votre module dans un processus de surveillance et dans un processus qui sert : vous avez deux clients MQTT. Utilisez use_reloader=False, ou démarrez MQTT uniquement dans le processus qui sert : if os.environ.get("WERKZEUG_RUN_MAIN") == "true" or not app.debug:.
• Plusieurs workers Gunicorn. -w 4 signifie quatre clients. Pour un tableau de bord, c'est correct (chaque worker a besoin de tous les messages pour ses navigateurs), mais tout ce qui ne doit se produire qu'une fois — une insertion en base, une alerte — doit passer par un abonnement partagé, pour que le broker remette chaque message à un seul worker. Voir abonnements partagés.
• Client ID fixe. Si tous ces processus utilisent le même client ID, le broker coupe l'ancienne connexion à chaque nouvelle arrivée, et ils s'éjectent mutuellement sans fin. Des suffixes aléatoires évitent cela — voir client ID dupliqué.
• gunicorn --preload. L'application est importée dans le master avant le fork, mais le thread paho ne survit pas à fork() : les workers héritent d'un objet client sans boucle réseau. N'utilisez pas le preload, ou démarrez le client dans un hook post_fork.

Un tableau de bord en direct simple

Branchez maintenant le service sur Flask : un instantané JSON, un endpoint de commande, un flux SSE et une petite page HTML.

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>Télémétrie en direct</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)                          # jamais de saisie brute dans un 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"  # empêche les proxys de fermer le flux
        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)

Chaque tableau de bord ouvert maintient une connexion SSE et, avec le serveur multithread, cela fait un thread par navigateur — suffisant pour un tableau de bord interne avec quelques spectateurs. En production, lancez gunicorn -w 1 --threads 16 app:app (ou davantage de workers en respectant les règles ci-dessus). Pour de nombreux spectateurs simultanés, un framework asynchrone convient mieux ; voir MQTT avec FastAPI, ou laissez les navigateurs se connecter directement au broker comme dans le tutoriel de tableau de bord en direct.

Testez

1. Lancez python app.py et ouvrez http://localhost:5000.
2. Dans le client MQTT en ligne, connectez-vous à broker.connectmq.com et publiez {"temp": 23.1} sur testmqtt/flask/greenhouse/telemetry. Une ligne doit apparaître immédiatement dans le tableau de bord.
3. Abonnez-vous à testmqtt/flask/# dans le client en ligne, puis envoyez une commande :

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

1. Vérifiez que curl localhost:5000/api/latest renvoie la dernière valeur par topic.
2. Arrêtez le broker ou coupez le réseau : l'endpoint de commande doit répondre 503, et le tableau de bord doit reprendre seul dès que paho s'est reconnecté.

Envoyer de la télémétrie de test à votre dashboard Flask

Si votre projet utilise Django plutôt que Flask, les mêmes principes s'appliquent avec une management command comme subscriber de longue durée — voir MQTT avec Django.

MQTT dans Flask fonctionne bien dès qu'on le traite pour ce qu'il est : un client durable dans un thread d'arrière-plan, un par processus. Démarrez-le une seule fois, gardez des callbacks courts et thread-safe, donnez à chaque processus un client ID unique et testez la boucle complète — de l'appareil au navigateur et du navigateur à l'appareil — avant d'ajouter des workers.

Questions fréquentes