MQTT asíncrono en Python con aiomqtt

aiomqtt es la forma idiomática de usar MQTT desde código Python con asyncio. Envuelve paho-mqtt en un context manager asíncrono: async with aiomqtt.Client(...) conecta, await client.publish(...) envía y async for message in client.messages recibe — sin callbacks y sin un hilo en segundo plano que coordinar con tu event loop.

Este tutorial cubre la instalación, la suscripción y la publicación, un bucle de reconexión que sobrevive a reinicios del broker, el procesamiento concurrente de mensajes lentos, TLS y cómo integrar el cliente en una aplicación FastAPI.

Instalación y primera conexión

aiomqtt necesita Python 3.8 o superior y trae paho-mqtt como única dependencia:

pip install aiomqtt

El programa mínimo que funciona publica un mensaje y termina:

import asyncio
import aiomqtt

async def main():
    async with aiomqtt.Client("broker.connectmq.com", port=1883) as client:
        await client.publish("testmqtt/aiomqtt/hello", payload="hola desde asyncio", qos=1)

asyncio.run(main())

Al salir del bloque async with se envía un DISCONNECT limpio. Si el broker exige credenciales, pasa username y password; fija un client ID estable con identifier (genera uno único con el generador de client ID para evitar choques con otros clientes de prueba):

import os

client = aiomqtt.Client(
    "broker.connectmq.com",
    port=1883,
    identifier="testmqtt-aiomqtt-demo-01",
    username=os.environ["MQTT_USER"],
    password=os.environ["MQTT_PASS"],
    keepalive=30,
    protocol=aiomqtt.ProtocolVersion.V5,  # por defecto es MQTT 3.1.1
)

Suscribirse con async for

Desde aiomqtt 2.0, todos los mensajes entrantes llegan a un único iterador asíncrono, client.messages. Primero te suscribes y luego iteras:

import asyncio
import aiomqtt

async def main():
    async with aiomqtt.Client("broker.connectmq.com") as client:
        await client.subscribe("testmqtt/aiomqtt/#", qos=1)
        async for message in client.messages:
            if message.topic.matches("testmqtt/aiomqtt/+/temperature"):
                print("temp", message.topic.value, message.payload.decode())
            elif message.topic.matches("testmqtt/aiomqtt/+/status"):
                print("status", message.topic.value, message.payload.decode())

asyncio.run(main())

message.topic es un objeto Topic: .value es el string y .matches() aplica las reglas de wildcards de MQTT, así que puedes enrutar mensajes de varias suscripciones en un solo bucle. Si no tienes claro si un filtro coincide, compruébalo con el topic matcher. message.payload suele ser bytes; decodifícalo o pásalo a json.loads. También tienes message.qos, message.retain y, en MQTT 5, message.properties.

Para probar el suscriptor, abre el cliente MQTT online, conéctate a broker.connectmq.com y publica en testmqtt/aiomqtt/kitchen/temperature.

Publicar y recibir a la vez

Un mismo cliente puede publicar mientras otra corrutina consume client.messages. Ejecuta ambas dentro de un TaskGroup (Python 3.11+):

import asyncio
import json
import time
import aiomqtt

TOPIC = "testmqtt/aiomqtt/sensor-1/temperature"

async def produce(client: aiomqtt.Client):
    while True:
        payload = json.dumps({"t": round(time.time()), "value": 21.5})
        await client.publish(TOPIC, payload=payload, qos=1)
        await asyncio.sleep(2)

async def consume(client: aiomqtt.Client):
    await client.subscribe("testmqtt/aiomqtt/#")
    async for message in client.messages:
        print(message.topic.value, json.loads(message.payload))

async def main():
    async with aiomqtt.Client("broker.connectmq.com") as client:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(produce(client))
            tg.create_task(consume(client))

asyncio.run(main())

await client.publish() retorna cuando el mensaje se ha entregado con el QoS pedido (con QoS 1, tras el PUBACK). No lo llames desde otro hilo: mantén todo en el event loop.

Un patrón de reconexión que funciona de verdad

aiomqtt no se reconecta por sí solo. Cuando la red se cae o el broker se reinicia, el bloque async with lanza aiomqtt.MqttError, normalmente con un mensaje como Disconnected during message iteration o [code:7] The connection was lost. El objeto cliente es reutilizable, así que el patrón estándar es un bucle alrededor del context manager:

import asyncio
import random
import aiomqtt

