MQTT con FastAPI: un puente entre HTTP y MQTT

Un puente FastAPI–MQTT permite que clientes HTTP normales hablen con dispositivos: un POST se convierte en una publicación MQTT y los mensajes de los dispositivos llegan a los navegadores por WebSocket. Como FastAPI está construido sobre asyncio, la forma más limpia de hacerlo es un cliente nativo de asyncio (aiomqtt) gestionado por una única tarea en segundo plano que arranca en el lifespan de la app.

Este tutorial construye un servicio pequeño pero con forma de producción: un endpoint REST que envía comandos, un suscriptor que sobrevive a reinicios del broker, un WebSocket que empuja telemetría en vivo a los navegadores y las reglas de escalado con varios workers.

Arquitectura

Una conexión MQTT por proceso, abierta al arrancar y compartida por todas las peticiones. Los handlers HTTP nunca crean clientes: abrir una sesión MQTT por petición es lento, inunda el broker de conexiones y falla en cuanto dos peticiones usan el mismo client ID.

pip install "fastapi[standard]" aiomqtt

El puente MQTT con reconexiones

El puente es dueño del cliente, sabe si está conectado y reparte los mensajes entrantes a cada WebSocket abierto mediante pequeñas colas por conexión:

import asyncio
import os
import uuid
import aiomqtt

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


class Hub:
    """Reparte los mensajes MQTT a los clientes 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  # navegador lento: descartar en lugar de bloquear a todos


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]}",  # único por proceso
            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"Conexión MQTT perdida: {err}; reintento en {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)

El bucle run() es el patrón de reconexión de la guía de aiomqtt: suscribirse dentro del async with, backoff exponencial y un flag connected fiable para que los handlers HTTP fallen rápido en lugar de quedarse colgados.

La app FastAPI y el 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)

Algunas decisiones deliberadas:

• El parámetro de ruta se valida. device_id pasa a formar parte de un topic, así que una regex rechaza +, # y otros caracteres que darían un topic inválido o demasiado amplio. Nunca interpoles entrada de usuario sin validar en nombres de topic.
• 202 Accepted, no 200. Una publicación QoS 1 correcta significa que el broker tiene el mensaje, no que el dispositivo haya ejecutado el comando. Si necesitas confirmación, haz que el dispositivo responda en un response topic.
• El arranque espera hasta 10 segundos a la primera conexión, pero no hace fallar la app si el broker está caído; /health informa del estado para que tu orquestador decida.
• Colas acotadas por WebSocket: un navegador lento no puede frenar la entrega a todos los demás.

Ejecútalo con fastapi dev main.py (o uvicorn main:app).

Push al navegador

El navegador solo necesita la API estándar de WebSocket — nunca habla MQTT directamente, así que no expones credenciales del broker en el 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 prefieres que el navegador se suscriba directamente al broker, usa MQTT sobre WebSockets en el puerto 8084 — consulta MQTT sobre WebSockets. El enfoque de puente es mejor cuando quieres que la API filtre, autorice o enriquezca los mensajes antes de que los vean los usuarios; para una UI completa construida así, consulta el tutorial de dashboard en vivo.

Probar el puente de extremo a extremo

1. Arranca la app y comprueba que curl localhost:8000/health devuelve {"mqtt_connected": true}.
2. Abre el cliente MQTT online, conéctate a broker.connectmq.com y suscríbete a testmqtt/fastapi/#.
3. Envía un comando por HTTP y mira cómo aparece en el cliente online:

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

1. Conéctate al WebSocket (por ejemplo con websocat ws://localhost:8000/ws) y publica {"temp": 22.4} desde el cliente online en testmqtt/fastapi/sensor-1/telemetry. websocat debería imprimirlo al instante.
2. Reinicia tu broker o corta la red: /health debería pasar a false, el endpoint de comandos debería devolver 503, y ambos deberían recuperarse solos.

Observa el tráfico del puente FastAPI

Para tests automatizados, with TestClient(app) as client: ejecuta el lifespan, de modo que el puente se conecta igual que en producción; apunta MQTT_HOST a un broker en un contenedor. Tienes fixtures listas en fixtures de pytest para MQTT.

Ejecutar varios workers

uvicorn --workers 4 o Gunicorn arrancan varios procesos, y cada uno ejecuta el lifespan: tendrás cuatro conexiones MQTT. No es un problema si lo planificas:

• Client IDs únicos por proceso. El puente añade un sufijo aleatorio. Con un ID fijo, los workers se desconectarían unos a otros en bucle; consulta client ID duplicado.
• El fan-out por WebSocket necesita que cada worker reciba todos los mensajes, porque un navegador puede estar conectado a cualquiera de ellos. Las suscripciones normales hacen justo eso.
• El procesamiento debe ocurrir una sola vez. Si un worker también escribe telemetría en una base de datos, usa para ese consumidor una shared subscription como $share/api/testmqtt/fastapi/+/telemetry, para que el broker reparta los mensajes entre workers en lugar de duplicarlos. Consulta shared subscriptions.

Alternativas

• paho-mqtt con loop_start() funciona, pero sus callbacks se ejecutan en otro hilo. Reenvía cada mensaje con loop.call_soon_threadsafe(...) y nunca toques objetos de asyncio directamente desde el callback.
• fastapi-mqtt es una extensión de la comunidad basada en gmqtt que añade decoradores para las suscripciones. Ahorra código repetitivo; el puente explícito de arriba te da más control sobre las reconexiones y la contrapresión.

Un puente FastAPI–MQTT se reduce a una conexión de larga duración por proceso, gestionada en lifespan, con un estado de salud fiable y colas acotadas hacia los navegadores. Valida todo lo que acabe en un topic, mantén client IDs únicos por worker y prueba el circuito completo — HTTP entra, MQTT sale, MQTT entra, WebSocket sale — antes de poner dispositivos detrás.

Preguntas frecuentes