Async MQTT in Python with aiomqtt and asyncio: Tasks, Workers and Reconnects

aiomqtt with asyncio gives you MQTT in Python without callbacks: you await a publish, iterate over incoming messages with async for, and run as many concurrent tasks as you need on a single thread. In this tutorial you will build an async client connected to broker.connectmq.com that publishes sensor readings, consumes messages through a worker queue, and reconnects cleanly when the network drops. You will also see the Windows event loop detail that trips up almost everyone.

Why aiomqtt Instead of Plain paho-mqtt?

paho-mqtt is a callback library. Its network loop runs in its own thread and calls your functions when something happens. That is fine for small scripts, but it gets awkward when your application is already built on asyncio, for example a FastAPI service, a gateway that also talks HTTP and Modbus asynchronously, or a bot. Mixing threads and an event loop means run_coroutine_threadsafe, locks and subtle shutdown bugs.

aiomqtt wraps paho-mqtt and exposes an idiomatic asyncio API instead:

Under the hood it is still paho, so protocol behaviour is identical. If you are coming from callbacks, the paho mqtt python tutorial is a useful reference for the concepts.

Install aiomqtt

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install "aiomqtt>=2.0"

The examples use asyncio.TaskGroup and except*, which need Python 3.11 or newer. Pick a random topic prefix for the public MQTT test broker so your tests stay separate from other users; here we use connectmq/demo/k7f3q2.

The Windows event loop gotcha

aiomqtt relies on the event loop's add_reader and add_writer methods. On Windows the default event loop is the Proactor loop, which does not implement them, so you get a NotImplementedError the moment you connect. Run your program on a selector event loop instead, as the examples below do. Linux and macOS already use a selector loop by default.

A Minimal aiomqtt asyncio Example

Start with the smallest working subscriber and publisher in one file:

import asyncio
import sys
import uuid

import aiomqtt

BROKER = "broker.connectmq.com"
PREFIX = "connectmq/demo/k7f3q2"


async def main():
    async with aiomqtt.Client(BROKER, 1883, identifier=f"aio-{uuid.uuid4().hex[:8]}") as client:
        await client.subscribe(f"{PREFIX}/hello", qos=1)
        await client.publish(f"{PREFIX}/hello", payload="hello from asyncio", qos=1)
        async for message in client.messages:
            print(message.topic, message.payload.decode())
            break  # stop after the first message


if sys.platform == "win32":
    asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop)
else:
    asyncio.run(main())

The async with block connects on entry and disconnects on exit, even if an exception is raised. client.messages is an async iterator that yields every message from every subscription in arrival order.

A Realistic Client: Concurrent Tasks and a Worker Queue

Real applications do several things at once: publish readings on a schedule, react to commands, and process incoming data that may involve slow I/O like a database write or an HTTP call. The trick is to never block the loop that reads messages. Save this as aio_client.py:

import asyncio
import json
import random
import sys
import time
import uuid

import aiomqtt

BROKER = "broker.connectmq.com"
PORT = 1883
PREFIX = "connectmq/demo/k7f3q2"
CLIENT_ID = f"aio-{uuid.uuid4().hex[:8]}"
RECONNECT_INTERVAL = 5
WORKERS = 3


async def publish_readings(client: aiomqtt.Client) -> None:
    seq = 0
    while True:
        reading = {"seq": seq, "temperature": round(random.uniform(20, 25), 2), "ts": time.time()}
        await client.publish(f"{PREFIX}/sensors/room1", json.dumps(reading), qos=1)
        seq += 1
        await asyncio.sleep(2)


async def consume(client: aiomqtt.Client, queue: asyncio.Queue) -> None:
    await client.subscribe(f"{PREFIX}/sensors/+", qos=1)
    await client.subscribe(f"{PREFIX}/commands/#", qos=1)
    async for message in client.messages:
        await queue.put(message)  # hand off quickly, keep reading


async def handle(message: aiomqtt.Message) -> None:
    if message.topic.matches(f"{PREFIX}/commands/#"):
        print(f"Command on {message.topic}: {message.payload.decode()}")
        return
    data = json.loads(message.payload)
    await asyncio.sleep(0.5)  # stands in for a slow database or HTTP call
    print(f"Processed reading {data['seq']}: {data['temperature']} C")


async def worker(name: int, queue: asyncio.Queue) -> None:
    while True:
        message = await queue.get()
        try:
            await handle(message)
        except Exception as exc:  # one bad message must not kill the worker
            print(f"Worker {name} failed on {message.topic}: {exc}")
        finally:
            queue.task_done()


async def main() -> None:
    queue: asyncio.Queue = asyncio.Queue(maxsize=1000)
    workers = [asyncio.create_task(worker(i, queue)) for i in range(WORKERS)]

    client = aiomqtt.Client(
        BROKER,
        PORT,
        identifier=CLIENT_ID,
        protocol=aiomqtt.ProtocolVersion.V5,
        keepalive=30,
        will=aiomqtt.Will(f"{PREFIX}/status/{CLIENT_ID}", payload="offline", qos=1, retain=True),
        # Private ConnectMQ broker:
        # username="YOUR_USERNAME", password="YOUR_PASSWORD",
        # tls_context=ssl.create_default_context(), and port 8883
    )

    try:
        while True:
            try:
                async with client:
                    print("Connected")
                    await client.publish(f"{PREFIX}/status/{CLIENT_ID}", "online", qos=1, retain=True)
                    async with asyncio.TaskGroup() as tg:
                        tg.create_task(publish_readings(client))
                        tg.create_task(consume(client, queue))
            except* aiomqtt.MqttError as group:
                print(f"Connection lost: {group.exceptions[0]}. Reconnecting in {RECONNECT_INTERVAL}s")
            await asyncio.sleep(RECONNECT_INTERVAL)
    finally:
        for w in workers:
            w.cancel()


