MQTT en Python avec paho-mqtt : tutoriel publish et subscribe

paho-mqtt est la bibliothèque cliente MQTT de référence pour Python. Elle prend en charge MQTT 3.1, 3.1.1 et 5, TCP brut, TLS et WebSockets, et tourne partout, d’un Raspberry Pi à un serveur cloud. Ce tutoriel utilise l’API actuelle de paho-mqtt 2.x — les anciens articles écrits pour la 1.x provoquent des erreurs si vous les copiez tels quels, c’est pourquoi les différences sont signalées au fil du texte.

Installer paho-mqtt

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

Vérifiez votre version avec python -c "import paho.mqtt; print(paho.mqtt.__version__)". Tous les exemples ci-dessous ciblent la 2.x et le bac à sable public HiveMQ sur broker.hivemq.com : vous pouvez donc les exécuter sans installer de broker. Si vous préférez utiliser le vôtre, voyez comment installer Mosquitto.

Ce qui a changé dans paho-mqtt 2.x

Le grand changement, c’est la version de l’API de callbacks. Le premier argument de mqtt.Client() est désormais un CallbackAPIVersion, et avec VERSION2 tous les callbacks reçoivent des codes de raison et des propriétés à la manière de MQTT 5 — même si vous vous connectez en MQTT 3.1.1.

reason_code est un objet ReasonCode : testez reason_code.is_failure au lieu de comparer rc == 0. Notre outil de recherche des codes de raison MQTT explique la signification de chaque valeur.

Un abonné complet

L’abonné ci-dessous se connecte, s’abonne dans on_connect et affiche chaque message. S’abonner dans on_connect plutôt qu’après connect() est important : quand la connexion tombe et que paho se reconnecte, le callback s’exécute de nouveau et vos abonnements sont rétablis.

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()

Un client_id vide laisse paho en générer un aléatoire. Si vous définissez le vôtre, faites en sorte qu’il soit unique — deux clients avec le même ID s’éjectent mutuellement du broker en boucle. Le générateur de client ID peut créer des ID sûrs pour vous.

Un éditeur complet

Un éditeur a généralement d’autres tâches à accomplir — lire un capteur, interroger une API — il exécute donc la boucle réseau dans un thread d’arrière-plan avec loop_start() :

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()

Lancez l’abonné dans un terminal et l’éditeur dans un autre. Vous pouvez aussi observer depuis le navigateur les messages publiés par votre script :

Voir vos messages Python en direct

loop_forever vs loop_start vs loop

• loop_forever() — bloque le thread courant, gère les reconnexions et ne rend la main qu’après disconnect(). Idéal pour les abonnés purs et les services.
• loop_start() / loop_stop() — exécute la boucle dans un thread d’arrière-plan. Idéal quand votre thread principal fait autre chose. Les callbacks s’exécutent sur ce thread d’arrière-plan : protégez l’état partagé en conséquence.
• loop(timeout) — traite le trafic réseau une seule fois. Utile uniquement si vous intégrez paho à votre propre boucle d’événements.

QoS, messages retenus et Last Will

Passez qos=0, 1 ou 2 à publish() comme à subscribe() ; la QoS effective est la plus basse des deux. retain=True demande au broker de conserver le dernier message pour les nouveaux abonnés. Définissez un Last Will avant de vous connecter pour que le broker annonce une déconnexion inattendue :

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)

Pour aller plus loin : les niveaux de QoS MQTT, les messages retenus et Last Will and Testament.

TLS et authentification par nom d’utilisateur/mot de passe

Tout déploiement réel doit utiliser TLS sur le port 8883 ainsi que des identifiants. Les deux se configurent avant connect() :

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()

Évitez tls_insecure_set(True) en dehors de tests locaux rapides — cela désactive la vérification du nom d’hôte. Pour MQTT 5, créez le client avec protocol=mqtt.MQTTv5. Pour les WebSockets, passez transport="websockets" et appelez client.ws_set_options(path="/mqtt").

Se reconnecter de façon fiable

loop_forever() comme loop_start() se reconnectent automatiquement après une perte de connexion. Ajustez le délai de backoff avec reconnect_delay_set() :

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

Le tout premier connect() est différent : il lève une exception (par exemple ConnectionRefusedError ou un timeout de socket) si le broker est injoignable. Enveloppez-le dans une boucle de nouvelles tentatives, ou utilisez connect_async() suivi de loop_start(), qui réessaie en arrière-plan. Si les connexions échouent sans cesse, le guide des erreurs de connexion MQTT passe en revue les suspects habituels : mauvais port, pare-feu bloquant, identifiants erronés ou client ID en double.

Utiliser les fonctionnalités de MQTT 5

Passez protocol=mqtt.MQTTv5 pour débloquer MQTT 5 : expiration de session, user properties, expiration des messages et codes de raison détaillés. Les options de connexion passent de clean_session (argument du constructeur en 3.1.1) à clean_start sur connect(), et les métadonnées de chaque message vont dans un objet Properties :

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()

Dans on_message, les propriétés reçues sont disponibles via msg.properties. Vous ne savez pas si vous avez besoin de la version 5 ? Notre comparatif MQTT 5 vs 3.1.1 liste les différences qui comptent en pratique.

Erreurs courantes

• Copier du code 1.x. mqtt.Client("my-id") passe le client ID là où la 2.x attend la version de l’API de callbacks. Utilisez l’argument nommé client_id=.
• Oublier la boucle. Sans loop_forever() ni loop_start(), rien n’est envoyé ni reçu et les callbacks ne se déclenchent jamais.
• Quitter trop tôt. Un script qui appelle publish() puis se termine aussitôt peut s’arrêter avant que le message ait quitté le socket. Appelez wait_for_publish() ou utilisez publish.single().
• Décoder à l’aveugle. msg.payload est de type bytes. Utilisez msg.payload.decode("utf-8") pour du texte et json.loads() pour du JSON, et interceptez les erreurs de décodage — d’autres clients sur le même topic peuvent envoyer n’importe quoi.
• S’abonner à # sur un broker public. Vous recevrez le trafic de tous les autres utilisateurs. Restez dans votre propre préfixe de topic ; voir topics et jokers MQTT.

Publication ponctuelle

Pour les tâches cron et les scripts qui n’envoient qu’un message, l’utilitaire paho.mqtt.publish gère connexion, publication et déconnexion en un seul appel :

import paho.mqtt.publish as publish

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

Vous voulez ce code prérempli avec votre hôte, votre port, vos identifiants et votre topic ? Utilisez le générateur de code MQTT pour le produire selon les paramètres de votre broker.

Questions fréquentes