MQTT 5 en Python avec gmqtt

gmqtt est un client MQTT asyncio pour Python conçu autour de MQTT 5 dès le départ. Chaque propriété MQTT 5 — session expiry, message expiry, user properties, response topic, correlation data, subscription identifiers — est un simple argument nommé, et les propriétés entrantes arrivent sous forme de dictionnaire. Si vous voulez expérimenter les fonctionnalités MQTT 5 en Python sans construire des objets Properties à la main, gmqtt est le chemin le plus court.

Ce tutoriel couvre la connexion en MQTT 5, la publication avec propriétés, les options d'abonnement, le request/response, la lecture des reason codes et le test de l'ensemble contre un vrai broker.

Installer et se connecter

pip install gmqtt

gmqtt utilise des callbacks exécutés sur l'event loop asyncio. Connectez-vous, abonnez-vous dans on_connect et maintenez le programme en vie avec un asyncio.Event :

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("connecté, rc =", rc, "propriétés du broker :", properties)
    client.subscribe("testmqtt/gmqtt/#", qos=1)

def on_message(client, topic, payload, qos, properties):
    print("message", topic, payload.decode(), properties)
    return 0  # reason code du PUBACK

def on_disconnect(client, packet, exc=None):
    print("déconnecté", exc)

async def main():
    client = MQTTClient(
        "testmqtt-gmqtt-demo-01",
        session_expiry_interval=3600,   # garder la session 1 h après la déconnexion
        receive_maximum=100,            # au plus 100 messages QoS 1/2 non acquittés
    )
    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())

Deux points sont propres à gmqtt. Les arguments nommés passés au constructeur MQTTClient deviennent des propriétés du CONNECT — ici session_expiry_interval et receive_maximum. Et l'argument properties de on_connect contient ce que le broker a renvoyé dans le CONNACK : son propre receive maximum, topic alias maximum, maximum packet size ou un client identifier attribué, selon le broker. L'afficher une fois est le moyen le plus rapide de voir ce que votre broker prend réellement en charge.

Pour TLS, passez ssl=True (ou un ssl.SSLContext) et le port 8883 : await client.connect("broker.connectmq.com", 8883, ssl=True).

Publier avec des propriétés MQTT 5

Les propriétés de publication sont des arguments nommés de 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,                     # abandonner si non livré après 60 s
    user_property=("firmware", "2.4.1"),
)

Côté subscriber, les mêmes valeurs apparaissent dans properties. Attention à la forme : les valeurs sont des listes, car MQTT 5 autorise certaines propriétés à apparaître plusieurs fois :

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}

Lisez-les de façon défensive, par exemple ctype = properties.get("content_type", [None])[0]. Si le message a attendu dans une file avant livraison, le broker envoie le temps d'expiration restant : le subscriber peut donc voir une valeur inférieure à 60 — un moyen pratique de vérifier que le message expiry fonctionne. Pour savoir quand les user properties sont le bon outil et quand elles ne le sont pas, voir MQTT user properties.

Options d'abonnement

MQTT 5 ajoute des options par abonnement. gmqtt les expose via des objets Subscription :

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

Quand un message correspond à l'abonnement ci-dessus, properties["subscription_identifier"] contient [7]. Le routage dans on_message devient alors simple et explicite, sans recomparer les chaînes de topic.

Request/response avec response topic et correlation data

MQTT 5 standardise le request/response : le demandeur définit un response topic et des correlation data, le répondeur publie sa réponse sur ce topic et renvoie les correlation data.

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

Dans le on_message du demandeur, comparez properties["correlation_data"][0] à corr pour associer la réponse à sa requête. Le schéma complet, timeouts compris, est décrit dans MQTT request/response.

Reason codes et déconnexions propres

MQTT 5 remplace les anciens codes de retour d'un octet par des reason codes détaillés dans CONNACK, SUBACK, PUBACK et DISCONNECT. Dans gmqtt, vous les trouvez à plusieurs endroits :

• rc dans on_connect — 0 signifie succès ; des valeurs comme 134 (Bad User Name or Password) ou 135 (Not authorized) expliquent une connexion refusée.
• Le tuple qos dans on_subscribe(client, mid, qos, properties) — une valeur de 128 ou plus pour un topic signifie que le broker a refusé cet abonnement, généralement à cause d'une ACL.
• La valeur de retour de on_message sert de reason code dans le PUBACK/PUBREC que vous renvoyez ; 0 signifie succès.

Recherchez n'importe quel code avec l'outil de recherche de reason codes MQTT.

Pour vous déconnecter tout en laissant le Last Will partir (utile pour tester la supervision), utilisez le reason code 4 : await client.disconnect(reason_code=4). Le Will lui-même se configure avec un objet Message, éventuellement avec un 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)

Reconnexions et session expiry

Contrairement à aiomqtt, gmqtt se reconnecte automatiquement. Vous pouvez régler le nombre de tentatives et le délai :

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

Combinez les reconnexions avec session_expiry_interval et un client ID fixe : le broker conserve vos abonnements et met en file les messages QoS 1 pendant votre absence, et les flags du CONNACK indiquent si une session a été reprise. Lors d'une nouvelle connexion, flags dans on_connect indique session present ; s'il vaut 0 après une panne, la session a expiré et vous devez vous réabonner (s'abonner dans on_connect, comme ci-dessus, s'en charge). Pour le vérifier concrètement, voir tester le session expiry.

Tester votre client gmqtt

1. Lancez le subscriber du premier exemple.
2. Ouvrez le client MQTT en ligne, connectez-vous à broker.connectmq.com en MQTT 5 et publiez sur testmqtt/gmqtt/sensor-1/temperature avec une user property et un message expiry.
3. Vérifiez que le dictionnaire de propriétés dans votre terminal contient exactement ce que vous avez envoyé.
4. Arrêtez le subscriber, publiez un message QoS 1, relancez-le dans la fenêtre de session expiry : le message doit arriver à la reconnexion.

Publier des messages MQTT 5 vers votre client gmqtt

gmqtt face aux autres clients Python

gmqtt convient bien quand les métadonnées MQTT 5 sont au cœur de votre conception — request/response, content types, expiration — et que votre code est déjà asynchrone. Comme il est maintenu par une équipe plus réduite que paho-mqtt, vérifiez les releases récentes et figez la version avec laquelle vous testez. Pour une vue d'ensemble, voir MQTT 5 vs 3.1.1.

gmqtt donne à MQTT 5 l'allure de Python ordinaire : les propriétés sont des arguments, les résultats des dictionnaires, et les reconnexions se font toutes seules. Affichez une fois les propriétés du CONNACK, lisez les valeurs de propriétés comme des listes et vérifiez chaque fonctionnalité avec un second client avant de vous y fier en production.

Questions fréquentes