MQTT avec FastAPI : un pont entre HTTP et MQTT

Un pont FastAPI–MQTT permet à des clients HTTP ordinaires de dialoguer avec des appareils : un POST devient une publication MQTT, et les messages des appareils sont diffusés aux navigateurs via WebSocket. FastAPI reposant sur asyncio, la façon la plus propre d'y parvenir est un client natif asyncio (aiomqtt) piloté par une seule tâche de fond démarrée dans le lifespan de l'application.

Ce tutoriel construit un service petit mais conçu comme en production : un endpoint REST qui envoie des commandes, un subscriber qui survit aux redémarrages du broker, un WebSocket qui pousse la télémétrie en direct vers les navigateurs, et les règles à suivre avec plusieurs workers.

Architecture

Une connexion MQTT par processus, ouverte au démarrage et partagée par toutes les requêtes. Les handlers HTTP ne créent jamais de client : ouvrir une session MQTT par requête est lent, inonde le broker de connexions et casse dès que deux requêtes utilisent le même client ID.

pip install "fastapi[standard]" aiomqtt

Le pont MQTT avec reconnexions

Le pont possède le client, sait s'il est connecté et redistribue les messages entrants à chaque WebSocket ouvert grâce à de petites files par connexion :

import asyncio
import os
import uuid
import aiomqtt

BROKER = os.getenv("MQTT_HOST", "broker.connectmq.com")
TELEMETRY = "testmqtt/fastapi/+/telemetry"


class Hub:
    """Redistribue les messages MQTT aux clients WebSocket."""

    def __init__(self):
        self.queues: set[asyncio.Queue] = set()

    def join(self) -> asyncio.Queue:
        q: asyncio.Queue = asyncio.Queue(maxsize=100)
        self.queues.add(q)
        return q

    def leave(self, q: asyncio.Queue):
        self.queues.discard(q)

    def broadcast(self, item: dict):
        for q in list(self.queues):
            try:
                q.put_nowait(item)
            except asyncio.QueueFull:
                pass  # navigateur lent : on jette plutôt que de bloquer tout le monde


class MqttBridge:
    def __init__(self, hub: Hub):
        self.hub = hub
        self.connected = asyncio.Event()
        self.client = aiomqtt.Client(
            BROKER,
            port=1883,
            identifier=f"testmqtt-fastapi-{uuid.uuid4().hex[:8]}",  # unique par processus
            username=os.getenv("MQTT_USER"),
            password=os.getenv("MQTT_PASS"),
        )

    async def run(self):
        delay = 1
        while True:
            try:
                async with self.client:
                    self.connected.set()
                    delay = 1
                    await self.client.subscribe(TELEMETRY, qos=1)
                    async for msg in self.client.messages:
                        self.hub.broadcast({
                            "topic": msg.topic.value,
                            "payload": msg.payload.decode(errors="replace"),
                        })
            except aiomqtt.MqttError as err:
                self.connected.clear()
                print(f"Connexion MQTT perdue : {err} ; nouvel essai dans {delay}s")
                await asyncio.sleep(delay)
                delay = min(delay * 2, 30)

    async def publish(self, topic: str, payload: str, qos: int = 1):
        if not self.connected.is_set():
            raise ConnectionError("MQTT broker not connected")
        await self.client.publish(topic, payload, qos=qos)

La boucle run() reprend le schéma de reconnexion du guide aiomqtt : abonnement à l'intérieur du async with, backoff exponentiel et un flag connected fiable pour que les handlers HTTP échouent vite au lieu de rester bloqués.

L'application FastAPI et le lifespan

import asyncio
import contextlib
import json
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import FastAPI, HTTPException, Path, WebSocket, WebSocketDisconnect
from pydantic import BaseModel

from bridge import Hub, MqttBridge

hub = Hub()
bridge = MqttBridge(hub)


@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(bridge.run())
    with contextlib.suppress(TimeoutError):
        await asyncio.wait_for(bridge.connected.wait(), timeout=10)
    yield
    task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await task


app = FastAPI(lifespan=lifespan)
DeviceId = Annotated[str, Path(pattern=r"^[A-Za-z0-9_-]{1,64}$")]


class Command(BaseModel):
    action: str
    value: float | None = None


@app.post("/devices/{device_id}/commands", status_code=202)
async def send_command(device_id: DeviceId, cmd: Command):
    try:
        await bridge.publish(f"testmqtt/fastapi/{device_id}/commands", cmd.model_dump_json())
    except ConnectionError:
        raise HTTPException(503, "MQTT broker not connected")
    return {"queued": True, "topic": f"testmqtt/fastapi/{device_id}/commands"}


@app.get("/health")
async def health():
    return {"mqtt_connected": bridge.connected.is_set()}


@app.websocket("/ws")
async def live(websocket: WebSocket):
    await websocket.accept()
    queue = hub.join()
    try:
        while True:
            await websocket.send_text(json.dumps(await queue.get()))
    except WebSocketDisconnect:
        pass
    finally:
        hub.leave(queue)

