MQTT mit FastAPI: Brücke zwischen HTTP und MQTT

Eine FastAPI-MQTT-Brücke lässt normale HTTP-Clients mit Geräten sprechen: Ein POST wird zu einem MQTT-Publish, und Nachrichten der Geräte werden per WebSocket an Browser gestreamt. Da FastAPI auf asyncio aufbaut, ist ein asyncio-nativer Client (aiomqtt) der sauberste Weg — verwaltet von einem einzigen Hintergrund-Task, der im lifespan der App startet.

Dieses Tutorial baut einen kleinen, aber produktionsnahen Dienst: einen REST-Endpoint, der Kommandos sendet, einen Subscriber, der Broker-Neustarts übersteht, einen WebSocket, der Live-Telemetrie an Browser pusht, und die Regeln für den Betrieb mit mehreren Workern.

Architektur

Eine MQTT-Verbindung pro Prozess, beim Start geöffnet und von allen Requests geteilt. HTTP-Handler erzeugen nie eigene Clients — eine MQTT-Session pro Request ist langsam, überflutet den Broker mit Connects und bricht, sobald zwei Requests dieselbe Client-ID verwenden.

pip install "fastapi[standard]" aiomqtt

Die MQTT-Brücke mit Reconnects

Der Brücke gehört der Client, sie weiß, ob sie verbunden ist, und verteilt eingehende Nachrichten über kleine Queues pro Verbindung an jeden offenen WebSocket:

import asyncio
import os
import uuid
import aiomqtt

BROKER = os.getenv("MQTT_HOST", "broker.connectmq.com")
TELEMETRY = "testmqtt/fastapi/+/telemetry"


class Hub:
    """Verteilt MQTT-Nachrichten an WebSocket-Clients."""

    def __init__(self):
        self.queues: set[asyncio.Queue] = set()

    def join(self) -> asyncio.Queue:
        q: asyncio.Queue = asyncio.Queue(maxsize=100)
        self.queues.add(q)
        return q

    def leave(self, q: asyncio.Queue):
        self.queues.discard(q)

    def broadcast(self, item: dict):
        for q in list(self.queues):
            try:
                q.put_nowait(item)
            except asyncio.QueueFull:
                pass  # langsamer Browser: verwerfen statt alle zu blockieren


class MqttBridge:
    def __init__(self, hub: Hub):
        self.hub = hub
        self.connected = asyncio.Event()
        self.client = aiomqtt.Client(
            BROKER,
            port=1883,
            identifier=f"testmqtt-fastapi-{uuid.uuid4().hex[:8]}",  # eindeutig pro Prozess
            username=os.getenv("MQTT_USER"),
            password=os.getenv("MQTT_PASS"),
        )

    async def run(self):
        delay = 1
        while True:
            try:
                async with self.client:
                    self.connected.set()
                    delay = 1
                    await self.client.subscribe(TELEMETRY, qos=1)
                    async for msg in self.client.messages:
                        self.hub.broadcast({
                            "topic": msg.topic.value,
                            "payload": msg.payload.decode(errors="replace"),
                        })
            except aiomqtt.MqttError as err:
                self.connected.clear()
                print(f"MQTT-Verbindung verloren: {err}; neuer Versuch in {delay}s")
                await asyncio.sleep(delay)
                delay = min(delay * 2, 30)

    async def publish(self, topic: str, payload: str, qos: int = 1):
        if not self.connected.is_set():
            raise ConnectionError("MQTT broker not connected")
        await self.client.publish(topic, payload, qos=qos)

Die run()-Schleife ist das Reconnect-Muster aus der aiomqtt-Anleitung: Abonnieren innerhalb von async with, exponentielles Backoff und ein ehrliches connected-Flag, damit HTTP-Handler schnell scheitern statt zu hängen.

Die FastAPI-App und der Lifespan

import asyncio
import contextlib
import json
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import FastAPI, HTTPException, Path, WebSocket, WebSocketDisconnect
from pydantic import BaseModel

from bridge import Hub, MqttBridge

hub = Hub()
bridge = MqttBridge(hub)


@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(bridge.run())
    with contextlib.suppress(TimeoutError):
        await asyncio.wait_for(bridge.connected.wait(), timeout=10)
    yield
    task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await task


app = FastAPI(lifespan=lifespan)
DeviceId = Annotated[str, Path(pattern=r"^[A-Za-z0-9_-]{1,64}$")]


class Command(BaseModel):
    action: str
    value: float | None = None


@app.post("/devices/{device_id}/commands", status_code=202)
async def send_command(device_id: DeviceId, cmd: Command):
    try:
        await bridge.publish(f"testmqtt/fastapi/{device_id}/commands", cmd.model_dump_json())
    except ConnectionError:
        raise HTTPException(503, "MQTT broker not connected")
    return {"queued": True, "topic": f"testmqtt/fastapi/{device_id}/commands"}


@app.get("/health")
async def health():
    return {"mqtt_connected": bridge.connected.is_set()}


@app.websocket("/ws")
async def live(websocket: WebSocket):
    await websocket.accept()
    queue = hub.join()
    try:
        while True:
            await websocket.send_text(json.dumps(await queue.get()))
    except WebSocketDisconnect:
        pass
    finally:
        hub.leave(queue)

Einige bewusste Entscheidungen:

