Python

Async (asyncio) bindings generated by UniFFI's built-in Python backend and packaged by maturin. Client and broker are separate wheels. Read the shared concepts first for the option records and semantics.

Install

pip install stem-mqtt-client stem-mqtt-broker    # either one on its own works too

The distribution names are stem-mqtt-client / stem-mqtt-broker (mqtt-client is taken on PyPI), the import names are not:

import mqtt_client
import mqtt_broker

Requirements

• Python 3.9+ and asyncio. The wheels are py3-none-<platform> (the library is loaded with ctypes), so there is one wheel per OS/arch, not per Python version.
• Wheels bundle the native library: Linux (manylinux_2_28, x86_64) from the Gitea release; the GitHub release adds macOS and Windows. Other platforms build from source (below).

Quick start

import asyncio
import mqtt_broker
import mqtt_client


def options(port, client_id, username=None):
    return mqtt_client.ConnectOptions(
        host="127.0.0.1", port=port, client_id=client_id,
        version=mqtt_client.MqttVersion.V5, clean_start=True, keep_alive_secs=30,
        username=username, password=None, will=None,
        connect_timeout_secs=10, operation_timeout_secs=5,
        auto_reconnect=False, reconnect_backoff_secs=0, reconnect_max_backoff_secs=0, tls=None,
    )


class Inbox(mqtt_client.MqttMessageListener):
    def __init__(self):
        self.queue = asyncio.Queue()
        self.loop = asyncio.get_running_loop()

    def on_message(self, message):          # called from a Rust thread
        self.loop.call_soon_threadsafe(self.queue.put_nowait, message)

    def on_disconnected(self, reason):
        print("disconnected:", reason)


async def main():
    broker = mqtt_broker.MqttBroker(mqtt_broker.MqttBrokerConfig(
        bind_address="127.0.0.1", port=0, ws_port=None, allow_anonymous=True, max_clients=0,
        max_qos=mqtt_broker.QoS.EXACTLY_ONCE, max_retained_messages=100,
        max_queued_per_client=100, redelivery_interval_secs=0, tls=None,
    ))
    await broker.start()
    port = broker.bound_port()

    inbox = Inbox()
    sub = mqtt_client.MqttClient(options(port, "py-sub"))
    sub.set_message_listener(inbox)
    await sub.connect()
    await sub.subscribe("demo/#", mqtt_client.QoS.AT_LEAST_ONCE)

    pub = mqtt_client.MqttClient(options(port, "py-pub"))
    await pub.connect()
    await pub.publish("demo/hello", b"hello", mqtt_client.QoS.AT_LEAST_ONCE, False)

    message = await asyncio.wait_for(inbox.queue.get(), 5)
    print(message.topic, bytes(message.payload))

    await pub.disconnect(); await sub.disconnect(); await broker.stop()


asyncio.run(main())

max_packet_size, max_outbound_queue, session_expiry_secs, auth_method and auth_data have defaults and can be omitted, as above.

Client

• Every I/O method is a coroutine: await client.connect(), subscribe(filter, qos), unsubscribe(filter), publish(topic, payload, qos, retain), disconnect(). is_connected(), set_message_listener(), enable_message_queue() and last_disconnect_reason() are plain calls. payload is bytes; received message.payload is bytes-like — wrap it in bytes(...).
• Enums are UPPER_SNAKE: mqtt_client.QoS.AT_MOST_ONCE | AT_LEAST_ONCE | EXACTLY_ONCE, mqtt_client.MqttVersion.V311 | V5.
• Listener callbacks run on a Rust thread, not on the event loop. Hand work back with loop.call_soon_threadsafe(...) as above (or a threading.Event). Keep them short: an exception raised in a callback is caught and logged so it cannot take down the I/O task, but it is not propagated to you.
• Or pull: client.enable_message_queue(16) then await client.next_message(5000) returns a message or None — no threads involved.

