Asynchrones MQTT in Python mit aiomqtt

aiomqtt ist der idiomatische Weg, MQTT aus Python-Code mit asyncio zu nutzen. Die Bibliothek verpackt paho-mqtt in einen asynchronen Context Manager: async with aiomqtt.Client(...) verbindet, await client.publish(...) sendet und async for message in client.messages empfängt — ohne Callbacks und ohne Hintergrund-Thread, den du mit deinem Event Loop koordinieren musst.

Dieses Tutorial zeigt Installation, Abonnieren und Publishen, eine Reconnect-Schleife, die Broker-Neustarts übersteht, die parallele Verarbeitung langsamer Nachrichten, TLS und die Einbindung des Clients in eine FastAPI-Anwendung.

Installation und erste Verbindung

aiomqtt benötigt Python 3.8 oder neuer und bringt paho-mqtt als einzige Abhängigkeit mit:

pip install aiomqtt

Das kleinste lauffähige Programm publisht eine Nachricht und beendet sich:

import asyncio
import aiomqtt

async def main():
    async with aiomqtt.Client("broker.connectmq.com", port=1883) as client:
        await client.publish("testmqtt/aiomqtt/hello", payload="Hallo aus asyncio", qos=1)

asyncio.run(main())

Beim Verlassen des async with-Blocks wird ein sauberes DISCONNECT gesendet. Verlangt der Broker Zugangsdaten, übergib username und password; setze mit identifier eine stabile Client-ID (erzeuge eine eindeutige mit dem Client-ID-Generator, damit es keine Kollisionen mit anderen Test-Clients gibt):

import os

client = aiomqtt.Client(
    "broker.connectmq.com",
    port=1883,
    identifier="testmqtt-aiomqtt-demo-01",
    username=os.environ["MQTT_USER"],
    password=os.environ["MQTT_PASS"],
    keepalive=30,
    protocol=aiomqtt.ProtocolVersion.V5,  # Standard ist MQTT 3.1.1
)

Abonnieren mit async for

Seit aiomqtt 2.0 kommen alle eingehenden Nachrichten über einen einzigen asynchronen Iterator, client.messages. Erst abonnieren, dann iterieren:

import asyncio
import aiomqtt

async def main():
    async with aiomqtt.Client("broker.connectmq.com") as client:
        await client.subscribe("testmqtt/aiomqtt/#", qos=1)
        async for message in client.messages:
            if message.topic.matches("testmqtt/aiomqtt/+/temperature"):
                print("temp", message.topic.value, message.payload.decode())
            elif message.topic.matches("testmqtt/aiomqtt/+/status"):
                print("status", message.topic.value, message.payload.decode())

asyncio.run(main())

message.topic ist ein Topic-Objekt: .value ist der String, .matches() wendet die MQTT-Wildcard-Regeln an — so kannst du Nachrichten aus mehreren Subscriptions in einer Schleife routen. Wenn du unsicher bist, ob ein Filter passt, prüfe ihn mit dem Topic Matcher. message.payload ist meist bytes; dekodiere es oder übergib es an json.loads. Außerdem gibt es message.qos, message.retain und bei MQTT 5 message.properties.

Zum Testen des Subscribers öffne den Online-MQTT-Client, verbinde dich mit broker.connectmq.com und publishe auf testmqtt/aiomqtt/kitchen/temperature.

Gleichzeitig publishen und empfangen

Ein Client kann publishen, während eine andere Coroutine client.messages konsumiert. Lass beide in einer TaskGroup laufen (Python 3.11+):

import asyncio
import json
import time
import aiomqtt

TOPIC = "testmqtt/aiomqtt/sensor-1/temperature"

async def produce(client: aiomqtt.Client):
    while True:
        payload = json.dumps({"t": round(time.time()), "value": 21.5})
        await client.publish(TOPIC, payload=payload, qos=1)
        await asyncio.sleep(2)

async def consume(client: aiomqtt.Client):
    await client.subscribe("testmqtt/aiomqtt/#")
    async for message in client.messages:
        print(message.topic.value, json.loads(message.payload))

async def main():
    async with aiomqtt.Client("broker.connectmq.com") as client:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(produce(client))
            tg.create_task(consume(client))

asyncio.run(main())

await client.publish() kehrt zurück, sobald die Nachricht mit dem gewünschten QoS übergeben wurde (bei QoS 1 nach dem PUBACK). Ruf es nicht aus einem anderen Thread auf — alles bleibt auf dem Event Loop.

Ein Reconnect-Muster, das wirklich funktioniert

aiomqtt verbindet sich nicht von selbst neu. Fällt das Netz aus oder startet der Broker neu, wirft der async with-Block aiomqtt.MqttError, typischerweise mit einer Meldung wie Disconnected during message iteration oder [code:7] The connection was lost. Das Client-Objekt ist wiederverwendbar, daher ist das Standardmuster eine Schleife um den Context Manager:

import asyncio
import random
import aiomqtt

