MQTT mit Python und paho-mqtt: Tutorial zu Publish & Subscribe

paho-mqtt ist die Standard-MQTT-Client-Bibliothek für Python. Sie unterstützt MQTT 3.1, 3.1.1 und 5, reines TCP, TLS und WebSockets und läuft überall – vom Raspberry Pi bis zum Cloud-Server. Dieses Tutorial nutzt die aktuelle API von paho-mqtt 2.x. Ältere Blogposts für 1.x werfen Fehler, wenn du sie unverändert kopierst, deshalb weisen wir unterwegs auf die Unterschiede hin.

paho-mqtt installieren

python -m pip install "paho-mqtt>=2.0"

Prüfe deine Version mit python -c "import paho.mqtt; print(paho.mqtt.__version__)". Alle folgenden Beispiele zielen auf 2.x und die öffentliche HiveMQ-Sandbox unter broker.hivemq.com, du kannst sie also ohne eigenen Broker ausführen. Wenn du lieber deinen eigenen betreibst, sieh dir Mosquitto installieren an.

Was sich in paho-mqtt 2.x geändert hat

Die große Änderung ist die Callback-API-Version. Das erste Argument von mqtt.Client() ist jetzt eine CallbackAPIVersion, und mit VERSION2 bekommen alle Callbacks Reason Codes und Properties im Stil von MQTT 5 – selbst wenn du dich mit MQTT 3.1.1 verbindest.

reason_code ist ein ReasonCode-Objekt: Prüfe reason_code.is_failure, statt rc == 0 zu vergleichen. Unser Nachschlagewerk für MQTT-Reason-Codes erklärt, was jeder Wert bedeutet.

Ein vollständiger Subscriber

Der folgende Subscriber verbindet sich, subscribt innerhalb von on_connect und gibt jede Nachricht aus. Dass du in on_connect statt nach connect() subscribst, ist wichtig: Wenn die Verbindung abbricht und paho sich neu verbindet, läuft der Callback erneut und deine Subscriptions werden wiederhergestellt.

import paho.mqtt.client as mqtt

BROKER = "broker.hivemq.com"
TOPIC = "testmqtt/python-demo/#"


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"Connect failed: {reason_code}")
        return
    print("Connected, subscribing to", TOPIC)
    client.subscribe(TOPIC, qos=1)


def on_subscribe(client, userdata, mid, reason_code_list, properties):
    for rc in reason_code_list:
        if rc.is_failure:
            print(f"Subscription rejected: {rc}")
        else:
            print(f"Subscribed with granted QoS {rc.value}")


def on_message(client, userdata, msg):
    print(f"{msg.topic} [QoS {msg.qos}, retain={msg.retain}]: {msg.payload.decode()}")


def on_disconnect(client, userdata, disconnect_flags, reason_code, properties):
    print(f"Disconnected: {reason_code}")


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="")
client.on_connect = on_connect
client.on_subscribe = on_subscribe
client.on_message = on_message
client.on_disconnect = on_disconnect

client.connect(BROKER, 1883, keepalive=60)
client.loop_forever()

Mit einer leeren client_id generiert paho eine zufällige. Wenn du eine eigene setzt, muss sie eindeutig sein – zwei Clients mit derselben ID werfen sich gegenseitig vom Broker. Der Client-ID-Generator erzeugt dir sichere IDs.

Ein vollständiger Publisher

Ein Publisher hat meist noch anderes zu tun – einen Sensor auslesen, eine API abfragen –, deshalb lässt er den Netzwerk-Loop mit loop_start() in einem Hintergrund-Thread laufen:

import json
import random
import time

import paho.mqtt.client as mqtt

BROKER = "broker.hivemq.com"
TOPIC = "testmqtt/python-demo/sensor"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.connect(BROKER, 1883, keepalive=60)
client.loop_start()

try:
    while True:
        payload = json.dumps({
            "temperature": round(random.uniform(18, 26), 1),
            "ts": int(time.time()),
        })
        info = client.publish(TOPIC, payload, qos=1)
        info.wait_for_publish(timeout=5)
        print("Published", payload, "mid", info.mid)
        time.sleep(2)
except KeyboardInterrupt:
    pass
finally:
    client.disconnect()
    client.loop_stop()

Starte den Subscriber in einem Terminal und den Publisher in einem anderen. Die Nachrichten deines Skripts kannst du auch im Browser verfolgen:

Deine Python-Nachrichten live ansehen

loop_forever vs. loop_start vs. loop

• loop_forever() – blockiert den aktuellen Thread, kümmert sich um Reconnects und kehrt erst nach disconnect() zurück. Am besten für reine Subscriber und Services.
• loop_start() / loop_stop() – lässt den Loop in einem Hintergrund-Thread laufen. Am besten, wenn dein Haupt-Thread anderes zu tun hat. Callbacks laufen in diesem Hintergrund-Thread, schütze geteilten Zustand also entsprechend.
• loop(timeout) – verarbeitet den Netzwerk-Traffic einmal. Nur nötig, wenn du paho in deinen eigenen Event-Loop integrierst.

QoS, Retained Messages und Last Will

Übergib qos=0, 1 oder 2 sowohl an publish() als auch an subscribe(); die effektive QoS ist die niedrigere der beiden. Mit retain=True speichert der Broker die letzte Nachricht für neue Subscriber. Setze vor dem Verbinden einen Last Will, damit der Broker einen unerwarteten Verbindungsabbruch meldet:

