pron. "ohm cue tee tee" or "omega cutie"
A reliable and persistent MQTT 5.0 client library for Python.
ΩQTT supports publish and subscribing all QoS levels with optional persistence to disk for QoS >0. When not persisting to disk, a fast (but volatile) in memory store is used. Either way, publishing a QoS >0 message returns a handle with a method to wait for the message to be fully acknowledged by the broker.
Set an alias policy when publishing a message and a topic alias will be generated, if allowed by the broker. You can also force QoS 0 messages to require a topic alias, to avoid silently bleeding bandwidth.
Access all optional properties of all MQTT control packet types. If you ever wanted to check the user properties of your packets, welcome home.
ΩQTT is tested on Linux, Windows and MacOS with CPython versions 3.10-3.13. It should work on any platform that CPython runs on.
ΩQTT has been implemented to a high standard of test coverage and static analysis, from the beginning. It continues to improve.
ΩQTT is published to the Python Package Index.
python3 -m pip install ohmqtt
See the examples/ directory for examples of using ΩQTT.
With uv installed, run them like e.g.
uv run python3 -m examples.publish
By default, all of the examples try to connect to an MQTT broker at localhost:1883.
Supply an alternate --address argument if you want to connect to a broker at a different location.
If you want to run a local broker in Docker, try https://hub.docker.com/_/eclipse-mosquitto
Besides compressing your payloads, you can reduce MQTT publish overhead by using topic aliases.
Instruct the broker that it can send topic aliases while connecting:
from ohmqtt import Client, MQTTConnectProps
# The maximum value for TopicAliasMaximum is USHRT_MAX (65535).
# It must be >0 for the broker to use topic aliases when sending messages to the client.
connect_props = MQTTConnectProps(TopicAliasMaximum=0xffff)
with Client() as client:
client.connect(address, connect_properties=connect_props)Specify that topic aliases should be used when publishing messages:
from ohmqtt import AliasPolicy
client.publish("some/topic", b"the payload", alias_policy=AliasPolicy.TRY)This will automatically assign topic aliases to topics up to the maximum topic alias ID reported by the broker.
Higher QoS levels will use more overall bandwidth. You may force a maximum QoS level when subscribing to a topic:
client.subscribe("some/topic", callback, max_qos=0)You can wait for QoS>0 messages to be acknowledged and access the acknowledgement packets. For QoS 1 this will be PUBACK, for QoS 2 it may be PUBREC (in case of an error) or PUBCOMP.
from ohmqtt import LostMessageError, MQTTError, MQTTQoS
handle = client.publish("topic", b"payload", qos=MQTTQoS.Q1)
try:
ack = handle.wait_for_ack(timeout=5.0)
print(ack)
except TimeoutError: # Was not acknowledged in time
...
except LostMessageError: # The message was dropped upon new session from the broker
...
except MQTTError as exc: # The acknowledgement had an error code
print(f"Error from server while publishing: {exc.reason_code}")You can persist QoS>0 publish state to disk to guarantee delivery beyond the lifetime of your application.
ΩQTT will follow the MQTT specification, meaning that when you connect to a broker and it does not recognize your session, the persistent state will be cleared.
Specify the path to the database when creating the Client (a new database will be created if it does not exist).
Also specify a client_id and session expiry interval when connecting to the broker:
from ohmqtt import Client, MQTTConnectProps
# Set SessionExpiryInterval=UINT_MAX to indicate that the session should never expire.
# Otherwise the interval is in seconds and must be >0 to persist the session.
connect_props = MQTTConnectProps(SessionExpiryInterval=0xffffffff)
with Client(db_path="/path/to/ohmqtt.db") as client:
client.connect(address, client_id="my_client_id", connect_properties=connect_props)By default the database will operate in a very safe mode.
Calls to publish with QoS>0 will not return, and data will not be sent over the wire, until the message is fully committed to disk.
You can specify to use a faster, less synchronous configuration which may be good enough for your use case.
The implementation will use SQLite WAL. Set db_fast=True when constructing the client like so:
from ohmqtt import Client, MQTTConnectProps
connect_props = MQTTConnectProps(SessionExpiryInterval=0xffffffff)
with Client(db_path="/path/to/ohmqtt.db", db_fast=True) as client:
client.connect(address, client_id="my_client_id", connect_properties=connect_props)- Review error handling and validation
- AUTH connection state
- Session data reset policies
This project uses nox and uv to run the tests against all supported Python versions.
To do all of this in a venv:
python3 -m venv .venv
source .venv/bin/activate
pip install nox uv
noxCreate a release on GitHub to publish to PyPI.
The docs are automatically built on readthedocs upon release.
To manually build the docs: uv run make clean && uv run make html
Please submit issues and pull requests via GitHub.