zmqtt

Pure asyncio MQTT 3.1.1 and 5.0 for Python 3.10+, with deterministic subscription routing, bounded queues, and correlation-safe request/response.

Documentation · PyPI · Changelog

Why zmqtt?

• Application-facing subscriptions. Each Subscription owns its filters, bounded queue, acknowledgement policy, and async iterator.
• Deterministic routing. Wildcards, shared subscriptions, broker decorator prefixes, and MQTT 5 subscription identifiers are handled by the built-in router. One incoming PUBLISH is delivered once to the selected subscription.
• High-level QoS. Packet identifiers and the QoS 1/2 protocol handshakes are managed by the client. Manual acknowledgement is opt-in per subscription.
• Safe MQTT 5 request/response. Concurrent requests are matched by both response topic and correlation data and delivered exclusively to one pending request.
• Bounded by default. Subscription queues and pending request futures apply backpressure instead of growing without limit.
• Self-contained MQTT stack. The packet codec and protocol engine live in zmqtt; there is no Paho or other MQTT client dependency.
• One API for MQTT 3.1.1 and 5.0. create_client() returns a version-typed client, so a type checker can reject MQTT 5-only calls on a 3.1.1 connection.

zmqtt and aiomqtt 3

aiomqtt 3 is no longer a Paho wrapper. It is also pure asyncio, fully typed, and has MQTT 5 flow control and opt-in automatic reconnection, but its API deliberately exposes more low-level MQTT 5 mechanics. zmqtt provides a higher-level application API and supports both current protocol versions.

The comparison targets the aiomqtt 3.0 alpha API, where application code supplies packet identifiers for QoS 1/2 publishes, drives the remaining QoS 2 steps, and implements message routing and response correlation. zmqtt keeps the decisions applications need explicit — QoS, acknowledgement timing, MQTT 5 properties, subscription lifecycle, and reconnect policy — while handling that protocol machinery in the client. The result is less application code, fewer protocol edge cases, and one typed API for both MQTT 3.1.1 and 5.0.

Installation

pip install zmqtt

Quick start

import asyncio

from zmqtt import QoS, create_client


async def main() -> None:
    async with create_client("localhost") as client:
        async with client.subscribe(
            "sensors/#",
            qos=QoS.AT_LEAST_ONCE,
        ) as messages:
            await client.publish(
                "sensors/temperature",
                "23.5",
                qos=QoS.AT_LEAST_ONCE,
            )
            msg = await messages.get_message()
            print(msg.topic, msg.payload.decode())


asyncio.run(main())

create_client() defaults to MQTT 3.1.1. Pass version="5.0" for the MQTT 5 API.

Publish

from zmqtt import QoS

await client.publish("events/online", b"device-42")
await client.publish(
    "commands/restart",
    b"device-42",
    qos=QoS.AT_LEAST_ONCE,
    retain=False,
)

Payloads may be bytes or str; strings are encoded as UTF-8. QoS 1 waits for PUBACK. QoS 2 completes the PUBREC/PUBREL/PUBCOMP handshake before returning.

Subscribe and route

async with client.subscribe(
    "sensors/+/temperature",
    "sensors/#",
    receive_buffer_size=100,
) as messages:
    async for msg in messages:
        print(msg.topic, msg.payload)

If one PUBLISH matches multiple filters in the same subscription, zmqtt selects the most specific filter (literal before + before #) and enqueues the message once. MQTT 5 subscription identifiers are used when the broker supplies them, which disambiguates overlapping subscriptions reliably.

Shared subscriptions and broker decorator prefixes work with the same API:

async with client.subscribe("$share/workers/jobs/#") as jobs:
    async for job in jobs:
        await process(job)

The public topic_matches() helper follows the same matching rules, including $share, $queue, $exclusive, and configured stripped prefixes.

Manual acknowledgement

Manual acknowledgement only has an effect at QoS 1 or 2:

from zmqtt import QoS, create_client

client = create_client(
    "localhost",
    client_id="orders-worker-1",
    clean_session=False,
)

async with client:
    async with client.subscribe(
        "orders/#",
        qos=QoS.AT_LEAST_ONCE,
        auto_ack=False,
    ) as messages:
        async for msg in messages:
            await save_to_database(msg)
            await msg.ack()

A stable client ID and persistent broker session make an unacknowledged message eligible for redelivery after reconnect. They do not make application processing exactly once, so handlers should still be idempotent. MQTT 5 clients also need a positive session_expiry_interval for a session to survive disconnection.

MQTT 5 request/response

from zmqtt import create_client

async with create_client("localhost", version="5.0") as client:
    reply = await client.request(
        "services/echo",
        b"hello",
        timeout=5.0,
    )
    print(reply.payload)

zmqtt subscribes to the response topic before publishing, generates a response topic and correlation data when omitted, and matches the reply by the exact (response_topic, correlation_data) pair. Concurrent requests may share a response topic. A matching reply is delivered only to its pending request(); it is not also enqueued for an ordinary subscription. Unmatched or late replies fall through to normal routing, which selects at most one subscription. The request dispatcher itself does not retain them.

The responder must copy the request's correlation data unchanged:

from zmqtt import PublishProperties

async with client.subscribe("services/echo") as requests:
    async for request in requests:
        assert request.properties is not None
        assert request.properties.response_topic is not None
        assert request.properties.correlation_data is not None
        await client.publish(
            request.properties.response_topic,
            request.payload,
            properties=PublishProperties(
                correlation_data=request.properties.correlation_data,
            ),
        )

The timeout argument bounds the reply wait after the response subscription is ready and the request has been published. Use asyncio.wait_for() around the whole coroutine when setup and publishing must be included in the cancellation budget; cancellation cleanup can still extend the wall-clock completion time.

Reconnection

Automatic reconnection is enabled by default. Active subscriptions are re-registered after a successful reconnect while their application queues stay alive. The default policy makes at most five connection attempts:

from zmqtt import ReconnectConfig, create_client

client = create_client(
    "localhost",
    reconnect=ReconnectConfig(max_attempts=None),  # retry indefinitely
)

Broker refusals such as invalid credentials are not retried. See Reconnection for the complete failure semantics.

MQTT 5 properties

from zmqtt import PublishProperties, create_client

async with create_client("localhost", version="5.0") as client:
    await client.publish(
        "events/reading",
        b'{"value": 42}',
        properties=PublishProperties(
            content_type="application/json",
            message_expiry_interval=300,
            user_properties=(("source", "sensor-01"),),
        ),
    )

MQTT 5 also adds session expiry, subscription identifiers, no_local, retain handling, publish properties, and a low-level AUTH packet API.

Learn more

• Getting started
• Connecting and TLS
• Subscription routing
• Request / response
• API reference