MQTT zu REST API mit Flask: Tutorial für Python

Mit einer MQTT zu REST API in Flask kann jede Anwendung, die HTTP spricht, Live-Gerätedaten lesen und Befehle senden, ohne MQTT lernen zu müssen. In diesem Tutorial schreiben Sie eine Flask-App, die Geräte-Topics abonniert, den jeweils letzten Wert pro Gerät im Speicher hält, ihn über JSON-Endpunkte bereitstellt und Befehle aus einer POST-Route publiziert. Außerdem lernen Sie die zwei Deployment-Fehler kennen, die die meisten Kombinationen aus Flask und MQTT unbemerkt lahmlegen.

Warum eine REST API vor MQTT setzen

MQTT ist das richtige Protokoll für Geräte: persistente Verbindungen, kleine Pakete und Push-Zustellung. Viele Konsumenten dieser Daten sind jedoch keine Geräte. Das Backend einer Mobile-App, eine MES- oder ERP-Integration, ein Reporting-Skript oder ein Partnersystem erwarten in der Regel Request und Response über HTTP. In der Fertigung ist das typische Szenario ein MES, das den aktuellen Maschinenzustand per REST abfragt, während die Maschinendaten selbst in einem Unified Namespace (UNS) auf dem Broker liegen. Statt jedem dieser Systeme MQTT beizubringen, übersetzt eine schlanke Bridge:

Konsument möchte	REST-Aufruf	Was die Bridge tut
Aktuellen Zustand aller Geräte	`GET /api/devices`	Liefert die zuletzt gecachte Nachricht pro Gerät
Letzten Messwert eines Geräts	`GET /api/devices/<id>`	Liefert einen gecachten Eintrag samt Alter
Einen Befehl senden	`POST /api/devices/<id>/command`	Publiziert JSON auf das Befehls-Topic des Geräts
Health Check	`GET /health`	Meldet, ob MQTT verbunden ist

Die Bridge ist ein „Last Value Cache". Sie ist kein Historian: Sie beantwortet „Welchen Wert gibt es jetzt?", nicht „Welcher Wert war es letzten Dienstag?". Für die Historie speichern Sie den Datenstrom zusätzlich zum Cache in einer Datenbank.

Was Sie brauchen

Python 3.9+
pip install "flask>=3.0" "paho-mqtt>=2.0"
Einen Broker. Für die Entwicklung nutzen Sie den kostenlosen öffentlichen MQTT-Broker broker.connectmq.com mit einem zufälligen Topic-Präfix; für die Produktion einen privaten Broker mit Zugangsdaten.

Die Flask-MQTT-Bridge

Speichern Sie dies als app.py:

import json
import os
import threading
import time
import uuid
from functools import wraps

import paho.mqtt.client as mqtt
from flask import Flask, abort, jsonify, request

BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USERNAME = os.environ.get("MQTT_USERNAME")
PASSWORD = os.environ.get("MQTT_PASSWORD")
CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", f"flask-bridge-{uuid.uuid4().hex[:8]}")
PREFIX = os.environ.get("TOPIC_PREFIX", "connectmq/demo/flask-3391")
API_KEY = os.environ.get("API_KEY", "change-me")
STALE_AFTER = int(os.environ.get("STALE_AFTER", "120"))  # Sekunden

app = Flask(__name__)
cache = {}                 # device -> {"data": ..., "topic": ..., "received_at": ...}
cache_lock = threading.Lock()
mqtt_connected = threading.Event()


# ---------- MQTT ----------

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"MQTT connect failed: {reason_code}")
        return
    mqtt_connected.set()
    client.subscribe(f"{PREFIX}/+/state", qos=1)


def on_disconnect(client, userdata, flags, reason_code, properties):
    mqtt_connected.clear()


def on_message(client, userdata, msg):
    device = msg.topic.split("/")[-2]
    try:
        data = json.loads(msg.payload)
    except (json.JSONDecodeError, UnicodeDecodeError):
        data = msg.payload.decode("utf-8", errors="replace")
    with cache_lock:
        cache[device] = {"data": data, "topic": msg.topic, "received_at": time.time()}


mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
if USERNAME:
    mqttc.username_pw_set(USERNAME, PASSWORD)
if PORT == 8883:
    mqttc.tls_set()
mqttc.on_connect = on_connect
mqttc.on_disconnect = on_disconnect
mqttc.on_message = on_message
mqttc.connect_async(BROKER, PORT, keepalive=60)
mqttc.loop_start()


# ---------- HTTP ----------

def require_api_key(view):
    @wraps(view)
    def wrapper(*args, **kwargs):
        if request.headers.get("X-API-Key") != API_KEY:
            abort(401)
        return view(*args, **kwargs)
    return wrapper


def serialize(device, entry):
    age = time.time() - entry["received_at"]
    return {
        "device": device,
        "data": entry["data"],
        "topic": entry["topic"],
        "age_seconds": round(age, 1),
        "stale": age > STALE_AFTER,
    }


@app.get("/health")
def health():
    ok = mqtt_connected.is_set()
    return jsonify({"mqtt_connected": ok}), (200 if ok else 503)


