MQTT mit Flask (flask-mqtt und paho)

Flask hat keinen eigenen Event Loop und bearbeitet jeden Request in einem Thread. MQTT in Flask heißt deshalb: Ein MQTT-Client läuft neben dem Webserver, ein Netzwerk-Thread im Hintergrund empfängt Nachrichten, und Request-Handler publishen über denselben Client. Das geht mit der Erweiterung flask-mqtt oder direkt mit paho-mqtt 2.x. Beides funktioniert, und beide teilen dieselben Fallstricke bei Threads und mehreren Prozessen — das eigentliche Thema dieser Anleitung.

Wir richten beide Varianten ein, lösen das Problem „jede Nachricht kommt doppelt“ und bauen zum Schluss ein kleines Live-Dashboard, das MQTT-Daten per Server-Sent Events in den Browser streamt.

Option 1: flask-mqtt

flask-mqtt kapselt paho-mqtt und liest die Einstellungen aus 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}

Das ist schnell eingerichtet, aber zwei Dinge solltest du wissen. Die Erweiterung wird nur wenig gepflegt, und ältere Releases wurden für die Callback-API von paho-mqtt 1.x geschrieben — bekommst du nach der Installation von paho-mqtt 2.0 Fehler zu Callback-Signaturen, suche nach einem neueren flask-mqtt-Release oder pinne paho-mqtt<2. Außerdem verbindet sie sich, sobald Mqtt(app) ausgeführt wird; jeder Prozess, der die App importiert, öffnet also eine Verbindung (dazu gleich mehr).

Option 2: paho-mqtt 2.x direkt

Mit paho direkt brauchst du ein paar Zeilen mehr, bekommst aber die aktuelle API, automatische Reconnects und volle Kontrolle. Leg den Client in ein eigenes Modul:

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]}",  # eindeutig pro Prozess
        )
        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()          # Netzwerk-Thread, verbindet sich selbst neu

    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)  # läuft nach jedem Reconnect erneut

    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                        # langsamer Browser: verwerfen, nie den Netzwerk-Thread blockieren

    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() sorgt dafür, dass Flask auch startet, wenn der Broker nicht erreichbar ist; paho versucht es im Hintergrund weiter, mit dem Backoff aus reconnect_delay_set(). Das Abonnieren in on_connect stellt die Subscription nach jedem Reconnect wieder her. Die Grundlagen der paho-API findest du unter MQTT in Python.

Threading-Regeln

In diesem Aufbau greifen drei Threads auf MQTT-Daten zu. Ein paar Regeln halten das sicher:

Schwere Arbeit — Datenbank-Writes, HTTP-Aufrufe an andere Dienste — gehört in eine Queue, die ein Worker-Thread oder eine Task-Queue wie Celery oder RQ abarbeitet, nicht in on_message.

Das Problem „jede Nachricht doppelt“

Der häufigste Flask-MQTT-Bug: Nachrichten werden doppelt verarbeitet, oder das Log zeigt den Client in einer Schleife aus Verbinden und Trennen. Ursachen und Lösungen:

• Debug-Reloader. app.run(debug=True) führt dein Modul in einem Überwachungsprozess und einem ausliefernden Prozess aus — du hast zwei MQTT-Clients. Nutze use_reloader=False oder starte MQTT nur im ausliefernden Prozess: if os.environ.get("WERKZEUG_RUN_MAIN") == "true" or not app.debug:.
• Mehrere Gunicorn-Worker. -w 4 bedeutet vier Clients. Für ein Dashboard ist das in Ordnung (jeder Worker braucht alle Nachrichten für seine Browser), aber alles, was nur einmal passieren darf — ein Datenbank-Insert, ein Alarm —, sollte eine Shared Subscription nutzen, damit der Broker jede Nachricht nur einem Worker gibt. Siehe Shared Subscriptions.
• Feste Client-ID. Nutzen alle Prozesse dieselbe Client-ID, trennt der Broker die alte Verbindung jedes Mal, wenn eine neue kommt, und sie werfen sich endlos gegenseitig raus. Zufällige Suffixe verhindern das — siehe doppelte Client-ID.
• gunicorn --preload. Die App wird vor dem Fork im Master importiert, aber der paho-Thread überlebt fork() nicht — die Worker haben ein Client-Objekt ohne Netzwerk-Loop. Verzichte auf Preload oder starte den Client in einem post_fork-Hook.

Ein einfaches Live-Dashboard

Jetzt verbindest du den Service mit Flask: ein JSON-Snapshot, ein Kommando-Endpoint, ein SSE-Stream und eine kleine HTML-Seite.

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-Telemetrie</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)                          # nie ungeprüfte Eingaben in ein Topic schreiben
    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"  # verhindert, dass Proxys den Stream schließen
        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)

Jedes offene Dashboard hält eine SSE-Verbindung, und mit dem Thread-Server ist das ein Thread pro Browser — ausreichend für ein internes Dashboard mit wenigen Betrachtern. In Produktion startest du gunicorn -w 1 --threads 16 app:app (oder mehr Worker unter Beachtung der Regeln oben). Für viele gleichzeitige Betrachter passt ein asynchrones Framework besser; siehe MQTT mit FastAPI, oder lass die Browser direkt mit dem Broker verbinden wie im Live-Dashboard-Tutorial.

Testen

1. Starte python app.py und öffne http://localhost:5000.
2. Verbinde dich im Online-MQTT-Client mit broker.connectmq.com und publishe {"temp": 23.1} auf testmqtt/flask/greenhouse/telemetry. Im Dashboard sollte sofort eine Zeile erscheinen.
3. Abonniere im Online-Client testmqtt/flask/# und sende ein Kommando:

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

1. Prüfe, dass curl localhost:5000/api/latest den letzten Wert pro Topic liefert.
2. Stoppe den Broker oder trenn das Netzwerk: Der Kommando-Endpoint sollte 503 liefern, und das Dashboard sollte von selbst weiterlaufen, sobald paho neu verbunden ist.

Test-Telemetrie an dein Flask-Dashboard senden

Nutzt dein Projekt Django statt Flask, gelten dieselben Muster mit einem Management Command als langlebigem Subscriber — siehe MQTT mit Django.

MQTT in Flask funktioniert gut, sobald du es als das behandelst, was es ist: ein langlebiger Client in einem Hintergrund-Thread, einer pro Prozess. Starte ihn einmal, halte Callbacks kurz und thread-safe, gib jedem Prozess eine eindeutige Client-ID und teste den kompletten Kreislauf — vom Gerät zum Browser und vom Browser zum Gerät —, bevor du weitere Worker hinzufügst.

Häufig gestellte Fragen