MQTT 5 in Python mit gmqtt

gmqtt ist ein asyncio-MQTT-Client für Python, der von Anfang an rund um MQTT 5 entworfen wurde. Jede MQTT-5-Property — Session Expiry, Message Expiry, User Properties, Response Topic, Correlation Data, Subscription Identifiers — ist ein einfaches Keyword-Argument, und eingehende Properties kommen als Dictionary an. Wenn du MQTT-5-Features in Python ausprobieren willst, ohne Properties-Objekte von Hand zu bauen, ist gmqtt der kürzeste Weg.

Dieses Tutorial zeigt die Verbindung mit MQTT 5, Publishen mit Properties, Subscription-Optionen, Request/Response, das Auslesen von Reason Codes und das Testen gegen einen echten Broker.

Installieren und verbinden

pip install gmqtt

gmqtt arbeitet mit Callbacks, die auf dem asyncio-Event-Loop laufen. Verbinde dich, abonniere in on_connect und halte das Programm mit einem asyncio.Event am Leben:

import asyncio
import os
from gmqtt import Client as MQTTClient
from gmqtt.mqtt.constants import MQTTv50

STOP = asyncio.Event()

def on_connect(client, flags, rc, properties):
    print("verbunden, rc =", rc, "Broker-Properties:", properties)
    client.subscribe("testmqtt/gmqtt/#", qos=1)

def on_message(client, topic, payload, qos, properties):
    print("Nachricht", topic, payload.decode(), properties)
    return 0  # Reason Code für das PUBACK

def on_disconnect(client, packet, exc=None):
    print("getrennt", exc)

async def main():
    client = MQTTClient(
        "testmqtt-gmqtt-demo-01",
        session_expiry_interval=3600,   # Session 1 h nach dem Trennen behalten
        receive_maximum=100,            # höchstens 100 unbestätigte QoS-1/2-Nachrichten
    )
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_disconnect = on_disconnect
    client.set_auth_credentials(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])

    await client.connect("broker.connectmq.com", 1883, keepalive=30, version=MQTTv50)
    await STOP.wait()
    await client.disconnect()

asyncio.run(main())

Zwei Dinge sind typisch für gmqtt. Keyword-Argumente im MQTTClient-Konstruktor werden zu CONNECT-Properties — hier session_expiry_interval und receive_maximum. Und das Argument properties von on_connect enthält, was der Broker im CONNACK zurückgeschickt hat: je nach Broker sein eigenes Receive Maximum, Topic Alias Maximum, Maximum Packet Size oder einen zugewiesenen Client Identifier. Einmal ausgeben ist der schnellste Weg zu sehen, was dein Broker tatsächlich unterstützt.

Für TLS übergibst du ssl=True (oder einen ssl.SSLContext) und Port 8883: await client.connect("broker.connectmq.com", 8883, ssl=True).

Publishen mit MQTT-5-Properties

Publish-Properties sind Keyword-Argumente von client.publish():

import json, time

client.publish(
    "testmqtt/gmqtt/sensor-1/temperature",
    json.dumps({"value": 21.7, "t": time.time()}),
    qos=1,
    content_type="application/json",
    message_expiry_interval=60,                     # verwerfen, wenn nach 60 s nicht zugestellt
    user_property=("firmware", "2.4.1"),
)

Beim Subscriber tauchen dieselben Werte in properties auf. Achte auf die Form — die Werte sind Listen, weil MQTT 5 manche Properties mehrfach erlaubt:

message testmqtt/gmqtt/sensor-1/temperature {"value": 21.7, "t": 1792223400.12}
  {'content_type': ['application/json'], 'message_expiry_interval': [60],
   'user_property': [('firmware', '2.4.1')], 'dup': 0, 'retain': 0}

Lies sie defensiv, z. B. ctype = properties.get("content_type", [None])[0]. Lag die Nachricht vor der Zustellung in einer Queue, sendet der Broker die verbleibende Ablaufzeit — der Subscriber sieht dann einen Wert unter 60. So prüfst du leicht, ob Message Expiry funktioniert. Wann User Properties das richtige Werkzeug sind und wann nicht, erklärt MQTT User Properties.

Subscription-Optionen

MQTT 5 bringt Optionen pro Subscription. gmqtt stellt sie über Subscription-Objekte bereit:

from gmqtt import Subscription

client.subscribe(
    [
        Subscription("testmqtt/gmqtt/+/temperature", qos=1),
        Subscription("testmqtt/gmqtt/chat/#", qos=1, no_local=True),
        Subscription("testmqtt/gmqtt/config/#", qos=1, retain_handling_options=2),
    ],
    subscription_identifier=7,
)

Passt eine Nachricht auf die obige Subscription, enthält properties["subscription_identifier"] den Wert [7]. Das macht das Routing in on_message günstig und explizit, ohne Topic-Strings erneut abzugleichen.

Request/Response mit Response Topic und Correlation Data

