MQTT en Python con paho-mqtt: tutorial de publicación y suscripción

paho-mqtt es la librería cliente MQTT estándar para Python. Soporta MQTT 3.1, 3.1.1 y 5, TCP sin cifrar, TLS y WebSockets, y funciona en cualquier sitio, desde una Raspberry Pi hasta un servidor en la nube. Este tutorial usa la API actual de paho-mqtt 2.x — los artículos antiguos escritos para 1.x lanzarán errores si copias su código tal cual, así que iremos señalando las diferencias por el camino.

Instalar paho-mqtt

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

Comprueba tu versión con python -c "import paho.mqtt; print(paho.mqtt.__version__)". Todos los ejemplos siguientes usan 2.x y el sandbox público de HiveMQ en broker.hivemq.com, así que puedes ejecutarlos sin montar un broker. Si prefieres tener el tuyo, consulta cómo instalar Mosquitto.

Qué cambió en paho-mqtt 2.x

El gran cambio es la versión de la API de callbacks. El primer argumento de mqtt.Client() es ahora un CallbackAPIVersion, y con VERSION2 todos los callbacks reciben códigos de motivo y propiedades al estilo de MQTT 5 — incluso cuando te conectas con MQTT 3.1.1.

reason_code es un objeto ReasonCode: comprueba reason_code.is_failure en lugar de comparar rc == 0. Nuestro buscador de códigos de motivo MQTT explica qué significa cada valor.

Un suscriptor completo

El suscriptor siguiente se conecta, se suscribe dentro de on_connect e imprime cada mensaje. Suscribirse en on_connect y no después de connect() es importante: cuando la conexión se cae y paho se reconecta, el callback se vuelve a ejecutar y tus suscripciones se restauran.

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 vacío deja que paho genere uno aleatorio. Si defines el tuyo, haz que sea único — dos clientes con el mismo ID se expulsan mutuamente del broker una y otra vez. El generador de client ID puede crear IDs seguros por ti.

Un publicador completo

Un publicador suele tener otras tareas — leer un sensor, consultar una API — así que ejecuta el bucle de red en un hilo en segundo plano con 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()

Ejecuta el suscriptor en una terminal y el publicador en otra. También puedes ver desde el navegador los mensajes que publica tu script:

Mira tus mensajes de Python en directo

loop_forever vs loop_start vs loop

• loop_forever() — bloquea el hilo actual, gestiona las reconexiones y solo retorna tras disconnect(). Lo mejor para suscriptores puros y servicios.
• loop_start() / loop_stop() — ejecuta el bucle en un hilo en segundo plano. Lo mejor cuando tu hilo principal hace otras cosas. Los callbacks se ejecutan en ese hilo en segundo plano, así que protege el estado compartido en consecuencia.
• loop(timeout) — procesa el tráfico de red una vez. Solo hace falta si integras paho en tu propio bucle de eventos.

QoS, mensajes retenidos y Last Will

Pasa qos=0, 1 o 2 tanto a publish() como a subscribe(); el QoS efectivo es el menor de los dos. retain=True hace que el broker guarde el último mensaje para los nuevos suscriptores. Configura un Last Will antes de conectar para que el broker anuncie una desconexión inesperada:

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)

Más información en niveles de QoS de MQTT, mensajes retenidos y Last Will and Testament.

TLS y autenticación por usuario/contraseña

Cualquier despliegue real debería usar TLS en el puerto 8883 además de credenciales. Ambas cosas se configuran antes de 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()

Evita tls_insecure_set(True) fuera de pruebas locales rápidas — desactiva la verificación del nombre de host. Para MQTT 5, crea el cliente con protocol=mqtt.MQTTv5. Para WebSockets, pasa transport="websockets" y llama a client.ws_set_options(path="/mqtt").

Reconectar de forma fiable

Tanto loop_forever() como loop_start() se reconectan automáticamente tras perder la conexión. Ajusta el backoff con reconnect_delay_set():

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

El primer connect() es distinto: lanza una excepción (por ejemplo ConnectionRefusedError o un timeout de socket) si el broker no es accesible. Envuélvelo en un bucle de reintentos o usa connect_async() seguido de loop_start(), que reintenta en segundo plano. Si las conexiones siguen fallando, la guía de errores de conexión MQTT repasa los sospechosos habituales: puerto equivocado, firewall bloqueando, credenciales incorrectas o client ID duplicados.

Usar funciones de MQTT 5

Pasa protocol=mqtt.MQTTv5 para desbloquear MQTT 5: caducidad de sesión, propiedades de usuario, caducidad de mensajes y códigos de motivo detallados. Las opciones de conexión pasan de clean_session (un argumento del constructor en 3.1.1) a clean_start en connect(), y los metadatos de cada mensaje van en un objeto 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()

En on_message, las propiedades recibidas están disponibles en msg.properties. ¿No tienes claro si necesitas la versión 5? Nuestra comparativa MQTT 5 vs 3.1.1 enumera las diferencias que importan en la práctica.

Errores habituales

• Copiar código de 1.x. mqtt.Client("my-id") pasa el client ID donde 2.x espera la versión de la API de callbacks. Usa el argumento con nombre client_id=.
• Olvidar el bucle. Sin loop_forever() o loop_start(), no se envía ni se recibe nada y los callbacks nunca se ejecutan.
• Salir demasiado pronto. Un script que llama a publish() y termina enseguida puede cerrarse antes de que el mensaje salga del socket. Llama a wait_for_publish() o usa publish.single().
• Decodificar a ciegas. msg.payload es bytes. Usa msg.payload.decode("utf-8") para texto y json.loads() para JSON, y captura los errores de decodificación — otros clientes del mismo topic pueden enviar cualquier cosa.
• Suscribirse a # en un broker público. Recibirás el tráfico de todos los demás usuarios. Quédate dentro de tu propio prefijo de topic; consulta topics y comodines MQTT.

Publicar un único mensaje

Para tareas cron y scripts que solo envían un mensaje, el helper paho.mqtt.publish se encarga de conectar, publicar y desconectar en una sola llamada:

import paho.mqtt.publish as publish

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

¿Quieres este código ya rellenado con tu host, puerto, credenciales y topic? Usa el generador de código MQTT para generarlo con la configuración de tu broker.

Preguntas frecuentes