API Reference

Factory

Create a version-typed MQTT client.

Returns MQTTClientV311 when version="3.1.1", MQTTClientV5 when version="5.0". The concrete type is always MQTTClient; the return type is a Protocol view.

Client protocols

Bases: Protocol

Type-safe view of MQTTClient for MQTT 5.0 connections.

Core classes

Asyncio MQTT client with automatic reconnection.

Use as an async context manager. subscribe() returns a Subscription that is itself an async context manager.

`connection_info` `property`

Current successful handshake.

Subscription restoration may still be in progress. Previously returned snapshots remain valid after disconnection. Effective values describe negotiation; they do not change ping scheduling or future CONNECT IDs.

Raises:

Type	Description
`MQTTDisconnectedError`	If no successful connection is active, including before connecting and during reconnection.

`aenter()` `async`

Connect to the broker and start the background run loop.

`aexit(*exc)` `async`

Disconnect cleanly and cancel the run loop.

`init(host, port=1883, *, client_id='', keepalive=60, clean_session=True, username=None, password=None, will=None, tls=None, reconnect=None, on_connection_recovery_failed=None, mqtt_connect_timeout=30.0, transport_factory=None, version='3.1.1', session_expiry_interval=0, stripped_prefixes=_DEFAULT_STRIPPED_PREFIXES, max_pending_requests=1000, session_replay_buffer_size=1000, session_replay_timeout=30.0)`

Create an MQTT client.

The client must be used as an async context manager to establish the connection::

async with MQTTClient("broker.example.com") as client:
    await client.publish("sensors/temp", "22.5")

Prefer :func:create_client for version-typed access.

Parameters:

Name	Type	Description	Default
`host`	`str`	Broker hostname or IP address.	required
`port`	`int`	TCP port. Defaults to `1883` (`8883` is conventional for TLS).	`1883`
`client_id`	`str`	Client identifier sent in CONNECT. An empty string lets the broker assign one.	`''`
`keepalive`	`int`	Keepalive interval in seconds. `0` disables keepalive.	`60`
`clean_session`	`bool`	Start with a clean session (MQTT 3.1.1) or discard any existing session state on connect.	`True`
`username`	`str \| None`	Optional username for broker authentication.	`None`
`password`	`str \| None`	Optional plain-text password for broker authentication.	`None`
`will`	`Will \| None`	Last Will published by the broker if the connection closes unexpectedly.	`None`
`tls`	`SSLContext \| bool \| None`	TLS configuration. Pass `True` for default TLS, an :class:`ssl.SSLContext` for custom settings, or `False` (default) for a plain TCP connection.	`None`
`reconnect`	`ReconnectConfig \| None`	Reconnection policy. Defaults to :class:`ReconnectConfig` with exponential back-off enabled.	`None`
`on_connection_recovery_failed`	`Callable[[], Awaitable[None]] \| None`	Async callback invoked once when a connection that was established successfully cannot be restored. Initial connection failures and clean disconnects do not invoke it.	`None`
`mqtt_connect_timeout`	`float`	Seconds to wait for the broker's CONNACK during the MQTT CONNECT/CONNACK handshake — distinct from the TCP socket connect — before raising :exc:`MQTTTimeoutError`. Must be positive. Defaults to `30.0`.	`30.0`
`transport_factory`	`TransportFactory \| None`	Override the low-level transport. Useful for testing.	`None`
`version`	`Literal['3.1.1', '5.0']`	MQTT protocol version to use. Either `"3.1.1"` (default) or `"5.0"`.	`'3.1.1'`
`session_expiry_interval`	`int`	MQTT 5.0 session expiry interval in seconds. `0` means the session expires on disconnect.	`0`
`stripped_prefixes`	`tuple[str, ...]`	Group-less subscription prefixes the broker strips before delivery — matched against incoming PUBLISH topics with the prefix removed. Defaults to `("$queue", "$exclusive")`; add a broker-specific decorator here instead of patching the library. `$share/<group>/` is always handled; real namespaces the broker delivers on unchanged (e.g. `$SYS`) must not be listed.	`_DEFAULT_STRIPPED_PREFIXES`
`max_pending_requests`	`int`	Maximum concurrent MQTT 5.0 `request()` calls. Additional calls wait until capacity is available.	`1000`
`session_replay_buffer_size`	`int`	Maximum unmatched messages held while a resumed persistent session waits for local subscriptions. `0` makes the buffer unbounded. Defaults to `1000`.	`1000`
`session_replay_timeout`	`float`	Seconds a persistent-session message may remain in the replay buffer. Remaining messages are dropped without acknowledgement after the timeout. Defaults to `30.0`.	`30.0`