polled = mqtt_client.MqttClient(options(port, "py-polled"))
polled.enable_message_queue(16)
await polled.connect()
await polled.subscribe("demo/#", mqtt_client.QoS.AT_LEAST_ONCE)
msg = await polled.next_message(5000)       # None on timeout

Broker

mqtt_broker.MqttBroker(config) — await start(), await stop(), is_running(), bound_port(), bound_ws_port(), bound_tls_port(), client_count(). Use port 0 and read bound_port() for an ephemeral port. The broker owns its own mqtt_broker.QoS; don't pass a mqtt_client.QoS to it.

class Events(mqtt_broker.MqttBrokerEventListener):
    def on_client_connected(self, client_id): print("+", client_id)
    def on_client_disconnected(self, client_id, reason): print("-", client_id, reason)
    def on_message_published(self, client_id, topic, qos): print(client_id, "->", topic)

broker.set_event_listener(Events())

Or poll: broker.enable_event_queue(64) then await broker.next_event(1000) returns a BrokerEvent (ClientConnected, ClientDisconnected, MessagePublished) or None.

Authentication

class Auth(mqtt_broker.MqttAuthProvider):
    def authenticate(self, client_id, username, password):   # password: bytes | None
        return username == "demo" and password == b"demo"

broker.set_auth_provider(Auth())

A refused client raises from await client.connect() (reason 0x86). MQTT 5 enhanced authentication:

class AddOne(mqtt_broker.MqttEnhancedAuthProvider):
    def step(self, client_id, method, data, round):
        if round == 0:
            return mqtt_broker.EnhancedAuthStep(outcome=mqtt_broker.EnhancedAuthOutcome.CONTINUE, data=b"\x07")
        ok = data is not None and bytes(data) == b"\x08"
        return mqtt_broker.EnhancedAuthStep(
            outcome=mqtt_broker.EnhancedAuthOutcome.SUCCESS if ok else mqtt_broker.EnhancedAuthOutcome.FAILURE,
            data=b"welcome" if ok else b"")

class AddOneHandler(mqtt_client.MqttAuthHandler):
    def respond(self, method, challenge):
        return bytes(b + 1 for b in challenge)

broker.set_enhanced_auth_provider(AddOne())
client = mqtt_client.MqttClient(options(port, "py-eauth"))   # with auth_method="X-ADD-ONE", auth_data=b"hello"
client.set_auth_handler(AddOneHandler())

TLS

ca = open("ca.pem", "rb").read()
tls = mqtt_client.TlsOptions(ca_cert_pem=ca, client_cert_pem=None, client_key_pem=None,
                             insecure_skip_certificate_verification=False)
# pass tls=tls in ConnectOptions

broker_tls = mqtt_broker.BrokerTlsConfig(port=8883, cert_pem=open("server.pem", "rb").read(),
                                         key_pem=open("server.key", "rb").read(), client_ca_pem=None)
# pass tls=broker_tls in MqttBrokerConfig

Language notes

• Optional values are None; byte values are bytes. Errors are exceptions (mqtt_client.MqttError.*, mqtt_broker.MqttBrokerError.*); catch Exception for "refused".
• A broker started in a script keeps running until await broker.stop(); closing the loop without stopping leaves the listener threads alive until the process exits.

Building from source

pip install maturin
cd crates/mqtt-client && maturin build --release      # wheel in target/wheels/
cd ../mqtt-broker && maturin build --release
pip install ../../target/wheels/*.whl

maturin runs uniffi-bindgen itself; there is no separate generation step. More: packaging/python/README.md. The end-to-end test is tests/bindings/python/smoke.py.

Troubleshooting

• ImportError: cannot import name 'mqtt_client' — you installed the broker wheel only (or vice versa); install both, or import just the component you have.
• ModuleNotFoundError after pip install mqtt-client — wrong package; the name is stem-mqtt-client.
• A callback never fires — it is invoked from another thread; make sure the loop you hand work to is still running and that the listener object is kept referenced (don't let it be garbage collected).
• no matching distribution — no wheel for your platform yet; build from source.