async def main():
    client = aiomqtt.Client("broker.connectmq.com", identifier="testmqtt-aiomqtt-worker")
    delay = 1
    while True:
        try:
            async with client:
                delay = 1                                  # conectado: reinicia el backoff
                await client.subscribe("testmqtt/aiomqtt/#", qos=1)
                async for message in client.messages:
                    print(message.topic.value, message.payload)
        except aiomqtt.MqttError as err:
            wait = delay + random.uniform(0, delay)        # jitter
            print(f"conexión perdida ({err}); reconectando en {wait:.1f}s")
            await asyncio.sleep(wait)
            delay = min(delay * 2, 60)                     # backoff exponencial

asyncio.run(main())

Hay tres detalles importantes. Las suscripciones se hacen dentro del bloque, así que se restauran tras cada reconexión. La espera crece exponencialmente con jitter, de modo que una flota de workers no machaca al broker de forma sincronizada cuando vuelve — consulta las buenas prácticas de reconexión. Y un identifier fijo te permite combinarlo con una sesión persistente (clean_session=False en MQTT 3.1.1, o clean_start más un session expiry interval en MQTT 5) para que los mensajes QoS 1 publicados mientras estabas desconectado se entreguen igualmente.

Para probarlo, ejecuta el script, reinicia tu broker local (o desconecta la red) y observa el log: deberías ver una línea conexión perdida y después los mensajes fluyendo de nuevo.

No bloquees el bucle de mensajes

El bucle async for procesa un mensaje cada vez. Si un handler espera una escritura lenta en base de datos o una llamada HTTP, todos los demás mensajes esperan detrás. Delega el trabajo lento a tareas y limita la concurrencia con un semáforo:

sem = asyncio.Semaphore(20)

async def handle(message: aiomqtt.Message):
    async with sem:
        await save_to_db(message)          # tu I/O lento

async with asyncio.TaskGroup() as tg:
    async for message in client.messages:
        tg.create_task(handle(message))

El trabajo intensivo en CPU (decodificar imágenes, inferencia de ML) sigue bloqueando el loop; muévelo a asyncio.to_thread() o a un pool de procesos. Los mensajes entrantes se almacenan en una cola interna; max_queued_incoming_messages limita su tamaño si un consumidor lento hiciera crecer la memoria sin control.

TLS y WebSockets

Para tráfico de producción, usa TLS en el puerto 8883. Pasa un ssl.SSLContext estándar:

import ssl

client = aiomqtt.Client(
    "broker.connectmq.com",
    port=8883,
    tls_context=ssl.create_default_context(),
    username=os.environ["MQTT_USER"],
    password=os.environ["MQTT_PASS"],
)

En redes que solo permiten HTTPS, cambia el transporte a WebSockets seguros: port=8084, transport="websockets", websocket_path="/mqtt" junto con el mismo tls_context. Los certificados de cliente y las CA propias se explican en certificados TLS de MQTT en Python.

Usar aiomqtt dentro de FastAPI

FastAPI ya corre sobre un event loop, así que aiomqtt encaja de forma natural. Abre la conexión en el handler lifespan, arranca el listener como tarea en segundo plano y expón el cliente a los handlers de peticiones mediante app.state:

import asyncio
from contextlib import asynccontextmanager
import aiomqtt
from fastapi import FastAPI

latest: dict[str, str] = {}

async def listen(client: aiomqtt.Client):
    await client.subscribe("testmqtt/aiomqtt/+/temperature")
    async for message in client.messages:
        latest[message.topic.value] = message.payload.decode()

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with aiomqtt.Client("broker.connectmq.com") as client:
        app.state.mqtt = client
        task = asyncio.create_task(listen(client))
        yield
        task.cancel()

app = FastAPI(lifespan=lifespan)

@app.post("/publish/{device}")
async def publish(device: str, value: float):
    await app.state.mqtt.publish(f"testmqtt/aiomqtt/{device}/temperature", payload=str(value), qos=1)
    return {"ok": True}

@app.get("/latest")
async def get_latest():
    return latest

Ejecútalo con uvicorn app:app — con un solo worker. Cada proceso worker abre su propia conexión MQTT, así que varios workers con el mismo identifier se expulsarían unos a otros del broker. Una versión completa con push por WebSocket a los navegadores y gestión de reconexión está en MQTT con FastAPI.

¿aiomqtt o paho a secas?

Para una visión más amplia de las opciones, consulta la comparativa de librerías cliente MQTT para Python.

Mira tus mensajes de aiomqtt en directo

aiomqtt mantiene el código MQTT legible: un context manager para la conexión, un bucle para los mensajes y un simple try/except para los fallos. Añade el bucle de reconexión desde el primer día, saca el trabajo lento del bucle de mensajes y verifica todo con un segundo cliente antes de desplegar.

Preguntas frecuentes