Quelques choix délibérés :

• Le paramètre de chemin est validé. device_id fait partie d'un topic : une regex rejette +, # et les autres caractères qui produiraient un topic invalide ou trop large. N'insérez jamais une saisie utilisateur brute dans un nom de topic.
• 202 Accepted, pas 200. Une publication QoS 1 réussie signifie que le broker a le message — pas que l'appareil a exécuté la commande. Si vous avez besoin d'une confirmation, faites répondre l'appareil sur un response topic.
• Le démarrage attend jusqu'à 10 secondes la première connexion, sans faire échouer l'application si le broker est indisponible ; /health expose l'état pour que votre orchestrateur décide.
• Des files bornées par WebSocket empêchent un navigateur lent de ralentir la livraison pour tous les autres.

Lancez-le avec fastapi dev main.py (ou uvicorn main:app).

Push vers le navigateur

Le navigateur n'a besoin que de l'API WebSocket standard — il ne parle jamais MQTT directement, vous n'exposez donc aucun identifiant du broker côté frontend :

<ul id="log"></ul>
<script>
  const ws = new WebSocket(`ws://${location.host}/ws`);
  ws.onmessage = (event) => {
    const { topic, payload } = JSON.parse(event.data);
    const li = document.createElement("li");
    li.textContent = `${topic}: ${payload}`;
    document.getElementById("log").prepend(li);
  };
</script>

Si vous préférez que le navigateur s'abonne lui-même au broker, utilisez MQTT sur WebSockets sur le port 8084 — voir MQTT sur WebSockets. L'approche par pont est préférable quand l'API doit filtrer, autoriser ou enrichir les messages avant que les utilisateurs ne les voient ; pour une interface complète construite ainsi, voir le tutoriel de tableau de bord en direct.

Tester le pont de bout en bout

1. Démarrez l'application et vérifiez que curl localhost:8000/health renvoie {"mqtt_connected": true}.
2. Ouvrez le client MQTT en ligne, connectez-vous à broker.connectmq.com et abonnez-vous à testmqtt/fastapi/#.
3. Envoyez une commande en HTTP et regardez-la apparaître dans le client en ligne :

curl -X POST localhost:8000/devices/sensor-1/commands \
  -H 'content-type: application/json' \
  -d '{"action": "set_interval", "value": 30}'

1. Connectez-vous au WebSocket (par exemple avec websocat ws://localhost:8000/ws), puis publiez {"temp": 22.4} depuis le client en ligne sur testmqtt/fastapi/sensor-1/telemetry. websocat doit l'afficher presque immédiatement.
2. Redémarrez votre broker ou coupez le réseau : /health doit passer à false, l'endpoint de commandes doit renvoyer 503, et les deux doivent se rétablir seuls.

Observer le trafic du pont FastAPI

Pour les tests automatisés, with TestClient(app) as client: exécute le lifespan, donc le pont se connecte comme en production ; pointez MQTT_HOST vers un broker dans un conteneur. Des fixtures prêtes à l'emploi sont décrites dans fixtures pytest pour MQTT.

Plusieurs workers

uvicorn --workers 4 ou Gunicorn démarrent plusieurs processus, et chacun exécute le lifespan — vous obtenez donc quatre connexions MQTT. Aucun problème si vous l'anticipez :

• Des client IDs uniques par processus. Le pont ajoute un suffixe aléatoire. Avec un ID fixe, les workers se déconnecteraient mutuellement en boucle ; voir client ID dupliqué.
• Le fan-out WebSocket exige que chaque worker reçoive chaque message, car un navigateur peut être rattaché à n'importe lequel d'entre eux. Les abonnements classiques font exactement cela.
• Le traitement ne doit avoir lieu qu'une fois. Si un worker écrit aussi la télémétrie en base, utilisez pour ce consommateur un abonnement partagé comme $share/api/testmqtt/fastapi/+/telemetry, afin que le broker répartisse les messages entre workers au lieu de les dupliquer. Voir abonnements partagés.

Alternatives

• paho-mqtt avec loop_start() fonctionne, mais ses callbacks s'exécutent dans un thread séparé. Transmettez chaque message avec loop.call_soon_threadsafe(...) et ne touchez jamais directement aux objets asyncio depuis le callback.
• fastapi-mqtt est une extension communautaire basée sur gmqtt qui ajoute des décorateurs pour les abonnements. Elle évite du code répétitif ; le pont explicite ci-dessus offre plus de contrôle sur les reconnexions et la contre-pression.

Un pont FastAPI–MQTT se résume à une connexion durable par processus, gérée dans lifespan, avec un état de santé fiable et des files bornées vers les navigateurs. Validez tout ce qui devient un topic, gardez des client IDs uniques par worker et testez la boucle complète — HTTP entrant, MQTT sortant, MQTT entrant, WebSocket sortant — avant d'y brancher des appareils.

Questions fréquentes