MQTT com FastAPI: uma ponte entre HTTP e MQTT

Uma ponte FastAPI–MQTT permite que clientes HTTP comuns conversem com dispositivos: um POST vira uma publicação MQTT, e as mensagens dos dispositivos chegam aos navegadores por WebSocket. Como o FastAPI é construído sobre asyncio, o jeito mais limpo é um cliente nativo de asyncio (aiomqtt) controlado por uma única task em segundo plano que inicia no lifespan do app.

Este tutorial constrói um serviço pequeno, mas com cara de produção: um endpoint REST que envia comandos, um subscriber que sobrevive a reinícios do broker, um WebSocket que envia telemetria ao vivo para navegadores e as regras de escala para vários workers.

Arquitetura

Uma conexão MQTT por processo, aberta na inicialização e compartilhada por todas as requisições. Handlers HTTP nunca criam clientes — abrir uma sessão MQTT por requisição é lento, inunda o broker de conexões e quebra assim que duas requisições usam o mesmo client ID.

pip install "fastapi[standard]" aiomqtt

A ponte MQTT com reconexão

A ponte é dona do cliente, sabe se está conectada e distribui as mensagens recebidas para cada WebSocket aberto por meio de pequenas filas por conexão:

import asyncio
import os
import uuid
import aiomqtt

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


class Hub:
    """Distribui mensagens MQTT para os 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 em vez de travar 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 processo
            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"Conexão MQTT perdida: {err}; nova tentativa em {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)

O loop run() é o padrão de reconexão do guia do aiomqtt: assinar dentro do async with, backoff exponencial e um flag connected confiável para que os handlers HTTP falhem rápido em vez de travar.

O app FastAPI e o 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)

Algumas escolhas deliberadas:

• O parâmetro de caminho é validado. device_id vira parte de um topic, então uma regex rejeita +, # e outros caracteres que gerariam um topic inválido ou amplo demais. Nunca interpole entrada do usuário sem validação em nomes de topic.
• 202 Accepted, não 200. Uma publicação QoS 1 bem-sucedida significa que o broker tem a mensagem — não que o dispositivo executou o comando. Se você precisa de confirmação, faça o dispositivo responder em um response topic.
• A inicialização espera até 10 segundos pela primeira conexão, mas não derruba o app se o broker estiver fora; /health informa o estado para que seu orquestrador decida.
• Filas limitadas por WebSocket fazem com que um navegador lento não trave a entrega para todos os outros.

Rode com fastapi dev main.py (ou uvicorn main:app).

Push para o navegador

O navegador só precisa da API padrão de WebSocket — ele nunca fala MQTT diretamente, então você não expõe credenciais do broker no 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>

Se preferir que o navegador assine direto no broker, use MQTT sobre WebSockets na porta 8084 — veja MQTT sobre WebSockets. A abordagem de ponte é melhor quando você quer que a API filtre, autorize ou enriqueça as mensagens antes de os usuários as verem; para uma UI completa construída assim, veja o tutorial de dashboard ao vivo.

Testar a ponte de ponta a ponta

1. Inicie o app e confira se curl localhost:8000/health retorna {"mqtt_connected": true}.
2. Abra o cliente MQTT online, conecte em broker.connectmq.com e assine testmqtt/fastapi/#.
3. Envie um comando por HTTP e veja-o aparecer no cliente online:

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

1. Conecte ao WebSocket (por exemplo com websocat ws://localhost:8000/ws) e publique {"temp": 22.4} pelo cliente online em testmqtt/fastapi/sensor-1/telemetry. O websocat deve imprimir a mensagem em instantes.
2. Reinicie o broker ou derrube a rede: /health deve mudar para false, o endpoint de comandos deve retornar 503, e ambos devem se recuperar sozinhos.

Acompanhe o tráfego da ponte FastAPI

Para testes automatizados, with TestClient(app) as client: executa o lifespan, então a ponte conecta como em produção; aponte MQTT_HOST para um broker em um container. Fixtures prontas estão em fixtures de pytest para MQTT.

Rodando vários workers

uvicorn --workers 4 ou Gunicorn iniciam vários processos, e cada um executa o lifespan — ou seja, quatro conexões MQTT. Tudo bem, desde que você planeje:

• Client IDs únicos por processo. A ponte acrescenta um sufixo aleatório. Com um ID fixo, os workers derrubariam uns aos outros em loop; veja client ID duplicado.
• O fan-out por WebSocket exige que cada worker receba todas as mensagens, porque um navegador pode estar conectado a qualquer um deles. Assinaturas normais fazem exatamente isso.
• O processamento deve acontecer uma vez. Se um worker também grava telemetria em um banco, use para esse consumidor uma shared subscription como $share/api/testmqtt/fastapi/+/telemetry, para que o broker distribua as mensagens entre os workers em vez de duplicá-las. Veja shared subscriptions.

Alternativas

• paho-mqtt com loop_start() funciona, mas seus callbacks rodam em outra thread. Repasse cada mensagem com loop.call_soon_threadsafe(...) e nunca mexa em objetos asyncio diretamente no callback.
• fastapi-mqtt é uma extensão da comunidade baseada no gmqtt que adiciona decorators para assinaturas. Economiza código repetitivo; a ponte explícita acima dá mais controle sobre reconexões e contrapressão.

Uma ponte FastAPI–MQTT se resume a uma conexão de longa duração por processo, gerenciada no lifespan, com relatório de saúde confiável e filas limitadas para os navegadores. Valide tudo o que vira topic, mantenha client IDs únicos por worker e teste o ciclo completo — HTTP entra, MQTT sai, MQTT entra, WebSocket sai — antes de colocar dispositivos atrás dela.

Perguntas frequentes