@app.get("/api/devices")
@require_api_key
def list_devices():
    with cache_lock:
        items = [serialize(d, e) for d, e in sorted(cache.items())]
    return jsonify(items)


@app.get("/api/devices/<device>")
@require_api_key
def get_device(device):
    with cache_lock:
        entry = cache.get(device)
        if entry is None:
            abort(404)
        return jsonify(serialize(device, entry))


@app.post("/api/devices/<device>/command")
@require_api_key
def send_command(device):
    if any(ch in device for ch in "/+#"):
        abort(400)
    body = request.get_json(silent=True)
    if not isinstance(body, dict) or "action" not in body:
        return jsonify({"error": "JSON body with an 'action' field is required"}), 400
    if not mqtt_connected.is_set():
        return jsonify({"error": "MQTT broker not connected"}), 503

    body["request_id"] = uuid.uuid4().hex
    info = mqttc.publish(f"{PREFIX}/{device}/cmd", json.dumps(body), qos=1)
    try:
        info.wait_for_publish(timeout=5)
    except (ValueError, RuntimeError):
        pass
    if not info.is_published():
        return jsonify({"error": "publish not acknowledged"}), 504
    return jsonify({"published": True, "request_id": body["request_id"]}), 202


if __name__ == "__main__":
    # use_reloader=False ist entscheidend: Der Reloader würde einen zweiten MQTT-Client starten
    app.run(host="0.0.0.0", port=5000, debug=True, use_reloader=False)

Wichtige Implementierungsdetails

Der MQTT-Client startet beim Import mit connect_async und loop_start und läuft damit in einem eigenen Thread – egal, ob Sie die App mit python app.py oder über einen WSGI-Server starten. connect_async sorgt außerdem dafür, dass die App auch dann hochfährt, wenn der Broker kurzzeitig nicht erreichbar ist; paho versucht es weiter.
Ein Lock schützt den Cache. paho schreibt aus seinem Netzwerk-Thread, während Flask aus Request-Threads liest.
Die POST-Route liefert 202 Accepted, nicht 200. Der Broker hat die Nachricht bestätigt (QoS 1 PUBACK), aber das heißt nicht, dass das Gerät den Befehl ausgeführt hat. Brauchen Sie eine Bestätigung, lassen Sie das Gerät ein Ergebnis mit derselben request_id auf einem Antwort-Topic publizieren, oder nutzen Sie das Response-Topic-Muster von MQTT 5, beschrieben in MQTT 5 Features.
Geräte-IDs werden validiert, damit ein Aufrufer kein +, # oder zusätzliche Topic-Ebenen einschleusen kann.

Die MQTT zu REST API Flask-Bridge ausprobieren

Starten Sie die App:

export API_KEY=dev-secret
python app.py

Publizieren Sie eine Retained-Zustandsnachricht, wie es ein Gerät tun würde. Retained Messages sind hier nützlich: Startet die Bridge neu, liefert der Broker sofort den letzten Zustand jedes Geräts aus, sodass der Cache gefüllt ist, ohne auf die nächste Meldung zu warten.

mosquitto_pub -h broker.connectmq.com -p 1883 -i flask-dev-3391 -r -q 1 \
  -t connectmq/demo/flask-3391/valve7/state -m '{"open": true, "flow": 12.4}'

Abfragen und Befehle über HTTP:

curl -s -H "X-API-Key: dev-secret" http://localhost:5000/api/devices
curl -s -H "X-API-Key: dev-secret" http://localhost:5000/api/devices/valve7
curl -s -X POST -H "X-API-Key: dev-secret" -H "Content-Type: application/json" \
  -d '{"action": "close"}' http://localhost:5000/api/devices/valve7/command

Beobachten Sie das Eintreffen des Befehls mit mosquitto_sub -h broker.connectmq.com -p 1883 -i flask-sub-3391 -v -t "connectmq/demo/flask-3391/+/cmd". Da der öffentliche Bereich für alle lesbar ist, löschen Sie die Retained-Testnachricht am Ende, indem Sie einen leeren Retained-Payload auf dasselbe Topic publizieren.

Fallstricke beim Deployment

Die meisten fehlerhaften Setups aus Flask und MQTT scheitern an einem dieser Gründe:

Der Debug-Reloader führt Ihren Code zweimal aus. Flasks Reloader startet einen Kindprozess, sodass sich zwei MQTT-Clients mit derselben Client ID verbinden und sich gegenseitig in einer Schleife trennen. Behalten Sie use_reloader=False bei.
Mehrere WSGI-Worker. gunicorn -w 4 importiert die App viermal: vier MQTT-Clients und vier getrennte Caches, sodass aufeinanderfolgende Requests unterschiedliche Werte liefern können. Betreiben Sie stattdessen einen Worker mit Threads: gunicorn -w 1 --threads 8 app:app.
Feste Client IDs über Instanzen hinweg. Skalieren Sie horizontal hinter einem Load Balancer, braucht jede Instanz eine eindeutige Client ID. Auf einem Broker, bei dem Client IDs an Zugangsdaten gebunden sind, vergeben Sie pro Instanz eigene Zugangsdaten.
Den Cache als Wahrheit behandeln. Ein Gerät, das offline gegangen ist, hat weiterhin einen gecachten Wert. Stellen Sie wie im Code age_seconds und stale bereit, damit Konsumenten selbst entscheiden können.
Keine Authentifizierung. Alles, was POST-Requests senden kann, kann Anlagen bedienen. Der API-Key-Header ist das Minimum; betreiben Sie die App hinter HTTPS und einer richtigen Auth-Schicht, sobald sie aus dem Internet erreichbar ist.