if __name__ == "__main__":
    try:
        if sys.platform == "win32":
            asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop)
        else:
            asyncio.run(main())
    except KeyboardInterrupt:
        print("Stopped")

Run python aio_client.py. Every two seconds a reading is published, received back through the subscription and processed by one of the three workers. Send a command from another terminal:

mosquitto_pub -h broker.connectmq.com -p 1883 -i aio-cmd-test -t 'connectmq/demo/k7f3q2/commands/led' -m 'on'

How It Works

Structured concurrency with TaskGroup

The publisher and the consumer run as sibling tasks inside a TaskGroup. If either raises, for example because the connection dropped and publish() fails, the group cancels the other task and re-raises the error wrapped in an exception group. except* aiomqtt.MqttError catches it whichever task failed. Without a task group, a failure in one task can go unnoticed while the other keeps running on a dead connection.

Keeping the message loop fast

The consumer does nothing except move messages into an asyncio.Queue. The workers do the slow part. This matters because aiomqtt delivers messages in order through a single iterator; if you awaited a half-second database write inside the async for loop, every other message would wait behind it. With three workers, three messages are processed concurrently.

The queue's maxsize provides backpressure. If workers cannot keep up, queue.put() waits, which slows reading from the broker rather than growing memory without limit. If ordering per topic matters, route messages to workers by topic hash instead of a shared queue.

Graceful reconnects

An aiomqtt client is reusable but not reentrant: you can enter async with client again after it exits, but not twice at the same time. The outer while True loop does exactly that. After a failure it waits a few seconds, reconnects, re-publishes the online status and re-subscribes, since subscriptions belong to the connection. The Last Will publishes "offline" if the process dies without a clean disconnect.

For production, replace the fixed interval with exponential backoff capped at a minute or two, so a fleet of clients does not hammer the broker in lockstep after an outage. The paho-mqtt reconnect guide discusses backoff and session strategy in more depth, and the same ideas apply here.

Using MQTT 5 Properties With aiomqtt

Because the client above connects with ProtocolVersion.V5, you can attach MQTT 5 properties to individual messages. aiomqtt reuses paho's Properties class, so the code is the same as in a callback-based client:

from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

props = Properties(PacketTypes.PUBLISH)
props.MessageExpiryInterval = 60          # drop if not delivered within 60 s
props.ContentType = "application/json"
await client.publish(f"{PREFIX}/sensors/room1", payload, qos=1, properties=props)

Message expiry is useful for commands: a "turn on" that arrives ten minutes late is often worse than one that never arrives. On the receiving side, message.properties exposes the same attributes. The MQTT 5 features guide explains which properties are worth using.

Common Problems

1. NotImplementedError on Windows: you are on the Proactor loop. Use the selector loop as shown.
2. MqttError: Could not connect: port 1883 is blocked or the hostname is wrong. Try TLS on 8883 with a tls_context.
3. Connection drops every few seconds: a second client is using the same identifier. Generate a unique one per process.
4. Messages arrive late or in bursts: something slow runs inside async for. Move it into workers.
5. Rate-limited on the public broker: it allows 10 messages per second per client and 10 subscriptions per client. Keep the publish interval sensible.
6. Blocking calls freeze everything: a synchronous requests.get() or time.sleep() inside a coroutine blocks the whole loop. Use async libraries or asyncio.to_thread().

Going to Production

On a private ConnectMQ broker you pass the generated username, password and client ID as identifier, plus tls_context=ssl.create_default_context() on port 8883. The Python MQTT TLS guide explains certificate verification, and the Python MQTT JSON tutorial shows how to version payloads like the readings above. For storing what the workers process, adapt the MQTT to SQLite Python logger, and for a comparison with clients in other languages see the MQTT client libraries overview. The aiomqtt project on GitHub documents every client option.

Run Your Async Python Services on ConnectMQ

ConnectMQ is a managed MQTT broker supporting MQTT 3.1.1 and 5.0 over TCP, TLS and WebSocket, with session expiry, retained messages and Last Will that this client relies on. Develop against broker.connectmq.com, then deploy to a private cloud MQTT broker without changing your code beyond the connection settings. Sign up free on the Serverless plan, no credit card required.

Frequently Asked Questions

Is aiomqtt faster than paho-mqtt?

It uses paho-mqtt internally, so raw protocol throughput is similar. The advantage is structure: async code lets one thread handle many concurrent operations without callbacks or locks.

Why does aiomqtt fail on Windows?

The default Windows Proactor event loop lacks the add_reader and add_writer methods aiomqtt needs. Run your program on asyncio.SelectorEventLoop and it works.

Does aiomqtt reconnect automatically?

No. A dropped connection raises aiomqtt.MqttError, and you reconnect by re-entering async with client in a loop, which also lets you control backoff and re-subscription.

Can I use aiomqtt with FastAPI?

Yes. Start the client in the application's lifespan handler and run the consumer as a background task on the same event loop.