MQTT 5 in Python with gmqtt

gmqtt is an asyncio MQTT client for Python that was designed around MQTT 5 from the start. Every MQTT 5 property — session expiry, message expiry, user properties, response topic, correlation data, subscription identifiers — is a plain keyword argument, and incoming properties arrive as a dictionary. If you want to experiment with MQTT 5 features from Python without building Properties objects by hand, gmqtt is the shortest path.

This tutorial walks through connecting with MQTT 5, publishing with properties, subscription options, request/response, reading reason codes and testing everything against a real broker.

Install and connect

pip install gmqtt

gmqtt uses callbacks that run on the asyncio event loop. Connect, subscribe in on_connect, and keep the program alive with an asyncio.Event:

import asyncio
import os
from gmqtt import Client as MQTTClient
from gmqtt.mqtt.constants import MQTTv50

STOP = asyncio.Event()

def on_connect(client, flags, rc, properties):
    print("connected, rc =", rc, "broker properties:", properties)
    client.subscribe("testmqtt/gmqtt/#", qos=1)

def on_message(client, topic, payload, qos, properties):
    print("message", topic, payload.decode(), properties)
    return 0  # reason code for the PUBACK

def on_disconnect(client, packet, exc=None):
    print("disconnected", exc)

async def main():
    client = MQTTClient(
        "testmqtt-gmqtt-demo-01",
        session_expiry_interval=3600,   # keep the session 1 h after disconnect
        receive_maximum=100,            # at most 100 unacknowledged QoS 1/2 messages
    )
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_disconnect = on_disconnect
    client.set_auth_credentials(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])

    await client.connect("broker.connectmq.com", 1883, keepalive=30, version=MQTTv50)
    await STOP.wait()
    await client.disconnect()

asyncio.run(main())

Two things are specific to gmqtt. Keyword arguments passed to the MQTTClient constructor become CONNECT properties — here session_expiry_interval and receive_maximum. And the properties argument of on_connect carries what the broker sent back in CONNACK: its own receive maximum, topic alias maximum, maximum packet size or an assigned client identifier, depending on the broker. Printing it once is the quickest way to see what your broker actually supports.

For TLS, pass ssl=True (or an ssl.SSLContext) and port 8883: await client.connect("broker.connectmq.com", 8883, ssl=True).

Publish with MQTT 5 properties

Publish properties are keyword arguments to client.publish():

import json, time

client.publish(
    "testmqtt/gmqtt/sensor-1/temperature",
    json.dumps({"value": 21.7, "t": time.time()}),
    qos=1,
    content_type="application/json",
    message_expiry_interval=60,                     # drop it if undelivered after 60 s
    user_property=("firmware", "2.4.1"),
)

On the subscriber side, the same values show up in properties. Note the shape — values are lists, because MQTT 5 allows some properties to appear more than once:

message testmqtt/gmqtt/sensor-1/temperature {"value": 21.7, "t": 1792223400.12}
  {'content_type': ['application/json'], 'message_expiry_interval': [60],
   'user_property': [('firmware', '2.4.1')], 'dup': 0, 'retain': 0}

Read them defensively, e.g. ctype = properties.get("content_type", [None])[0]. If the message sat in a queue before delivery, the broker sends the remaining expiry time, so the subscriber may see a value lower than 60 — a handy way to verify that message expiry works. For when user properties are the right tool and when they are not, see MQTT user properties.

Subscription options

MQTT 5 adds per-subscription options. gmqtt exposes them through Subscription objects:

from gmqtt import Subscription

client.subscribe(
    [
        Subscription("testmqtt/gmqtt/+/temperature", qos=1),
        Subscription("testmqtt/gmqtt/chat/#", qos=1, no_local=True),
        Subscription("testmqtt/gmqtt/config/#", qos=1, retain_handling_options=2),
    ],
    subscription_identifier=7,
)