• Der Pfad-Parameter wird validiert. device_id wird Teil eines Topics, daher lehnt eine Regex +, # und andere Zeichen ab, die ein ungültiges oder zu breites Topic ergeben würden. Setze niemals ungeprüfte Benutzereingaben in Topic-Namen ein.
• 202 Accepted statt 200. Ein erfolgreicher QoS-1-Publish bedeutet, dass der Broker die Nachricht hat — nicht, dass das Gerät das Kommando ausgeführt hat. Brauchst du eine Bestätigung, lass das Gerät auf einem Response Topic antworten.
• Der Start wartet bis zu 10 Sekunden auf die erste Verbindung, lässt die App aber nicht scheitern, wenn der Broker weg ist; /health meldet den Zustand, damit dein Orchestrator entscheiden kann.
• Begrenzte Queues pro WebSocket sorgen dafür, dass ein langsamer Browser die Zustellung an alle anderen nicht ausbremst.

Starte es mit fastapi dev main.py (oder uvicorn main:app).

Push in den Browser

Der Browser braucht nur die Standard-WebSocket-API — er spricht nie direkt MQTT, du gibst also keine Broker-Zugangsdaten ans Frontend:

<ul id="log"></ul>
<script>
  const ws = new WebSocket(`ws://${location.host}/ws`);
  ws.onmessage = (event) => {
    const { topic, payload } = JSON.parse(event.data);
    const li = document.createElement("li");
    li.textContent = `${topic}: ${payload}`;
    document.getElementById("log").prepend(li);
  };
</script>

Soll der Browser lieber selbst beim Broker abonnieren, nutze MQTT über WebSockets auf Port 8084 — siehe MQTT über WebSockets. Der Brücken-Ansatz ist besser, wenn die API Nachrichten filtern, autorisieren oder anreichern soll, bevor Nutzer sie sehen; eine vollständige UI auf dieser Basis zeigt das Live-Dashboard-Tutorial.

Die Brücke Ende-zu-Ende testen

1. Starte die App und prüfe, dass curl localhost:8000/health den Wert {"mqtt_connected": true} liefert.
2. Öffne den Online-MQTT-Client, verbinde dich mit broker.connectmq.com und abonniere testmqtt/fastapi/#.
3. Sende ein Kommando per HTTP und sieh zu, wie es im Online-Client erscheint:

curl -X POST localhost:8000/devices/sensor-1/commands \
  -H 'content-type: application/json' \
  -d '{"action": "set_interval", "value": 30}'

1. Verbinde dich mit dem WebSocket (z. B. mit websocat ws://localhost:8000/ws) und publishe {"temp": 22.4} aus dem Online-Client auf testmqtt/fastapi/sensor-1/telemetry. websocat sollte die Nachricht sofort ausgeben.
2. Starte deinen Broker neu oder trenn das Netzwerk: /health sollte auf false springen, der Kommando-Endpoint 503 liefern, und beides sollte sich von selbst erholen.

Den Traffic der FastAPI-Brücke beobachten

Für automatisierte Tests führt with TestClient(app) as client: den Lifespan aus, die Brücke verbindet sich also wie in Produktion; richte MQTT_HOST auf einen Broker in einem Container. Fertige Fixtures findest du unter MQTT-pytest-Fixtures.

Mehrere Worker betreiben

uvicorn --workers 4 oder Gunicorn starten mehrere Prozesse, und jeder führt den Lifespan aus — du bekommst also vier MQTT-Verbindungen. Das ist in Ordnung, wenn du es einplanst:

• Eindeutige Client-IDs pro Prozess. Die Brücke hängt ein zufälliges Suffix an. Mit einer festen ID würden sich die Worker in einer Schleife gegenseitig trennen; siehe doppelte Client-ID.
• Der WebSocket-Fan-out braucht jede Nachricht in jedem Worker, weil ein Browser an jedem davon hängen kann. Normale Subscriptions leisten genau das.
• Verarbeitung darf nur einmal passieren. Schreibt ein Worker zusätzlich Telemetrie in eine Datenbank, nutze für diesen Consumer eine Shared Subscription wie $share/api/testmqtt/fastapi/+/telemetry, damit der Broker die Nachrichten auf die Worker verteilt statt sie zu duplizieren. Siehe Shared Subscriptions.

Alternativen

• paho-mqtt mit loop_start() funktioniert, aber die Callbacks laufen in einem separaten Thread. Reiche jede Nachricht mit loop.call_soon_threadsafe(...) weiter und fass asyncio-Objekte nie direkt im Callback an.
• fastapi-mqtt ist eine Community-Erweiterung auf Basis von gmqtt, die Decorators für Subscriptions mitbringt. Sie spart Boilerplate; die explizite Brücke oben gibt dir mehr Kontrolle über Reconnects und Backpressure.

Eine FastAPI-MQTT-Brücke läuft auf eine langlebige Verbindung pro Prozess hinaus, verwaltet im lifespan, mit ehrlichem Health-Status und begrenzten Queues Richtung Browser. Validiere alles, was zu einem Topic wird, halte Client-IDs pro Worker eindeutig und teste den kompletten Kreislauf — HTTP rein, MQTT raus, MQTT rein, WebSocket raus —, bevor du Geräte dahinter hängst.

Häufig gestellte Fragen