`auth(method, data=None)` `async`

Send an AUTH packet for enhanced authentication (MQTT 5.0 only).

Parameters:

Name	Type	Description	Default
`method`	`str`	Authentication method name negotiated with the broker.	required
`data`	`bytes \| None`	Optional authentication data to include in the packet.	`None`

Raises:

Type	Description
`RuntimeError`	If the client is not using MQTT 5.0.
`MQTTDisconnectedError`	If the client is not currently connected.

`connect()` `async`

Connect to the broker and start the background run loop.

Equivalent to entering the async context manager. Must be paired with a corresponding :meth:disconnect call to send DISCONNECT, close the socket, and stop the background run loop.

Example::

client = create_client("broker.example.com")
await client.connect()
# ... use the client
await client.disconnect()

Raises:

Type	Description
`MQTTConnectError`	If the broker refuses the connection.

`disconnect()` `async`

Disconnect cleanly and stop the background run loop.

Equivalent to exiting the async context manager. Sends DISCONNECT, closes the socket, and cancels the internal run loop task. Safe to call even if the connection has already been lost.

`ping(timeout=10.0)` `async`

Send a PINGREQ and return the round-trip time in seconds.

Parameters:

Name	Type	Description	Default
`timeout`	`float`	Seconds to wait for PINGRESP before raising :exc:`MQTTTimeoutError`.	`10.0`

Returns:

Type	Description
`float`	RTT in seconds.

Raises:

Type	Description
`MQTTDisconnectedError`	If the client is not currently connected.
`MQTTTimeoutError`	If no PINGRESP is received within timeout seconds.

`publish(topic, payload, *, qos=QoS.AT_MOST_ONCE, retain=False, properties=None)` `async`

Publish a message to topic.

Parameters:

Name	Type	Description	Default
`topic`	`str`	Topic string. Must not contain wildcards.	required
`payload`	`bytes \| str`	Message body. `str` values are UTF-8 encoded automatically.	required
`qos`	`QoS`	Delivery guarantee level. Defaults to `AT_MOST_ONCE`.	`AT_MOST_ONCE`
`retain`	`bool`	Ask the broker to retain the message for future subscribers.	`False`
`properties`	`PublishProperties \| None`	MQTT 5.0 publish properties. Raises if used with MQTT 3.1.1.	`None`

Raises:

Type	Description
`MQTTInvalidTopicError`	If topic is empty, contains wildcards, or has `$` in a non-leading position.
`MQTTDisconnectedError`	If the client is not currently connected.
`RuntimeError`	If properties is supplied on an MQTT 3.1.1 connection.
`MQTTPublishError`	If the broker rejects a QoS 1/2 publish. Not raised for QoS 0 or MQTT 3.1.1.

`request(topic, payload, *, qos=QoS.AT_MOST_ONCE, timeout=30.0, properties=None)` `async`

Send a request and wait for exactly one reply (MQTT 5.0 only).

Publishes payload to topic with a response_topic property, then waits for a message whose correlation_data matches the request. Other messages on the response topic remain available to regular subscriptions and do not complete this request.

Both response_topic and correlation_data are taken from properties when set; otherwise they are generated automatically (a unique _zmqtt/reply/<32 hex chars> topic and 16 random bytes respectively).

Parameters:

Name	Type	Description	Default
`topic`	`str`	Request topic. Must not contain wildcards.	required
`payload`	`bytes \| str`	Request body. `str` values are UTF-8 encoded automatically.	required
`qos`	`QoS`	QoS for the outgoing request publish.	`AT_MOST_ONCE`
`timeout`	`float`	Seconds to wait for the reply before raising `asyncio.TimeoutError`.	`30.0`
`properties`	`PublishProperties \| None`	Publish properties for the request. `response_topic` selects the reply topic (must not contain wildcards [MQTT-3.3.2-14]). `correlation_data` is forwarded as-is to the responder.	`None`

Returns:

