MQTT with FastAPI: Bridge HTTP and MQTT

A FastAPI MQTT bridge lets ordinary HTTP clients talk to devices: a POST turns into an MQTT publish, and messages from devices are streamed to browsers over a WebSocket. Because FastAPI is built on asyncio, the cleanest way to do this is an asyncio-native client (aiomqtt) owned by one background task that starts in the app's lifespan.

This tutorial builds a small but production-shaped service: a REST endpoint that sends commands, a subscriber that survives broker restarts, a WebSocket that pushes live telemetry to browsers, and the scaling rules for multiple workers.

Architecture

One MQTT connection per process, opened at startup and shared by all requests. HTTP handlers never create clients — opening an MQTT session per request is slow, floods the broker with connects and breaks as soon as two requests use the same client ID.

pip install "fastapi[standard]" aiomqtt

The MQTT bridge with reconnects

The bridge owns the client, tracks whether it is connected, and fans incoming messages out to every open WebSocket through small per-connection queues:

import asyncio
import os
import uuid
import aiomqtt

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


class Hub:
    """Fans MQTT messages out to WebSocket clients."""

    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  # slow browser: drop rather than block everyone


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 per process
            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"MQTT connection lost: {err}; retrying in {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)

The run() loop is the reconnect pattern from the aiomqtt guide: subscribe inside the async with, back off exponentially, and keep the connected flag honest so HTTP handlers can fail fast instead of hanging.

The FastAPI app and 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)

A few deliberate choices:

• The path parameter is validated. device_id becomes part of a topic, so a regex rejects +, # and other characters that would produce an invalid or overly broad topic. Never interpolate raw user input into topic names.
• 202 Accepted, not 200. A successful QoS 1 publish means the broker has the message — not that the device has executed the command. If you need confirmation, have the device reply on a response topic.
• Startup waits up to 10 seconds for the first connection but does not fail the app if the broker is down; /health reports the state so your orchestrator can decide.
• Bounded queues per WebSocket mean one slow browser cannot stall delivery to everyone else.

Run it with fastapi dev main.py (or uvicorn main:app).

Push to the browser

The browser only needs the standard WebSocket API — it never talks MQTT directly, so you do not expose broker credentials to the 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>

If you would rather have the browser subscribe to the broker itself, use MQTT over WebSockets on port 8084 instead — see MQTT over WebSockets. The bridge approach is better when you want the API to filter, authorise or enrich messages before users see them; for a full UI built on it, see the live dashboard tutorial.

Test the bridge end to end

1. Start the app and check curl localhost:8000/health returns {"mqtt_connected": true}.
2. Open the online MQTT client, connect to broker.connectmq.com and subscribe to testmqtt/fastapi/#.
3. Send a command over HTTP and watch it appear in the online client:

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

1. Connect to the WebSocket (for example with websocat ws://localhost:8000/ws), then publish {"temp": 22.4} from the online client to testmqtt/fastapi/sensor-1/telemetry. It should be printed by websocat within a moment.
2. Restart your broker or disconnect the network: /health should flip to false, the command endpoint should return 503, and both should recover on their own.

Watch the FastAPI bridge traffic

For automated tests, with TestClient(app) as client: runs the lifespan, so the bridge connects as it does in production; point MQTT_HOST at a broker in a container. Ready-made fixtures are in MQTT pytest fixtures.

Running several workers

uvicorn --workers 4 or Gunicorn starts several processes, and each one runs the lifespan — so you get four MQTT connections. That is fine if you plan for it:

• Unique client IDs per process. The bridge appends a random suffix. A fixed ID would make the workers disconnect each other in a loop; see duplicate client ID.
• WebSocket fan-out needs every worker to receive every message, because a browser can be attached to any of them. Plain subscriptions do exactly that.
• Processing must happen once. If a worker also writes telemetry to a database, use a shared subscription such as $share/api/testmqtt/fastapi/+/telemetry for that consumer so the broker load-balances messages across workers instead of duplicating them. See shared subscriptions.

Alternatives

• paho-mqtt with loop_start() works, but its callbacks run in a separate thread. Forward each message with loop.call_soon_threadsafe(...) and never touch asyncio objects directly from the callback.
• fastapi-mqtt is a community extension built on gmqtt that adds decorators for subscriptions. It saves boilerplate; the explicit bridge above gives you more control over reconnects and back-pressure.

A FastAPI–MQTT bridge comes down to one long-lived connection per process, managed in lifespan, with honest health reporting and bounded queues towards browsers. Validate everything that becomes a topic, keep client IDs unique per worker, and test the full loop — HTTP in, MQTT out, MQTT in, WebSocket out — before you put devices behind it.

Frequently asked questions