MQTTium

MQTTium is a reliable, async-native MQTT client for modern Python.

It combines a synchronous protocol engine with an asyncio API, bounded backpressure, durable inflight persistence, explicit delivery receipts, and complete MQTT QoS state machines.

Status: alpha (0.1.0a2). The implementation is extensively tested, but the public API may still change before the first stable release.

Features

• MQTT 3.1.1 and MQTT 5;
• QoS 0, 1 and 2 with one authoritative protocol state machine;
• TCP, TLS, WebSocket and Unix transports;
• reconnect and session replay;
• bounded callback and async-iterator delivery;
• manual acknowledgement;
• in-memory and SQLite inflight persistence;
• aggregate publish_many() with bounded memory and measured throughput gains;
• an additive Paho VERSION2 compatibility façade, isolated from the native API;
• inline type information for type checkers.

Python 3.11–3.14 is supported. MQTTium is licensed under Apache-2.0.

Installation

python -m pip install mqttium

To work on the unreleased development version:

git clone https://github.com/yoch/mqttium.git
cd mqttium
python -m pip install -e ".[dev,fuzz,release]"

Quick start

import asyncio

from mqttium.api import AsyncClient


async def main() -> None:
    client = AsyncClient("demo-client")
    await client.connect("127.0.0.1", 1883)
    await client.subscribe("demo/#")

    receipt = await client.publish("demo/hello", b"world", qos=1)
    await receipt.wait()

    await client.disconnect()


asyncio.run(main())

Batched publishing

publish_many() consumes iterables in bounded chunks and returns one aggregate receipt rather than creating one task or event per message:

from mqttium.api import PublishMessage

batch = await client.publish_many(
    PublishMessage("telemetry/device-1", payload, qos=1)
    for payload in payloads
)
await batch.wait()

The retained paired A/B benchmark measured publisher-throughput geomean improvements of 36.9% for QoS 0, 15.9% for QoS 1, and 7.0% for QoS 2 against equivalent individual publishing pipelines on the validated source tree. Benchmark methodology and limitations are documented in docs/BENCHMARKING.md.

Bounded memory

Every queue that can grow with application load is bounded by default, so a producer that outruns its broker is slowed down rather than allowed to exhaust the process:

client = AsyncClient(
    max_pending_outbound_messages=10_000,   # unfinished QoS 1/2 publications
    max_pending_outbound_bytes=64 * 1024**2,  # their logical topic+payload+properties
    max_pending_delivery_bytes=64 * 1024**2,  # inbound messages awaiting a consumer
    publish_backpressure="wait",            # or "error" to refuse immediately
)

publish() waits for capacity by default and raises FlowControlError under publish_backpressure="error" or with nowait=True. A refusal is atomic: no packet identifier is allocated and no store record is written. Pass None for any limit to restore unbounded queueing.

These defaults are new in 0.1.0a2; before them a QoS 1/2 producer could queue until the 65 535 packet-identifier space was exhausted. See docs/MIGRATION.md.

Validation

The release gates include:

• more than 300 unit tests;
• Mosquitto integration tests on Python 3.11, 3.12, 3.13 and 3.14;
• deterministic and Hypothesis-based fuzzing;
• Ruff formatting and linting;
• mypy validation and a PEP 561 py.typed marker;
• an 80% coverage gate;
• wheel, source-distribution and isolated-install validation;
• delivery, persistence, TCP, TLS and WAN-profile benchmarks.

Documentation

• docs/DESIGN.md — architecture and invariants
• docs/IMPLEMENTATION-GUIDE.md — protocol contracts
• docs/COMPAT.md — Paho compatibility surface
• docs/MIGRATION.md — migration guidance
• docs/BENCHMARKING.md — benchmark validity contract
• docs/FUZZING.md — fuzzing strategy
• docs/ROADMAP.md — remaining stable-release work
• PROVENANCE.md — source history and licensing review

Contributing and security

See CONTRIBUTING.md. Report vulnerabilities through the private process described in SECURITY.md.