Async MQTT in Python with aiomqtt

aiomqtt is the idiomatic way to use MQTT from Python asyncio code. It wraps paho-mqtt in an async context manager: async with aiomqtt.Client(...) connects, await client.publish(...) sends, and async for message in client.messages receives — no callbacks, no background thread to coordinate with your event loop.

This tutorial covers installation, subscribing and publishing, a reconnect loop that survives broker restarts, handling slow messages concurrently, TLS, and how to plug the client into a FastAPI application.

Install and first connection

aiomqtt needs Python 3.8 or newer and pulls in paho-mqtt as its only dependency:

pip install aiomqtt

The smallest working program publishes one message and exits:

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="hi from asyncio", qos=1)

asyncio.run(main())

Leaving the async with block sends a clean DISCONNECT. If the broker requires credentials, pass username and password; set a stable client ID with identifier (generate a unique one with the client ID generator to avoid clashes with other test clients):

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,  # default is MQTT 3.1.1
)

Subscribe with async for

Since aiomqtt 2.0, all incoming messages arrive on one async iterator, client.messages. Subscribe first, then iterate:

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 is a Topic object: .value is the string, and .matches() applies MQTT wildcard rules, so you can route messages from several subscriptions in one loop. If you are unsure whether a filter matches, check it with the topic matcher. message.payload is usually bytes; decode it or pass it to json.loads. message.qos, message.retain and, for MQTT 5, message.properties are also available.

To test the subscriber, open the online MQTT client, connect to broker.connectmq.com and publish to testmqtt/aiomqtt/kitchen/temperature.

Publish and receive at the same time

A single client can publish while another coroutine is consuming client.messages. Run both under a 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() returns once the message has been handed over at the requested QoS (for QoS 1, after the PUBACK). Do not call it from another thread — keep everything on the event loop.

A reconnect pattern that actually works

aiomqtt does not reconnect on its own. When the network drops or the broker restarts, the async with block raises aiomqtt.MqttError, typically with a message like Disconnected during message iteration or [code:7] The connection was lost. The client object is reusable, so the standard pattern is a loop around the 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                                  # connected: reset 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"connection lost ({err}); reconnecting in {wait:.1f}s")
            await asyncio.sleep(wait)
            delay = min(delay * 2, 60)                     # exponential backoff

asyncio.run(main())

Three details matter. Subscriptions are made inside the block, so they are restored after every reconnect. The delay grows exponentially with jitter, so a fleet of workers does not hammer the broker in sync when it comes back — see reconnect best practices. And a fixed identifier lets you combine this with a persistent session (clean_session=False on MQTT 3.1.1, or clean_start plus a session expiry interval on MQTT 5) so QoS 1 messages published while you were offline are still delivered.

To test it, run the script, restart your local broker (or pull the network cable), and watch the log: you should see one connection lost line, then messages flowing again.

Don't block the message loop

The async for loop handles one message at a time. If a handler awaits a slow database write or HTTP call, every other message waits behind it. Hand slow work to tasks and cap concurrency with a semaphore:

sem = asyncio.Semaphore(20)

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

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

CPU-heavy work (decoding images, ML inference) still blocks the loop; move it to asyncio.to_thread() or a process pool. Incoming messages are buffered in an internal queue; max_queued_incoming_messages limits its size if a slow consumer would otherwise grow memory without bound.

TLS and WebSockets

For production traffic, use TLS on port 8883. Pass a standard ssl.SSLContext:

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

For networks that only allow HTTPS, switch the transport to secure WebSockets: port=8084, transport="websockets", websocket_path="/mqtt" together with the same tls_context. Client certificates and custom CAs are covered in MQTT TLS certificates in Python.

Using aiomqtt inside FastAPI

FastAPI already runs on an event loop, so aiomqtt fits naturally. Open the connection in the lifespan handler, start the listener as a background task, and expose the client to request handlers through 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

Run it with uvicorn app:app — with a single worker. Every worker process opens its own MQTT connection, so several workers with the same identifier would kick each other off the broker. A complete version with WebSocket push to browsers and reconnect handling is in MQTT with FastAPI.

aiomqtt or plain paho?

For a wider look at the options, see the Python MQTT client libraries comparison.

Watch your aiomqtt messages live

aiomqtt keeps MQTT code readable: one context manager for the connection, one loop for messages, and plain try/except for failures. Add the reconnect loop from day one, keep slow work out of the message loop, and verify everything with a second client before you deploy.

Frequently asked questions