client.will_set("testmqtt/python-demo/status", "offline", qos=1, retain=True)
client.connect(BROKER, 1883)
client.publish("testmqtt/python-demo/status", "online", qos=1, retain=True)

Mehr dazu unter MQTT-QoS-Stufen, Retained Messages und Last Will and Testament.

TLS und Authentifizierung mit Benutzername/Passwort

Jedes echte Deployment sollte TLS auf Port 8883 plus Zugangsdaten nutzen. Beides wird vor connect() konfiguriert:

import ssl
import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sensor-livingroom-01")
client.username_pw_set("my-username", "my-password")

# Uses the system CA store; fine for publicly trusted certificates.
client.tls_set()

# For a private CA or client-certificate auth instead:
# client.tls_set(ca_certs="ca.crt", certfile="client.crt", keyfile="client.key",
#                tls_version=ssl.PROTOCOL_TLS_CLIENT)

client.connect("broker.example.com", 8883, keepalive=60)
client.loop_forever()

Vermeide tls_insecure_set(True) außerhalb schneller lokaler Tests – es deaktiviert die Hostnamen-Prüfung. Für MQTT 5 erzeugst du den Client mit protocol=mqtt.MQTTv5. Für WebSockets übergibst du transport="websockets" und rufst client.ws_set_options(path="/mqtt") auf.

Zuverlässig neu verbinden

Sowohl loop_forever() als auch loop_start() verbinden sich nach einem Verbindungsverlust automatisch neu. Das Backoff stellst du mit reconnect_delay_set() ein:

client.reconnect_delay_set(min_delay=1, max_delay=60)

Das allererste connect() verhält sich anders: Es wirft eine Exception (zum Beispiel ConnectionRefusedError oder einen Socket-Timeout), wenn der Broker nicht erreichbar ist. Pack es in eine Retry-Schleife oder nutze connect_async() gefolgt von loop_start(), das es im Hintergrund erneut versucht. Wenn Verbindungen weiterhin scheitern, behandelt der Leitfaden zu MQTT-Verbindungsfehlern die üblichen Verdächtigen: falscher Port, blockierende Firewall, falsche Zugangsdaten oder doppelte Client-IDs.

MQTT-5-Features nutzen

Übergib protocol=mqtt.MQTTv5, um MQTT 5 freizuschalten: Session Expiry, User Properties, Message Expiry und detaillierte Reason Codes. Die Verbindungsoptionen wandern von clean_session (ein Konstruktor-Argument bei 3.1.1) zu clean_start in connect(), und Metadaten pro Nachricht kommen in ein Properties-Objekt:

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)

connect_props = Properties(PacketTypes.CONNECT)
connect_props.SessionExpiryInterval = 3600  # keep the session for 1 hour

client.connect("broker.hivemq.com", 1883, keepalive=60,
               clean_start=False, properties=connect_props)
client.loop_start()

publish_props = Properties(PacketTypes.PUBLISH)
publish_props.MessageExpiryInterval = 30          # drop if undelivered after 30 s
publish_props.ContentType = "application/json"
publish_props.UserProperty = [("source", "python-demo")]

info = client.publish("testmqtt/python-demo/v5", '{"ok": true}', qos=1,
                      properties=publish_props)
info.wait_for_publish(timeout=5)

client.disconnect()
client.loop_stop()

In on_message stehen die empfangenen Properties als msg.properties zur Verfügung. Nicht sicher, ob du Version 5 brauchst? Unser Vergleich MQTT 5 vs. 3.1.1 listet die Unterschiede auf, die in der Praxis zählen.

Häufige Fehler

• 1.x-Code kopieren. mqtt.Client("my-id") übergibt die Client-ID dort, wo 2.x die Callback-API-Version erwartet. Nutze das Keyword-Argument client_id=.
• Den Loop vergessen. Ohne loop_forever() oder loop_start() wird nichts gesendet oder empfangen, und Callbacks werden nie ausgelöst.
• Zu früh beenden. Ein Skript, das publish() aufruft und sofort endet, beendet sich womöglich, bevor die Nachricht den Socket verlassen hat. Ruf wait_for_publish() auf oder nutze publish.single().
• Blind dekodieren. msg.payload ist bytes. Nutze msg.payload.decode("utf-8") für Text und json.loads() für JSON und fange Dekodierfehler ab – andere Clients auf demselben Topic können alles Mögliche senden.
• # auf einem öffentlichen Broker abonnieren. Du empfängst den Traffic aller anderen Nutzer. Bleib innerhalb deines eigenen Topic-Präfixes; siehe MQTT-Topics und Wildcards.

Einmaliges Publishen

Für Cronjobs und Skripte, die nur eine Nachricht senden, erledigt der Helfer paho.mqtt.publish Verbinden, Publishen und Trennen in einem einzigen Aufruf:

import paho.mqtt.publish as publish

publish.single("testmqtt/python-demo/cron", "job finished", qos=1,
               hostname="broker.hivemq.com", port=1883)

Willst du diesen Code mit deinem eigenen Host, Port, Zugangsdaten und Topic vorausgefüllt? Erzeuge ihn mit dem MQTT-Code-Generator passend zu deinen Broker-Einstellungen.

Häufig gestellte Fragen