ebus-mqtt-client

Standalone MQTT client wrapper around paho-mqtt.

Features

• TLS support (secure with CA verification, insecure, or plaintext)
• Resilient connect: construction never blocks or raises on a down broker (connect_async); the connection is established and retried on the network loop started by start()
• QoS 0 retained publishes issued before the link is up are held (newest value per topic, bounded) and flushed on connect, rather than being discarded by paho and lost
• Automatic reconnection with configurable backoff
• Subscription recovery on reconnect
• Topic pattern matching via paho's MQTTMatcher
• Last Will and Testament (LWT)
• MQTTv3 and MQTTv5 protocol support
• Factory method for dict-based configuration
• Bounded, broker-independent shutdown: stop(timeout=...) returns promptly even against a dead broker
• Bounded publish flush: publish_and_flush(...) lands a final message before a clean disconnect, no fixed sleep
• Optional loop-native driving: AsyncioMqttDriver runs the network loop on your asyncio event loop instead of a background thread, for hosts that already own a loop

Install

pip install ebus-mqtt-client

Quick start

from ebus_mqtt_client import MqttClient

client = MqttClient(
    client_id="my-client",
    endpoint="broker.example.com",
    port=1883,
)
client.start()

client.subscribe("sensors/#", callback_param)
client.publish("sensors/temp", "22.5")

client.stop()

Graceful shutdown

Publish a final retained message and flush it (bounded) before disconnecting, then stop within a time bound even when the broker is unreachable:

# Land a final state update, waiting up to 1s for it to actually be sent.
# Returns True on flush; False (without blocking or raising) if not connected,
# the publish fails, or the flush exceeds the timeout.
client.publish_and_flush(
    "devices/my-client/state", "disconnected", retain=True, timeout=1.0
)

# Returns within ~timeout seconds even if the broker is gone.
client.stop(timeout=2.0)

publish() also returns paho's MQTTMessageInfo (or None if there is no client), so you can wait for a single message yourself: client.publish(topic, data).wait_for_publish(1.0).

Publishing before the connection is up

Construction registers the broker with connect_async and returns; CONNACK does not arrive until the network loop started by start() receives it, so paho refuses anything published in between with MQTT_ERR_NO_CONN. That is an expected transient of startup, not a fault, so it no longer logs a warning. What becomes of the message depends on its QoS, because paho only discards some of them:

• QoS 1 and 2 (publish() defaults to QoS 1): paho stores the message in its own out-queue before returning MQTT_ERR_NO_CONN, keeps it across reconnects, and re-sends it itself once CONNACK arrives. Nothing is held here, because a second copy would put every such message on the wire twice per connect.
• QoS 0, retained: paho keeps nothing, so the message would be lost. It is held here and flushed on connect, before the subscription recovery and before your on_connect_callback. The hold keeps the newest value per topic rather than replaying every attempt, because retained state is last-value-wins: a queue that replayed attempts in order could write a stale value on top of a newer one published after the connection came up. It is bounded by client.pending_limit (default 512 topics, evicting the oldest) so a client that never connects cannot grow without limit.
• QoS 0, not retained: dropped, with a debug-level line. These are events, and delivering one after an arbitrary delay announces something that was true once.

Any other publish failure still logs a warning, with the paho result code. A publish refused this way still returns paho's MQTTMessageInfo with rc == MQTT_ERR_NO_CONN whichever branch it took, so a caller that inspects the result sees exactly what paho said. stop() discards anything still held.

Two things this does not change, both of them paho's behavior rather than this wrapper's. paho's own out-queue is unbounded unless you set client.mqttc.max_queued_messages_set(...), so pending_limit bounds only the QoS 0 hold. And paho replays that queue after on_connect returns, in the order the publishes were attempted, so at QoS 1 and 2 a value revised while disconnected is re-sent stale-first: if you republish inside on_connect_callback, paho's older copy can land after yours. Publish state you care about at QoS 0, or republish after the connection is established.

