MQTT in Python with paho-mqtt: Publish & Subscribe Tutorial

paho-mqtt is the standard MQTT client library for Python. It supports MQTT 3.1, 3.1.1 and 5, plain TCP, TLS and WebSockets, and runs anywhere from a Raspberry Pi to a cloud server. This tutorial uses the current paho-mqtt 2.x API — older blog posts written for 1.x will throw errors if you copy them as-is, so the differences are called out along the way.

Install paho-mqtt

python -m pip install "paho-mqtt>=2.0"

Check your version with python -c "import paho.mqtt; print(paho.mqtt.__version__)". Every example below targets 2.x and the public HiveMQ sandbox at broker.hivemq.com, so you can run them without setting up a broker. If you would rather run your own, see how to install Mosquitto.

What changed in paho-mqtt 2.x

The big change is the callback API version. The first argument to mqtt.Client() is now a CallbackAPIVersion, and with VERSION2 all callbacks receive MQTT 5 style reason codes and properties — even when you connect with MQTT 3.1.1.

reason_code is a ReasonCode object: check reason_code.is_failure instead of comparing rc == 0. Our MQTT reason code lookup explains what each value means.

A complete subscriber

The subscriber below connects, subscribes inside on_connect and prints every message. Subscribing in on_connect rather than after connect() matters: when the connection drops and paho reconnects, the callback runs again and your subscriptions are restored.

import paho.mqtt.client as mqtt

BROKER = "broker.hivemq.com"
TOPIC = "testmqtt/python-demo/#"


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"Connect failed: {reason_code}")
        return
    print("Connected, subscribing to", TOPIC)
    client.subscribe(TOPIC, qos=1)


def on_subscribe(client, userdata, mid, reason_code_list, properties):
    for rc in reason_code_list:
        if rc.is_failure:
            print(f"Subscription rejected: {rc}")
        else:
            print(f"Subscribed with granted QoS {rc.value}")


def on_message(client, userdata, msg):
    print(f"{msg.topic} [QoS {msg.qos}, retain={msg.retain}]: {msg.payload.decode()}")


def on_disconnect(client, userdata, disconnect_flags, reason_code, properties):
    print(f"Disconnected: {reason_code}")


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="")
client.on_connect = on_connect
client.on_subscribe = on_subscribe
client.on_message = on_message
client.on_disconnect = on_disconnect

client.connect(BROKER, 1883, keepalive=60)
client.loop_forever()

An empty client_id lets paho generate a random one. If you set your own, make it unique — two clients with the same ID keep kicking each other off the broker. The client ID generator can create safe IDs for you.

A complete publisher

A publisher usually has other work to do — reading a sensor, polling an API — so it runs the network loop in a background thread with loop_start():

import json
import random
import time

import paho.mqtt.client as mqtt

BROKER = "broker.hivemq.com"
TOPIC = "testmqtt/python-demo/sensor"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.connect(BROKER, 1883, keepalive=60)
client.loop_start()

try:
    while True:
        payload = json.dumps({
            "temperature": round(random.uniform(18, 26), 1),
            "ts": int(time.time()),
        })
        info = client.publish(TOPIC, payload, qos=1)
        info.wait_for_publish(timeout=5)
        print("Published", payload, "mid", info.mid)
        time.sleep(2)
except KeyboardInterrupt:
    pass
finally:
    client.disconnect()
    client.loop_stop()

Run the subscriber in one terminal and the publisher in another. You can also watch the messages your script publishes from the browser:

Watch your Python messages live

loop_forever vs loop_start vs loop

• loop_forever() — blocks the current thread, handles reconnects, and returns only after disconnect(). Best for pure subscribers and services.
• loop_start() / loop_stop() — runs the loop in a background thread. Best when your main thread does other work. Callbacks run on that background thread, so guard shared state accordingly.
• loop(timeout) — processes network traffic once. Only needed if you integrate paho into your own event loop.

QoS, retained messages and Last Will

Pass qos=0, 1 or 2 to both publish() and subscribe(); the effective QoS is the lower of the two. retain=True makes the broker store the last message for new subscribers. Set a Last Will before connecting so the broker announces an unexpected disconnect:

