MQTT 5 en Python con gmqtt

gmqtt es un cliente MQTT asyncio para Python diseñado en torno a MQTT 5 desde el principio. Cada propiedad de MQTT 5 — session expiry, message expiry, user properties, response topic, correlation data, subscription identifiers — es un simple argumento con nombre, y las propiedades entrantes llegan como un diccionario. Si quieres experimentar con funciones de MQTT 5 desde Python sin construir objetos Properties a mano, gmqtt es el camino más corto.

Este tutorial recorre la conexión con MQTT 5, la publicación con propiedades, las opciones de suscripción, request/response, la lectura de reason codes y cómo probarlo todo contra un broker real.

Instalar y conectar

pip install gmqtt

gmqtt usa callbacks que se ejecutan en el event loop de asyncio. Conecta, suscríbete en on_connect y mantén vivo el programa con 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("conectado, rc =", rc, "propiedades del broker:", properties)
    client.subscribe("testmqtt/gmqtt/#", qos=1)

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

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

async def main():
    client = MQTTClient(
        "testmqtt-gmqtt-demo-01",
        session_expiry_interval=3600,   # conserva la sesión 1 h tras desconectar
        receive_maximum=100,            # como máximo 100 mensajes QoS 1/2 sin confirmar
    )
    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())

Dos cosas son propias de gmqtt. Los argumentos con nombre que pasas al constructor de MQTTClient se convierten en propiedades del CONNECT — aquí session_expiry_interval y receive_maximum. Y el argumento properties de on_connect trae lo que el broker devolvió en el CONNACK: su propio receive maximum, topic alias maximum, maximum packet size o un client identifier asignado, según el broker. Imprimirlo una vez es la forma más rápida de ver qué soporta realmente tu broker.

Para TLS, pasa ssl=True (o un ssl.SSLContext) y el puerto 8883: await client.connect("broker.connectmq.com", 8883, ssl=True).

Publicar con propiedades de MQTT 5

Las propiedades de publicación son argumentos con nombre 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,                     # descartar si no se entrega en 60 s
    user_property=("firmware", "2.4.1"),
)

En el suscriptor, los mismos valores aparecen en properties. Fíjate en la forma: los valores son listas, porque MQTT 5 permite que algunas propiedades aparezcan más de una vez:

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}

Léelos de forma defensiva, p. ej. ctype = properties.get("content_type", [None])[0]. Si el mensaje estuvo en cola antes de entregarse, el broker envía el tiempo de expiración restante, así que el suscriptor puede ver un valor menor que 60 — una forma práctica de verificar que el message expiry funciona. Para saber cuándo las user properties son la herramienta adecuada y cuándo no, consulta MQTT user properties.

Opciones de suscripción

MQTT 5 añade opciones por suscripción. gmqtt las expone mediante objetos 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,
)

Cuando un mensaje coincide con la suscripción anterior, properties["subscription_identifier"] contiene [7]. Así el enrutamiento en on_message es barato y explícito, sin volver a comparar strings de topic.

Request/response con response topic y correlation data

MQTT 5 estandariza request/response: quien pregunta fija un response topic y correlation data; quien responde publica allí su respuesta y copia de vuelta la 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,
)

En el on_message del solicitante, compara properties["correlation_data"][0] con corr para asociar la respuesta con su petición. El patrón completo, con timeouts, está en MQTT request/response.

Reason codes y desconexiones limpias

MQTT 5 sustituye los antiguos códigos de retorno de un byte por reason codes detallados en CONNACK, SUBACK, PUBACK y DISCONNECT. En gmqtt los ves en varios sitios:

• rc en on_connect — 0 es éxito; valores como 134 (Bad User Name or Password) o 135 (Not authorized) explican una conexión rechazada.
• La tupla qos en on_subscribe(client, mid, qos, properties) — un valor de 128 o más para un topic significa que el broker rechazó esa suscripción, normalmente por una ACL.
• El valor de retorno de on_message se usa como reason code en el PUBACK/PUBREC que devuelves; 0 significa éxito.

Busca cualquier número con el buscador de reason codes de MQTT.

Para desconectarte dejando que el Last Will se envíe igualmente (útil para probar la monitorización), usa el reason code 4: await client.disconnect(reason_code=4). El Will se configura con un objeto Message, opcionalmente con 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)

Reconexiones y session expiry

A diferencia de aiomqtt, gmqtt se reconecta automáticamente. Puedes ajustar cuántas veces lo intenta y cuánto espera:

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

Combina las reconexiones con session_expiry_interval y un client ID fijo: el broker conserva tus suscripciones y encola los mensajes QoS 1 mientras no estás, y los flags del CONNACK te dicen si se reanudó una sesión. En una nueva conexión, flags en on_connect indica session present; si es 0 tras una caída, la sesión expiró y debes volver a suscribirte (suscribirte en on_connect, como arriba, lo resuelve). Para comprobarlo en la práctica, consulta probar session expiry.

Probar tu cliente gmqtt

1. Ejecuta el suscriptor del primer ejemplo.
2. Abre el cliente MQTT online, conéctate a broker.connectmq.com con MQTT 5 y publica en testmqtt/gmqtt/sensor-1/temperature con una user property y un message expiry.
3. Confirma que el diccionario de propiedades en tu terminal contiene exactamente lo que enviaste.
4. Detén el suscriptor, publica un mensaje QoS 1 y vuelve a arrancarlo dentro de la ventana de session expiry: el mensaje debería llegar al reconectar.

Publica mensajes MQTT 5 hacia tu cliente gmqtt

gmqtt frente a otros clientes Python

gmqtt encaja bien cuando los metadatos de MQTT 5 son centrales en tu diseño — request/response, content types, expiración — y tu código ya es asíncrono. Como lo mantiene un equipo más pequeño que paho-mqtt, revisa los releases recientes y fija la versión con la que pruebas. Para el panorama general, consulta MQTT 5 vs 3.1.1.

gmqtt hace que MQTT 5 se sienta como Python normal: las propiedades son argumentos, los resultados son diccionarios y las reconexiones ocurren solas. Imprime una vez las propiedades del CONNACK, lee los valores de las propiedades como listas y verifica cada función con un segundo cliente antes de confiar en ella en producción.

Preguntas frecuentes