MQTT 5 standardisiert Request/Response: Der Anfragende setzt ein Response Topic und Correlation Data, der Antwortende publisht seine Antwort dorthin und gibt die Correlation Data zurück.

def on_connect(client, flags, rc, properties):
    client.subscribe("testmqtt/gmqtt/rpc/time", qos=1)

def on_message(client, topic, payload, qos, properties):
    reply_to = properties.get("response_topic", [None])[0]
    corr = properties.get("correlation_data", [None])[0]
    if reply_to:
        client.publish(reply_to, str(time.time()), qos=1, correlation_data=corr)
    return 0

import uuid

corr = uuid.uuid4().bytes
client.subscribe("testmqtt/gmqtt/rpc/replies/client-01", qos=1)
client.publish(
    "testmqtt/gmqtt/rpc/time", "",
    qos=1,
    response_topic="testmqtt/gmqtt/rpc/replies/client-01",
    correlation_data=corr,
)

Vergleiche im on_message des Anfragenden properties["correlation_data"][0] mit corr, um die Antwort ihrer Anfrage zuzuordnen. Das vollständige Muster inklusive Timeouts findest du unter MQTT Request/Response.

Reason Codes und sauberes Trennen

MQTT 5 ersetzt die alten Ein-Byte-Return-Codes durch detaillierte Reason Codes in CONNACK, SUBACK, PUBACK und DISCONNECT. In gmqtt begegnen sie dir an mehreren Stellen:

• rc in on_connect — 0 heißt Erfolg; Werte wie 134 (Bad User Name or Password) oder 135 (Not authorized) erklären eine abgelehnte Verbindung.
• Das qos-Tupel in on_subscribe(client, mid, qos, properties) — ein Wert ab 128 für ein Topic bedeutet, dass der Broker diese Subscription abgelehnt hat, meist wegen einer ACL.
• Der Rückgabewert von on_message wird als Reason Code im zurückgesendeten PUBACK/PUBREC verwendet; 0 bedeutet Erfolg.

Jede Zahl kannst du mit dem MQTT-Reason-Code-Lookup nachschlagen.

Um dich so zu trennen, dass der Last Will trotzdem gesendet wird (praktisch zum Testen des Monitorings), nutze Reason Code 4: await client.disconnect(reason_code=4). Der Will selbst wird mit einem Message-Objekt konfiguriert, optional mit Will Delay Interval:

from gmqtt import Client as MQTTClient, Message

will = Message("testmqtt/gmqtt/sensor-1/status", "offline", qos=1, retain=True, will_delay_interval=10)
client = MQTTClient("testmqtt-gmqtt-sensor-1", will_message=will)

Reconnects und Session Expiry

Anders als aiomqtt verbindet sich gmqtt automatisch neu. Wie oft und wie lange es wartet, stellst du ein:

client.set_config({"reconnect_retries": 10, "reconnect_delay": 5})

Kombiniere Reconnects mit session_expiry_interval und einer festen Client-ID: Der Broker behält deine Subscriptions und puffert QoS-1-Nachrichten, während du weg bist, und die CONNACK-Flags zeigen, ob eine Session fortgesetzt wurde. Bei einer neuen Verbindung zeigt flags in on_connect „Session Present“ an; ist der Wert nach einem Ausfall 0, ist die Session abgelaufen und du musst neu abonnieren (das Abonnieren in on_connect wie oben erledigt das). Zum praktischen Prüfen siehe Session Expiry testen.

Deinen gmqtt-Client testen

1. Starte den Subscriber aus dem ersten Beispiel.
2. Öffne den Online-MQTT-Client, verbinde dich mit MQTT 5 zu broker.connectmq.com und publishe auf testmqtt/gmqtt/sensor-1/temperature mit einer User Property und einem Message Expiry.
3. Prüfe, dass das Properties-Dictionary im Terminal genau das enthält, was du gesendet hast.
4. Stoppe den Subscriber, publishe eine QoS-1-Nachricht und starte ihn innerhalb des Session-Expiry-Fensters neu — die Nachricht sollte beim Reconnect ankommen.

MQTT-5-Nachrichten an deinen gmqtt-Client senden

gmqtt im Vergleich zu anderen Python-Clients

gmqtt passt gut, wenn MQTT-5-Metadaten im Zentrum deines Designs stehen — Request/Response, Content Types, Ablaufzeiten — und dein Code ohnehin asynchron ist. Da es von einem kleineren Team als paho-mqtt gepflegt wird, prüfe die jüngsten Releases und pinne die Version, mit der du testest. Den größeren Zusammenhang erklärt MQTT 5 vs 3.1.1.

gmqtt lässt MQTT 5 wie normales Python wirken: Properties sind Argumente, Ergebnisse sind Dictionaries, und Reconnects passieren von selbst. Gib die CONNACK-Properties einmal aus, lies Property-Werte als Listen und prüfe jedes Feature mit einem zweiten Client, bevor du dich in Produktion darauf verlässt.

Häufig gestellte Fragen