Type	Description
`Message`	The matching response `Message`.

Raises:

Type	Description
`RuntimeError`	If the client is not using MQTT 5.0.
`MQTTInvalidTopicError`	If `properties.response_topic` contains wildcards.
`MQTTDisconnectedError`	If the request cannot start because the client is disconnected, or the client is stopped while waiting.
`ValueError`	If the same response topic and correlation data are already used by another active request.
`TimeoutError`	If no matching reply arrives within timeout seconds.

`subscribe(*filters, qos=QoS.AT_MOST_ONCE, auto_ack=True, receive_buffer_size=1000, no_local=False, retain_as_published=False, retain_handling=RetainHandling.SEND_ON_SUBSCRIBE, subscription_identifier=None)`

Create a :class:Subscription for one or more topic filters.

The returned object must be used as an async context manager to activate the subscription and unsubscribe on exit::

async with client.subscribe("sensors/#", qos=QoS.AT_LEAST_ONCE) as sub:
    async for msg in sub:
        print(msg.topic, msg.payload)

Parameters:

Name	Type	Description	Default
`*filters`	`str`	One or more MQTT topic filters. Wildcards `+` (single level) and `#` (multi-level) are supported.	`()`
`qos`	`QoS`	Maximum QoS level requested from the broker.	`AT_MOST_ONCE`
`auto_ack`	`bool`	Automatically send PUBACK/PUBREC upon receipt. Set to `False` to acknowledge manually via :meth:`Message.ack`.	`True`
`receive_buffer_size`	`int`	Maximum messages buffered per internal queue. When the buffers are full the read loop stops pulling from the socket, so a slow consumer pushes back on the broker through the TCP window instead of growing memory. `0` makes the queues unbounded; the default is `1000`.	`1000`
`no_local`	`bool`	Do not receive messages published by this client (MQTT 5.0 only).	`False`
`retain_as_published`	`bool`	Preserve the retain flag on forwarded messages (MQTT 5.0 only).	`False`
`retain_handling`	`RetainHandling`	Control when the broker sends retained messages for this subscription (MQTT 5.0 only).	`SEND_ON_SUBSCRIBE`
`subscription_identifier`	`int \| None`	Numeric identifier (1..268435455) sent in the SUBSCRIBE properties (MQTT 5.0 only). The broker echoes it on every PUBLISH this subscription causes: incoming messages are routed to the exact subscription that matched (essential when filters overlap, e.g. a `$share/...` subscription plus its plain twin), and the value is readable on `Message.properties`.	`None`

Raises:

Type	Description
`MQTTInvalidTopicError`	If any filter is empty, has `$` in a non-leading position, or contains a malformed wildcard (e.g. `sensors#` or `a/b#/c`).
`RuntimeError`	If an MQTT 5.0-only subscription option is used on an MQTT 3.1.1 connection.

Async context manager for an active topic subscription.

Registers filters on enter, unsubscribes on exit. Messages are available via get_message() or async iteration. Survives reconnection transparently — the queue keeps buffering and delivery resumes when the connection restores.

`start()` `async`

Register the subscription filters with the broker. Equivalent to entering the async context manager. Must be paired with a corresponding :meth:stop call to send UNSUBSCRIBE and release internal resources.

Example::

sub = client.subscribe("sensors/#", qos=QoS.AT_LEAST_ONCE)
await sub.start()
# ... later
await sub.stop()

Raises:

Type	Description
`MQTTDisconnectedError`	If the client is not currently connected.

`stop()` `async`

Unsubscribe from all filters and stop message delivery.

Equivalent to exiting the async context manager. Sends UNSUBSCRIBE to the broker. Safe to call even if the connection has already been lost — the UNSUBSCRIBE is silently skipped in that case.

Example::

await sub.stop()

`get_message()` `async`

Wait for and return the next message from the subscription queue.

Raises:

Type	Description
`Exception`	The terminal error that stopped the client run loop.

`aenter()` `async`

Register the subscription filters with the broker.

Raises:

Type	Description
`MQTTDisconnectedError`	If the client is not currently connected.

`aexit(*exc)` `async`

Unsubscribe from all filters and stop message delivery.

`aiter()`

Return self as the async iterator.

`anext()` `async`

Return the next message, suspending until one is available.

Incoming MQTT message as delivered to application code.