client.will_set("testmqtt/python-demo/status", "offline", qos=1, retain=True)
client.connect(BROKER, 1883)
client.publish("testmqtt/python-demo/status", "online", qos=1, retain=True)

Read more in MQTT QoS levels, retained messages and Last Will and Testament.

TLS and username/password authentication

Any real deployment should use TLS on port 8883 plus credentials. Both are configured before connect():

import ssl
import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sensor-livingroom-01")
client.username_pw_set("my-username", "my-password")

# Uses the system CA store; fine for publicly trusted certificates.
client.tls_set()

# For a private CA or client-certificate auth instead:
# client.tls_set(ca_certs="ca.crt", certfile="client.crt", keyfile="client.key",
#                tls_version=ssl.PROTOCOL_TLS_CLIENT)

client.connect("broker.example.com", 8883, keepalive=60)
client.loop_forever()

Avoid tls_insecure_set(True) outside quick local tests — it disables hostname verification. For MQTT 5, create the client with protocol=mqtt.MQTTv5. For WebSockets, pass transport="websockets" and call client.ws_set_options(path="/mqtt").

Reconnecting reliably

Both loop_forever() and loop_start() reconnect automatically after a connection loss. Tune the backoff with reconnect_delay_set():

client.reconnect_delay_set(min_delay=1, max_delay=60)

The very first connect() is different: it raises an exception (for example ConnectionRefusedError or a socket timeout) if the broker is unreachable. Wrap it in a retry loop or use connect_async() followed by loop_start(), which retries in the background. If connections keep failing, the MQTT connection errors guide covers the usual suspects: wrong port, blocked firewall, bad credentials or duplicate client IDs.

Using MQTT 5 features

Pass protocol=mqtt.MQTTv5 to unlock MQTT 5: session expiry, user properties, message expiry and detailed reason codes. Connection options move from clean_session (a constructor argument for 3.1.1) to clean_start on connect(), and per-message metadata goes into a Properties object:

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)

connect_props = Properties(PacketTypes.CONNECT)
connect_props.SessionExpiryInterval = 3600  # keep the session for 1 hour

client.connect("broker.hivemq.com", 1883, keepalive=60,
               clean_start=False, properties=connect_props)
client.loop_start()

publish_props = Properties(PacketTypes.PUBLISH)
publish_props.MessageExpiryInterval = 30          # drop if undelivered after 30 s
publish_props.ContentType = "application/json"
publish_props.UserProperty = [("source", "python-demo")]

info = client.publish("testmqtt/python-demo/v5", '{"ok": true}', qos=1,
                      properties=publish_props)
info.wait_for_publish(timeout=5)

client.disconnect()
client.loop_stop()

In on_message, the received properties are available as msg.properties. Not sure whether you need version 5? Our MQTT 5 vs 3.1.1 comparison lists the differences that matter in practice.

Common mistakes

• Copying 1.x code. mqtt.Client("my-id") passes the client ID where 2.x expects the callback API version. Use the keyword argument client_id=.
• Forgetting the loop. Without loop_forever() or loop_start(), nothing is sent or received and callbacks never fire.
• Exiting too early. A script that calls publish() and exits immediately may quit before the message leaves the socket. Call wait_for_publish() or use publish.single().
• Decoding blindly. msg.payload is bytes. Use msg.payload.decode("utf-8") for text and json.loads() for JSON, and catch decode errors — other clients on the same topic may send anything.
• Subscribing to # on a public broker. You will receive traffic from every other user. Stay inside your own topic prefix; see MQTT topics and wildcards.

One-shot publishing

For cron jobs and scripts that just send one message, the paho.mqtt.publish helper handles connect, publish and disconnect in a single call:

import paho.mqtt.publish as publish

publish.single("testmqtt/python-demo/cron", "job finished", qos=1,
               hostname="broker.hivemq.com", port=1883)

Want this code pre-filled with your own host, port, credentials and topic? Use the MQTT code generator to generate it for your broker settings.

Frequently asked questions