publish_and_flush() is unaffected: it checks is_connected() and returns False immediately rather than holding, since its whole purpose is to confirm a message reached the broker now.

From a config dict

cfg = {
    "host": "broker.example.com",
    "port": 8883,
    "use_tls": True,
    "tls_insecure": False,
    "tls_ca_cert": "/path/to/ca.pem",
    "authentication": {
        "type": "USER_PASS",
        "username": "user",
        "password": "secret",
    },
}

client = MqttClient.from_config(cfg, client_id="my-client")
client.start()

mTLS (client-certificate authentication)

When the broker authenticates the client via the TLS handshake (no username/password), supply a client cert and key. File-path form:

cfg = {
    "host": "broker.example.com",
    "port": 8883,
    "use_tls": True,
    "tls_insecure": False,
    "tls_ca_cert": "/path/to/ca.pem",
    "tls_client_cert": "/path/to/client.crt",
    "tls_client_key": "/path/to/client.key",
    # "tls_client_key_password": "...",  # only if the key is encrypted
}

client = MqttClient.from_config(cfg, client_id="my-client")
client.start()

In-memory form — useful when the cert/key are fetched from a secret store rather than the filesystem. If both the path and *_data forms are supplied for the same item, the *_data form wins and a warning is logged:

cfg = {
    "host": "broker.example.com",
    "port": 8883,
    "use_tls": True,
    "tls_insecure": False,
    "tls_ca_data": ca_pem_str,
    "tls_client_cert_data": client_cert_pem_str,
    "tls_client_key_data": client_key_pem_str,
}

client = MqttClient.from_config(cfg, client_id="my-client")
client.start()

Loop-native driving (asyncio)

By default start() runs paho's network loop on a background thread. If your program already owns an asyncio event loop (for example a Home Assistant integration), you can drive the same client on that loop with no extra thread, via the optional AsyncioMqttDriver:

import asyncio
from ebus_mqtt_client import AsyncioMqttDriver, MqttClient

async def main():
    client = MqttClient.from_config(cfg, client_id="my-client")
    driver = client.asyncio_driver()      # or: AsyncioMqttDriver(client, loop=my_loop)
    await driver.start()                   # instead of client.start()
    client.subscribe("sensors/#", callback_param)
    # ... all MQTT I/O now runs on this event loop ...
    await driver.stop()

asyncio.run(main())

Thread mode (client.start()) and the driver are mutually exclusive per client: pick one. The driver module is imported lazily (only when you reference AsyncioMqttDriver or call asyncio_driver()), so a thread-only consumer never loads the asyncio machinery.

If you inject the client into ebus_sdk.Controller(mqttc=client) as a bring-your-own transport, wire Controller.resync onto the on-connect callback (client.on_connect_callback = controller.resync) so the retained tree re-walks after a reconnect; the SDK does that automatically only for a client it creates itself.

Releasing

The version lives in exactly one place: __version__ in src/ebus_mqtt_client/__init__.py. pyproject.toml reads it dynamically, the setup.py legacy shim reads it by regex, and the publish workflow refuses to release a tag that disagrees with it. To cut a release:

1. Bump __version__ in src/ebus_mqtt_client/__init__.py (the only place).
2. Move the CHANGELOG's [Unreleased] entries under a new version heading.
3. Commit: git commit -am "Release X.Y.Z".
4. Tag it to match, v-prefixed: git tag vX.Y.Z.
5. Push the tag: git push --tags (a plain git push does not trigger a release).

Pushing a v* tag runs the publish workflow, which verifies the tag equals v$__version__ (a mismatch fails before anything is built), builds the sdist and wheel, and publishes to PyPI via Trusted Publishing (OIDC, no stored token).

Contributing

See CONTRIBUTING.md for how to file Discussions, Issues, and pull requests. The library is intentionally a thin MQTT-only layer — Homie / eBus features belong in ebus-sdk.

License

MIT License — Copyright (c) 2026 Clark Communications Corporation