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: beta (0.2.0b3). The native Stable API tier follows the documented compatibility policy; Provisional extension surfaces may still evolve 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;
• immutable runtime statistics for queues, budgets, receipts and transports;
• manual acknowledgement;
• in-memory and SQLite inflight persistence;
• aggregate publish_many() with bounded memory and measured throughput gains;
• synchronous loop-bound publish_nowait() for non-suspending native producers;
• 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.

API stability

The native client is imported from mqttium.api; protocol enums and operational exceptions are imported from mqttium. Advanced protocol, persistence, transport, packet and Paho surfaces are classified separately as Provisional. See docs/API-STABILITY.md for the exact Stable, Provisional and Internal boundaries.

Installation

Install the current beta from PyPI:

python -m pip install --pre 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())

Non-suspending publishing

A producer already executing on the client's event loop can submit without creating or awaiting a coroutine:

receipt = client.publish_nowait("telemetry/device-1", payload, qos=1)
# Continue synchronously, then observe completion later if needed.
await receipt.wait()

publish_nowait() either admits the publication immediately or raises FlowControlError; it never waits for engine or writer capacity. It returns the normal PublishReceipt, so QoS 1/2 completion is observed in the same way as with publish().

The method follows the same ownership rule as asyncio.Queue.put_nowait(): it is intended for the client's owning event-loop thread, not as a generic thread-safe API. Cross-thread synchronous callers should use an adapter such as mqttium.compat.paho.Client, which coalesces submissions before handing a bounded batch to the loop.

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_inbound_bytes=64 * 1024**2,   # persisted inbound QoS handshakes
    max_pending_delivery_bytes=64 * 1024**2,  # inbound messages awaiting a consumer
    max_ingress_batch_bytes=1 * 1024**2,      # decoded work before delivery is drained
    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.

The two inbound byte limits cover different lifetimes: max_pending_inbound_bytes bounds QoS 2 and manually acknowledged QoS 1 data retained by the session store, while max_pending_delivery_bytes bounds data waiting for callback or iterator consumers. MQTT 5 reports a persisted-session quota violation with reason 0x97; MQTT 3.1.1 closes the connection.

The writer also has an independent max_outbound_bytes=1 MiB default. publish_nowait() refuses immediately when that wire queue is full, so callers publishing large payloads should size this byte limit explicitly rather than assuming max_outbound_messages is the only binding constraint.

Native QoS 0 uses its direct writer fast path only while on_publish is None. Installing that callback deliberately routes publications through the standard engine and EffectPump so completion callbacks retain their ordering semantics.

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.

Runtime statistics

stats() returns a frozen snapshot without starting a sampler or emitting logs:

snapshot = client.stats()
print(snapshot.outbound.pending_bytes)
print(snapshot.inbound.inflight)
print(snapshot.writer.queued_bytes)
print(snapshot.delivery.pending_bytes)

Each section is produced by the component that owns the state — the two protocol sessions, the effect and write pumps, the transport — and stats() only assembles them. The snapshot also includes lifetime high-water marks, batching decision counters, task state, receipt counts, decoder buffering and WebSocket/stream transport buffers. It is intended to be called on the client's owning event loop. See docs/API-STABILITY.md.

Validation

The release gates include:

• more than 500 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.

A separate finalisation workflow runs short reconnect/backpressure soaks on Linux for relevant pull requests. Extended Linux/macOS soaks and interoperability campaigns against multiple brokers are available by manual dispatch. Their acceptance criteria are documented in docs/STABILITY.md.

Documentation

• docs/DESIGN.md — architecture and invariants
• docs/API-STABILITY.md — public API policy and deprecations
• docs/STABILITY.md — soak and interoperability campaign
• docs/RELEASING.md — rehearsal, publication and failure handling
• 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.