When a message matches the subscription above, properties["subscription_identifier"] contains [7]. That makes routing in on_message cheap and explicit, without re-matching topic strings.

Request/response with response topic and correlation data

MQTT 5 standardises request/response: the requester sets a response topic and correlation data, the responder publishes its answer there and copies the correlation data back.

def on_connect(client, flags, rc, properties):
    client.subscribe("testmqtt/gmqtt/rpc/time", qos=1)

def on_message(client, topic, payload, qos, properties):
    reply_to = properties.get("response_topic", [None])[0]
    corr = properties.get("correlation_data", [None])[0]
    if reply_to:
        client.publish(reply_to, str(time.time()), qos=1, correlation_data=corr)
    return 0

import uuid

corr = uuid.uuid4().bytes
client.subscribe("testmqtt/gmqtt/rpc/replies/client-01", qos=1)
client.publish(
    "testmqtt/gmqtt/rpc/time", "",
    qos=1,
    response_topic="testmqtt/gmqtt/rpc/replies/client-01",
    correlation_data=corr,
)

In the requester's on_message, compare properties["correlation_data"][0] with corr to match the reply to its request. The full pattern, including timeouts, is covered in MQTT request/response.

Reason codes and clean disconnects

MQTT 5 replaces the old one-byte return codes with detailed reason codes on CONNACK, SUBACK, PUBACK and DISCONNECT. In gmqtt you see them in a few places:

• rc in on_connect — 0 is success; values such as 134 (Bad User Name or Password) or 135 (Not authorized) explain a refused connection.
• The qos tuple in on_subscribe(client, mid, qos, properties) — a value of 128 or higher for a topic means the broker refused that subscription, typically because of an ACL.
• The return value of on_message is used as the reason code in the PUBACK/PUBREC you send back; 0 means success.

Look up any number with the MQTT reason code lookup.

To leave with a Last Will still being sent (useful to test monitoring), disconnect with reason code 4: await client.disconnect(reason_code=4). A Will itself is configured with a Message object, optionally with a will delay interval:

from gmqtt import Client as MQTTClient, Message

will = Message("testmqtt/gmqtt/sensor-1/status", "offline", qos=1, retain=True, will_delay_interval=10)
client = MQTTClient("testmqtt-gmqtt-sensor-1", will_message=will)

Reconnects and session expiry

Unlike aiomqtt, gmqtt reconnects automatically. You can tune how often and how long it waits:

client.set_config({"reconnect_retries": 10, "reconnect_delay": 5})

Combine reconnects with session_expiry_interval and a fixed client ID: the broker keeps your subscriptions and queues QoS 1 messages while you are away, and the CONNACK flags tell you whether a session was resumed. On a new connection, flags in on_connect indicates session present; if it is 0 after an outage, the session expired and you must re-subscribe (subscribing in on_connect, as above, handles that). For a hands-on check, see testing session expiry.

Testing your gmqtt client

1. Run the subscriber from the first example.
2. Open the online MQTT client, connect to broker.connectmq.com with MQTT 5, and publish to testmqtt/gmqtt/sensor-1/temperature with a user property and a message expiry.
3. Confirm the properties dict in your terminal contains exactly what you sent.
4. Stop the subscriber, publish a QoS 1 message, start it again within the session expiry window — the message should arrive on reconnect.

Publish MQTT 5 messages to your gmqtt client

gmqtt vs other Python clients

gmqtt is a good fit when MQTT 5 metadata is central to your design — request/response, content types, expiry — and your code is already asynchronous. Since it is maintained by a smaller team than paho-mqtt, check recent releases and pin the version you test with. For the bigger picture, see MQTT 5 vs 3.1.1.

gmqtt makes MQTT 5 feel like normal Python: properties are arguments, results are dictionaries, and reconnects happen on their own. Print the CONNACK properties once, read property values as lists, and verify every feature with a second client before relying on it in production.

Frequently asked questions