Die Bridge erweitern

Sobald die Basis-Endpunkte funktionieren, decken einige Ergänzungen die meisten Praxisanforderungen ab:

Server-Sent Events für Live-Updates. GET /api/devices jede Sekunde abzufragen funktioniert, aber ein Endpunkt GET /api/stream, der bei jeder Cache-Änderung ein Event liefert, ermöglicht HTTP-Clients Push-Updates, ohne MQTT zu sprechen. Flask unterstützt Streaming-Responses über einen Generator; geben Sie jedem verbundenen Client eine eigene queue.Queue und legen Sie Updates aus on_message in alle Queues.
Filterung und Projektion. Query-Parameter wie ?stale=false oder ?fields=temperature halten die Antworten für mobile Clients klein.
Historie. Schreiben Sie jede Nachricht in on_message nach SQLite oder in eine Zeitreihendatenbank und ergänzen Sie GET /api/devices/<id>/history?since=.... Behalten Sie den In-Memory-Cache für den letzten Wert, denn er ist weit schneller, als bei jedem Request die Datenbank abzufragen.
OpenAPI-Dokumentation. Nutzen andere Teams die API, beschreiben Sie die Endpunkte in einer OpenAPI-Datei, damit sie Clients generieren können und genau wissen, welche Felder sie erwarten.

Jede Ergänzung bleibt im selben Muster: ein MQTT-Client pro Prozess, ein Lock um den gemeinsamen Zustand und HTTP-Handler, die nie auf den Broker warten.

Umstieg auf einen privaten Broker

Für die Produktion verbinden Sie die Bridge per TLS auf Port 8883 mit einem privaten Broker. Erstellen Sie im ConnectMQ-Portal einen Broker und kopieren Sie den generierten Benutzernamen, das Passwort und die Client ID in MQTT_USERNAME, MQTT_PASSWORD und MQTT_CLIENT_ID. Ein Cloud-MQTT-Broker wie dieser erspart es Ihnen, Mosquitto neben Flask zu betreiben, und der Live-Topic-Explorer im Portal hilft zu prüfen, was im Cache stehen sollte.

Im ConnectMQ-Plan Serverless (Free) reicht QoS bis 1 – genau das, was die Bridge für Zustände und Befehle verwendet. Der Plan Serverless (Paid) ergänzt QoS 2 und Shared Subscriptions; Letztere sind praktisch, wenn Sie später mehrere Bridge-Instanzen betreiben, die sich den eingehenden Traffic teilen sollen.

Starten Sie mit ConnectMQ

Bringen Sie Ihre Flask-Bridge in wenigen Minuten auf einen privaten, TLS-gesicherten Broker. Registrieren Sie sich unter app.connectmq.com, wählen Sie den kostenlosen Serverless-Plan ohne Kreditkarte, erstellen Sie einen Broker und fügen Sie die Zugangsdaten in die Umgebung der Bridge ein. Nutzen Sie den MQTT-Client im Browser, um Gerätezustände zu publizieren, während Sie die Endpunkte testen.

Häufig gestellte Fragen

Warum nicht Flask-MQTT verwenden?

Das können Sie, aber es ist ein dünner Wrapper, der dieselben Probleme mit Reloader und mehreren Workern hat, und Sie müssen prüfen, welche paho-mqtt-Versionen er unterstützt. Wenn Sie paho-mqtt 2.x direkt verwenden, behalten Sie die Kontrolle über Threading und den Lebenszyklus des Clients.

Kann die REST API historische Daten liefern?

Nicht mit einem In-Memory-Cache. Speichern Sie die Nachrichten aus demselben Subscriber in einer Datenbank wie SQLite, PostgreSQL oder einem Zeitreihenspeicher und ergänzen Sie Abfrage-Endpunkte, die daraus lesen.

Woher weiß ich, dass ein Befehl ausgeführt wurde?

Die 202-Antwort bestätigt nur, dass der Broker die Nachricht angenommen hat. Lassen Sie das Gerät eine Bestätigung mit der request_id publizieren und stellen Sie diese über einen weiteren Endpunkt oder ein Statusfeld im Gerätezustand bereit.

Kann ich FastAPI statt Flask verwenden?

Ja. Die Struktur ist dieselbe: ein MQTT-Client pro Prozess, ein mit Lock geschützter Cache und Routen, die ihn lesen. Mit FastAPI bevorzugen Sie eventuell einen asynchronen Client wie aiomqtt.

MQTT zu REST API mit Flask: Live-Gerätedaten über HTTP bereitstellen