Attributes:

Name	Type	Description
`topic`	`str`	The topic on which the message was published.
`payload`	`bytes`	Raw message body as bytes.
`qos`	`QoS`	QoS level at which the message was delivered.
`retain`	`bool`	`True` if the broker sent this as a retained message.
`properties`	`PublishProperties \| None`	MQTT 5.0 publish properties, or `None` for MQTT 3.1.1 connections.

`topic` `instance-attribute`

`payload` `instance-attribute`

`qos` `instance-attribute`

`retain` `instance-attribute`

`properties = None` `class-attribute` `instance-attribute`

`ack()` `async`

Send the protocol-level ack for this message. Idempotent; no-op when auto_ack=True.

Immutable snapshot of a successful CONNECT/CONNACK handshake.

Attributes:

Name	Type	Description
`connection_id`	`int`	Successful network connection number within this client, starting at 1. A resumed MQTT session still gets a new number.
`session_present`	`bool`	Whether the broker resumed an existing MQTT session.
`return_code`	`int`	Successful CONNACK return code (0).
`properties`	`ConnAckProperties \| None`	Raw CONNACK properties, without substituted defaults.
`effective_client_id`	`str`	Sent client ID, or the broker-assigned ID if empty.
`effective_keepalive`	`int`	Server Keep Alive, falling back to CONNECT keepalive.
`effective_session_expiry_interval`	`int \| None`	CONNACK session expiry, falling back to CONNECT and then 0. None for MQTT 3.1.1.

`from_connack(connect_packet, connack, *, version, connection_id)` `classmethod`

Build a snapshot from a successful CONNECT/CONNACK exchange.

Configuration

Configuration for automatic reconnection on connection loss.

Attributes:

Name	Type	Description
`enabled`	`bool`	Whether to reconnect automatically. Set to `False` to let exceptions propagate immediately on disconnection.
`initial_delay`	`float`	Seconds to wait before the first reconnection attempt.
`max_delay`	`float`	Upper bound (seconds) for the exponential back-off delay.
`backoff_factor`	`float`	Multiplier applied to the delay after each failed attempt.
`max_attempts`	`int \| None`	Maximum total number of connection attempts before giving up. `None` retries indefinitely.

Last Will and Testament embedded in CONNECT.

Utilities

Return whether an MQTT topic filter matches a published topic.

