MQTT asynchrone en Python avec aiomqtt

aiomqtt est la façon idiomatique d'utiliser MQTT depuis du code Python asyncio. La bibliothèque enveloppe paho-mqtt dans un context manager asynchrone : async with aiomqtt.Client(...) se connecte, await client.publish(...) envoie et async for message in client.messages reçoit — sans callbacks, sans thread d'arrière-plan à coordonner avec votre event loop.

Ce tutoriel couvre l'installation, l'abonnement et la publication, une boucle de reconnexion qui survit aux redémarrages du broker, le traitement concurrent des messages lents, TLS et l'intégration du client dans une application FastAPI.

Installation et première connexion

aiomqtt nécessite Python 3.8 ou plus récent et n'a qu'une dépendance, paho-mqtt :

pip install aiomqtt

Le plus petit programme fonctionnel publie un message puis se termine :

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="bonjour depuis asyncio", qos=1)

asyncio.run(main())

Quitter le bloc async with envoie un DISCONNECT propre. Si le broker exige des identifiants, passez username et password ; fixez un client ID stable avec identifier (générez-en un unique avec le générateur de client ID pour éviter les collisions avec d'autres clients de test) :

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,  # par défaut : MQTT 3.1.1
)

S'abonner avec async for

Depuis aiomqtt 2.0, tous les messages entrants arrivent sur un seul itérateur asynchrone, client.messages. Abonnez-vous d'abord, puis itérez :

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 est un objet Topic : .value est la chaîne, et .matches() applique les règles de wildcards MQTT, ce qui permet de router les messages de plusieurs abonnements dans une seule boucle. En cas de doute sur un filtre, vérifiez-le avec le topic matcher. message.payload est généralement de type bytes ; décodez-le ou passez-le à json.loads. Vous disposez aussi de message.qos, message.retain et, en MQTT 5, de message.properties.

Pour tester le subscriber, ouvrez le client MQTT en ligne, connectez-vous à broker.connectmq.com et publiez sur testmqtt/aiomqtt/kitchen/temperature.

Publier et recevoir en même temps

Un même client peut publier pendant qu'une autre coroutine consomme client.messages. Exécutez les deux dans 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() rend la main une fois le message transmis au QoS demandé (en QoS 1, après le PUBACK). Ne l'appelez pas depuis un autre thread : gardez tout sur l'event loop.

Un schéma de reconnexion qui fonctionne vraiment

aiomqtt ne se reconnecte pas tout seul. Quand le réseau tombe ou que le broker redémarre, le bloc async with lève aiomqtt.MqttError, généralement avec un message comme Disconnected during message iteration ou [code:7] The connection was lost. L'objet client est réutilisable ; le schéma standard est donc une boucle autour du 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                                  # connecté : on réinitialise le 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"connexion perdue ({err}) ; reconnexion dans {wait:.1f}s")
            await asyncio.sleep(wait)
            delay = min(delay * 2, 60)                     # backoff exponentiel

asyncio.run(main())

Trois détails comptent. Les abonnements sont faits à l'intérieur du bloc, donc rétablis après chaque reconnexion. Le délai croît exponentiellement avec du jitter, pour qu'une flotte de workers ne martèle pas le broker de façon synchronisée à son retour — voir les bonnes pratiques de reconnexion. Et un identifier fixe permet de combiner cela avec une session persistante (clean_session=False en MQTT 3.1.1, ou clean_start plus un session expiry interval en MQTT 5) afin que les messages QoS 1 publiés pendant votre absence soient tout de même livrés.

Pour le tester, lancez le script, redémarrez votre broker local (ou coupez le réseau) et observez le log : vous devez voir une ligne connexion perdue, puis les messages qui reviennent.

Ne bloquez pas la boucle de messages

La boucle async for traite un message à la fois. Si un handler attend une écriture lente en base ou un appel HTTP, tous les autres messages patientent derrière. Confiez le travail lent à des tâches et limitez la concurrence avec un sémaphore :

sem = asyncio.Semaphore(20)

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

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

Le travail gourmand en CPU (décodage d'images, inférence ML) bloque toujours la boucle ; déplacez-le vers asyncio.to_thread() ou un pool de processus. Les messages entrants sont mis en tampon dans une file interne ; max_queued_incoming_messages en limite la taille si un consommateur lent risquait de faire grossir la mémoire sans limite.

TLS et WebSockets

Pour le trafic de production, utilisez TLS sur le port 8883. Passez un ssl.SSLContext standard :

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

Sur les réseaux qui n'autorisent que HTTPS, passez aux WebSockets sécurisés : port=8084, transport="websockets", websocket_path="/mqtt" avec le même tls_context. Les certificats client et les CA personnalisées sont traités dans certificats TLS MQTT en Python.

Utiliser aiomqtt dans FastAPI

FastAPI tourne déjà sur un event loop, aiomqtt s'y intègre donc naturellement. Ouvrez la connexion dans le handler lifespan, démarrez l'écoute comme tâche de fond et exposez le client aux handlers de requêtes 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

Lancez-le avec uvicorn app:app — avec un seul worker. Chaque processus worker ouvre sa propre connexion MQTT ; plusieurs workers avec le même identifier s'éjecteraient mutuellement du broker. Une version complète avec push WebSocket vers les navigateurs et gestion de la reconnexion se trouve dans MQTT avec FastAPI.

aiomqtt ou paho seul ?

Pour un panorama plus large, consultez la comparaison des bibliothèques client MQTT Python.

Voir vos messages aiomqtt en direct

aiomqtt garde le code MQTT lisible : un context manager pour la connexion, une boucle pour les messages et un simple try/except pour les erreurs. Ajoutez la boucle de reconnexion dès le premier jour, sortez le travail lent de la boucle de messages et vérifiez tout avec un second client avant de déployer.

Questions fréquentes