MQTT assíncrono em Python com aiomqtt

aiomqtt é a forma idiomática de usar MQTT em código Python com asyncio. Ele encapsula o paho-mqtt em um context manager assíncrono: async with aiomqtt.Client(...) conecta, await client.publish(...) envia e async for message in client.messages recebe — sem callbacks e sem uma thread em segundo plano para coordenar com o seu event loop.

Este tutorial cobre instalação, assinatura e publicação, um loop de reconexão que sobrevive a reinícios do broker, processamento concorrente de mensagens lentas, TLS e como encaixar o cliente em uma aplicação FastAPI.

Instalação e primeira conexão

O aiomqtt precisa de Python 3.8 ou mais recente e traz o paho-mqtt como única dependência:

pip install aiomqtt

O menor programa funcional publica uma mensagem e 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="oi do asyncio", qos=1)

asyncio.run(main())

Sair do bloco async with envia um DISCONNECT limpo. Se o broker exigir credenciais, passe username e password; defina um client ID estável com identifier (gere um único com o gerador de client ID para evitar conflitos com outros clientes de teste):

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,  # o padrão é MQTT 3.1.1
)

Assinar com async for

Desde o aiomqtt 2.0, todas as mensagens recebidas chegam por um único iterador assíncrono, client.messages. Assine primeiro e depois itere:

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 é um objeto Topic: .value é a string e .matches() aplica as regras de wildcard do MQTT, então você pode rotear mensagens de várias assinaturas em um só loop. Se não tiver certeza se um filtro casa, confira no topic matcher. message.payload normalmente é bytes; decodifique ou passe para json.loads. Também existem message.qos, message.retain e, no MQTT 5, message.properties.

Para testar o subscriber, abra o cliente MQTT online, conecte em broker.connectmq.com e publique em testmqtt/aiomqtt/kitchen/temperature.

Publicar e receber ao mesmo tempo

Um único cliente pode publicar enquanto outra corrotina consome client.messages. Rode as duas em um 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 quando a mensagem foi entregue no QoS pedido (no QoS 1, depois do PUBACK). Não chame a partir de outra thread — mantenha tudo no event loop.

Um padrão de reconexão que funciona de verdade

O aiomqtt não reconecta sozinho. Quando a rede cai ou o broker reinicia, o bloco async with lança aiomqtt.MqttError, normalmente com uma mensagem como Disconnected during message iteration ou [code:7] The connection was lost. O objeto cliente é reutilizável, então o padrão é um loop em volta do 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: zera o 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"conexão perdida ({err}); reconectando em {wait:.1f}s")
            await asyncio.sleep(wait)
            delay = min(delay * 2, 60)                     # backoff exponencial

asyncio.run(main())

Três detalhes importam. As assinaturas são feitas dentro do bloco, então são restauradas após cada reconexão. A espera cresce exponencialmente com jitter, para que uma frota de workers não martele o broker em sincronia quando ele volta — veja as boas práticas de reconexão. E um identifier fixo permite combinar isso com uma sessão persistente (clean_session=False no MQTT 3.1.1, ou clean_start mais um session expiry interval no MQTT 5) para que mensagens QoS 1 publicadas enquanto você estava offline ainda sejam entregues.

Para testar, rode o script, reinicie seu broker local (ou desconecte a rede) e observe o log: você deve ver uma linha conexão perdida e depois as mensagens voltando a fluir.

Não bloqueie o loop de mensagens

O loop async for trata uma mensagem por vez. Se um handler aguarda uma escrita lenta no banco ou uma chamada HTTP, todas as outras mensagens esperam atrás dele. Passe o trabalho lento para tasks e limite a concorrência com um semáforo:

sem = asyncio.Semaphore(20)

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

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

Trabalho pesado de CPU (decodificar imagens, inferência de ML) continua bloqueando o loop; mova-o para asyncio.to_thread() ou um pool de processos. As mensagens recebidas ficam em uma fila interna; max_queued_incoming_messages limita o tamanho dela caso um consumidor lento fizesse a memória crescer sem limite.

TLS e WebSockets

Para tráfego de produção, use TLS na porta 8883. Passe um ssl.SSLContext padrão:

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

Em redes que só liberam HTTPS, troque o transporte para WebSockets seguros: port=8084, transport="websockets", websocket_path="/mqtt" junto com o mesmo tls_context. Certificados de cliente e CAs próprias estão em certificados TLS do MQTT em Python.

Usando aiomqtt dentro do FastAPI

O FastAPI já roda em um event loop, então o aiomqtt se encaixa naturalmente. Abra a conexão no handler lifespan, inicie o listener como task em segundo plano e exponha o cliente aos handlers de requisição via 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

Rode com uvicorn app:app — com um único worker. Cada processo worker abre a própria conexão MQTT, então vários workers com o mesmo identifier derrubariam uns aos outros do broker. Uma versão completa com push por WebSocket para navegadores e tratamento de reconexão está em MQTT com FastAPI.

aiomqtt ou paho puro?

Para uma visão mais ampla, veja a comparação de bibliotecas cliente MQTT para Python.

Veja suas mensagens do aiomqtt ao vivo

O aiomqtt mantém o código MQTT legível: um context manager para a conexão, um loop para as mensagens e um try/except simples para as falhas. Adicione o loop de reconexão desde o primeiro dia, tire o trabalho lento do loop de mensagens e verifique tudo com um segundo cliente antes de fazer o deploy.

Perguntas frequentes