MQTT wildcards (+ and #), shared subscriptions ($share/<group>/...), and broker-stripped decorator prefixes are supported. By default, $queue and $exclusive are treated as stripped prefixes; pass stripped_prefixes to configure another broker.

Enumerations

Bases: IntEnum

MQTT Quality of Service delivery guarantee levels.

Attributes:

Name	Type	Description
`AT_MOST_ONCE`		Fire-and-forget. No acknowledgement, no retries (QoS 0).
`AT_LEAST_ONCE`		Acknowledged delivery. The message may arrive more than once (QoS 1).
`EXACTLY_ONCE`		Four-way handshake guarantees exactly-once delivery (QoS 2).

Bases: IntEnum

MQTT 5.0 subscription option controlling which retained messages are delivered.

Properties (MQTT 5.0)

MQTT 5.0 properties attached to a PUBLISH packet.

Attributes:

Name	Type	Description
`payload_format_indicator`	`int \| None`	`0` for unspecified binary, `1` for UTF-8 encoded character data.
`message_expiry_interval`	`int \| None`	Seconds the broker retains the message. Omit to keep indefinitely.
`topic_alias`	`int \| None`	Topic alias integer used instead of the full topic string on the wire.
`response_topic`	`str \| None`	Topic the receiver should use when replying (request/response pattern).
`correlation_data`	`bytes \| None`	Opaque bytes the sender uses to match a response to a request.
`subscription_identifier`	`int \| None`	Identifier of the subscription that caused this message to be delivered.
`content_type`	`str \| None`	MIME type describing the payload content.
`user_properties`	`tuple[tuple[str, str], ...]`	Arbitrary key-value pairs forwarded with the message.

MQTT 5.0 properties sent in the CONNECT packet.

Attributes:

Name	Type	Description
`session_expiry_interval`	`int \| None`	Seconds before the broker discards the session after disconnect. `0` means discard immediately; `0xFFFFFFFF` means never expire.
`receive_maximum`	`int \| None`	Maximum number of QoS 1/2 messages the client will process concurrently.
`maximum_packet_size`	`int \| None`	Largest packet size (bytes) the client will accept.
`topic_alias_maximum`	`int \| None`	Number of topic aliases the client supports. `0` disables topic aliases.
`request_response_information`	`bool \| None`	Ask the broker to include response information in CONNACK.
`request_problem_information`	`bool \| None`	Ask the broker to include reason strings and user properties on errors.
`authentication_method`	`str \| None`	Method name for enhanced authentication.
`authentication_data`	`bytes \| None`	Initial data for enhanced authentication.
`user_properties`	`tuple[tuple[str, str], ...]`	Arbitrary key-value pairs forwarded to the broker.

MQTT 5.0 properties sent in an AUTH packet.

Attributes:

Name	Type	Description
`authentication_method`	`str \| None`	Method name for the enhanced authentication exchange.
`authentication_data`	`bytes \| None`	Method-specific data for this step of the authentication exchange.
`reason_string`	`str \| None`	Human-readable reason included for diagnostic purposes.
`user_properties`	`tuple[tuple[str, str], ...]`	Arbitrary key-value pairs forwarded with the packet.

Exceptions

Bases: Exception

Base class for all zmqtt exceptions.

Bases: MQTTError

CONNACK returned a non-zero return code.

Bases: MQTTError

Unexpected or malformed packet received.

Bases: MQTTError

Connection lost unexpectedly.

Bases: MQTTError

An MQTT operation did not complete within the allotted time.

Bases: MQTTError

The broker rejected one or more filters in a SUBSCRIBE (SUBACK >= 0x80).

Most commonly an authorization denial: without this error the subscription looks successful and silently never receives anything.

Bases: MQTTError

The broker rejected a QoS 1/2 publish.

reason_name is the spec's name for reason_code (None for a code zmqtt does not recognize). reason_string is the broker's optional Reason String property.

Bases: MQTTError

Topic string or topic filter failed MQTT validation.

API Reference

Factory

Client protocols

Core classes

connection_info property

__aenter__() async

__aexit__(*exc) async

auth(method, data=None) async

connect() async

disconnect() async

ping(timeout=10.0) async

publish(topic, payload, *, qos=QoS.AT_MOST_ONCE, retain=False, properties=None) async

request(topic, payload, *, qos=QoS.AT_MOST_ONCE, timeout=30.0, properties=None) async

subscribe(*filters, qos=QoS.AT_MOST_ONCE, auto_ack=True, receive_buffer_size=1000, no_local=False, retain_as_published=False, retain_handling=RetainHandling.SEND_ON_SUBSCRIBE, subscription_identifier=None)

start() async

stop() async

get_message() async

__aenter__() async

__aexit__(*exc) async

__aiter__()

__anext__() async

topic instance-attribute

payload instance-attribute

qos instance-attribute

retain instance-attribute

properties = None class-attribute instance-attribute

ack() async

from_connack(connect_packet, connack, *, version, connection_id) classmethod

Configuration

Utilities

Enumerations

Properties (MQTT 5.0)

Exceptions

`connection_info` `property`

`aenter()` `async`

`aexit(*exc)` `async`

`auth(method, data=None)` `async`

`connect()` `async`

`disconnect()` `async`

`ping(timeout=10.0)` `async`

`publish(topic, payload, *, qos=QoS.AT_MOST_ONCE, retain=False, properties=None)` `async`

`request(topic, payload, *, qos=QoS.AT_MOST_ONCE, timeout=30.0, properties=None)` `async`

`subscribe(*filters, qos=QoS.AT_MOST_ONCE, auto_ack=True, receive_buffer_size=1000, no_local=False, retain_as_published=False, retain_handling=RetainHandling.SEND_ON_SUBSCRIBE, subscription_identifier=None)`

`start()` `async`

`stop()` `async`

`get_message()` `async`

`aenter()` `async`

`aexit(*exc)` `async`

`aiter()`

`anext()` `async`

`topic` `instance-attribute`

`payload` `instance-attribute`

`qos` `instance-attribute`

`retain` `instance-attribute`

`properties = None` `class-attribute` `instance-attribute`

`ack()` `async`

`from_connack(connect_packet, connack, *, version, connection_id)` `classmethod`