async def main():
    client = aiomqtt.Client("broker.connectmq.com", identifier="testmqtt-aiomqtt-worker")
    delay = 1
    while True:
        try:
            async with client:
                delay = 1                                  # verbunden: Backoff zurücksetzen
                await client.subscribe("testmqtt/aiomqtt/#", qos=1)
                async for message in client.messages:
                    print(message.topic.value, message.payload)
        except aiomqtt.MqttError as err:
            wait = delay + random.uniform(0, delay)        # Jitter
            print(f"Verbindung verloren ({err}); neuer Versuch in {wait:.1f}s")
            await asyncio.sleep(wait)
            delay = min(delay * 2, 60)                     # exponentielles Backoff

asyncio.run(main())

Drei Details sind entscheidend. Die Subscriptions passieren innerhalb des Blocks und werden so nach jedem Reconnect wiederhergestellt. Die Wartezeit wächst exponentiell mit Jitter, damit eine Flotte von Workern den Broker nach seiner Rückkehr nicht synchron bestürmt — siehe Best Practices für Reconnects. Und ein fester identifier lässt sich mit einer persistenten Session kombinieren (clean_session=False bei MQTT 3.1.1 bzw. clean_start plus Session Expiry Interval bei MQTT 5), sodass QoS-1-Nachrichten, die während deiner Offline-Zeit gepublisht wurden, trotzdem ankommen.

Zum Testen starte das Skript, starte deinen lokalen Broker neu (oder zieh das Netzwerkkabel) und beobachte das Log: Du solltest eine Zeile Verbindung verloren sehen und danach wieder eingehende Nachrichten.

Die Nachrichtenschleife nicht blockieren

Die async for-Schleife verarbeitet eine Nachricht nach der anderen. Wartet ein Handler auf einen langsamen Datenbank-Write oder HTTP-Call, stauen sich alle anderen Nachrichten dahinter. Gib langsame Arbeit an Tasks ab und begrenze die Parallelität mit einem Semaphor:

sem = asyncio.Semaphore(20)

async def handle(message: aiomqtt.Message):
    async with sem:
        await save_to_db(message)          # dein langsames I/O

async with asyncio.TaskGroup() as tg:
    async for message in client.messages:
        tg.create_task(handle(message))

CPU-lastige Arbeit (Bilder dekodieren, ML-Inferenz) blockiert den Loop trotzdem; verlagere sie mit asyncio.to_thread() oder in einen Prozess-Pool. Eingehende Nachrichten landen in einer internen Queue; max_queued_incoming_messages begrenzt ihre Größe, falls ein langsamer Consumer den Speicher sonst unbegrenzt wachsen ließe.

TLS und WebSockets

Für Produktionsverkehr nutzt du TLS auf Port 8883. Übergib einen normalen ssl.SSLContext:

import ssl

client = aiomqtt.Client(
    "broker.connectmq.com",
    port=8883,
    tls_context=ssl.create_default_context(),
    username=os.environ["MQTT_USER"],
    password=os.environ["MQTT_PASS"],
)

In Netzen, die nur HTTPS erlauben, wechselst du auf sichere WebSockets: port=8084, transport="websockets", websocket_path="/mqtt" mit demselben tls_context. Client-Zertifikate und eigene CAs behandelt MQTT-TLS-Zertifikate in Python.

aiomqtt in FastAPI verwenden

FastAPI läuft bereits auf einem Event Loop, aiomqtt passt also ganz natürlich. Öffne die Verbindung im lifespan-Handler, starte den Listener als Hintergrund-Task und stell den Client den Request-Handlern über app.state bereit:

import asyncio
from contextlib import asynccontextmanager
import aiomqtt
from fastapi import FastAPI

latest: dict[str, str] = {}

async def listen(client: aiomqtt.Client):
    await client.subscribe("testmqtt/aiomqtt/+/temperature")
    async for message in client.messages:
        latest[message.topic.value] = message.payload.decode()

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with aiomqtt.Client("broker.connectmq.com") as client:
        app.state.mqtt = client
        task = asyncio.create_task(listen(client))
        yield
        task.cancel()

app = FastAPI(lifespan=lifespan)

@app.post("/publish/{device}")
async def publish(device: str, value: float):
    await app.state.mqtt.publish(f"testmqtt/aiomqtt/{device}/temperature", payload=str(value), qos=1)
    return {"ok": True}

@app.get("/latest")
async def get_latest():
    return latest

Starte es mit uvicorn app:app — mit einem einzigen Worker. Jeder Worker-Prozess öffnet seine eigene MQTT-Verbindung; mehrere Worker mit demselben identifier würden sich gegenseitig vom Broker werfen. Eine vollständige Version mit WebSocket-Push in den Browser und Reconnect-Behandlung findest du unter MQTT mit FastAPI.

aiomqtt oder reines paho?

Einen breiteren Überblick bietet der Vergleich der Python-MQTT-Client-Bibliotheken.

Deine aiomqtt-Nachrichten live ansehen

aiomqtt hält MQTT-Code lesbar: ein Context Manager für die Verbindung, eine Schleife für die Nachrichten und ein schlichtes try/except für Fehler. Bau die Reconnect-Schleife vom ersten Tag an ein, halte langsame Arbeit aus der Nachrichtenschleife heraus und prüfe alles mit einem zweiten Client, bevor du deployst.

Häufig gestellte Fragen