MQTT em Python com paho-mqtt: tutorial de publish e subscribe

O paho-mqtt é a biblioteca cliente MQTT padrão para Python. Ela suporta MQTT 3.1, 3.1.1 e 5, TCP simples, TLS e WebSockets, e roda em qualquer lugar, de um Raspberry Pi a um servidor na nuvem. Este tutorial usa a API atual do paho-mqtt 2.x — posts antigos escritos para a 1.x vão gerar erros se você copiá-los como estão, então as diferenças são destacadas ao longo do texto.

Instalar o paho-mqtt

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

Confira sua versão com python -c "import paho.mqtt; print(paho.mqtt.__version__)". Todos os exemplos abaixo usam a 2.x e o sandbox público da HiveMQ em broker.hivemq.com, então você pode rodá-los sem configurar um broker. Se preferir rodar o seu próprio, veja como instalar o Mosquitto.

O que mudou no paho-mqtt 2.x

A grande mudança é a versão da API de callbacks. O primeiro argumento de mqtt.Client() agora é um CallbackAPIVersion e, com VERSION2, todos os callbacks recebem reason codes e propriedades no estilo do MQTT 5 — mesmo quando você se conecta com MQTT 3.1.1.

reason_code é um objeto ReasonCode: verifique reason_code.is_failure em vez de comparar rc == 0. Nossa consulta de reason codes MQTT explica o que cada valor significa.

Um assinante completo

O assinante abaixo se conecta, assina dentro do on_connect e imprime todas as mensagens. Assinar no on_connect, e não depois de connect(), faz diferença: quando a conexão cai e o paho se reconecta, o callback roda de novo e suas assinaturas são restauradas.

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()

Um client_id vazio deixa o paho gerar um aleatório. Se você definir o seu, garanta que seja único — dois clientes com o mesmo ID ficam derrubando um ao outro do broker. O gerador de client ID pode criar IDs seguros para você.

Um publicador completo

Um publicador geralmente tem outro trabalho a fazer — ler um sensor, consultar uma API — então ele roda o loop de rede em uma thread em segundo plano com 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()

Rode o assinante em um terminal e o publicador em outro. Você também pode acompanhar pelo navegador as mensagens que seu script publica:

Veja suas mensagens Python ao vivo

loop_forever vs loop_start vs loop

• loop_forever() — bloqueia a thread atual, cuida das reconexões e só retorna depois de disconnect(). Ideal para assinantes puros e serviços.
• loop_start() / loop_stop() — roda o loop em uma thread em segundo plano. Ideal quando sua thread principal faz outro trabalho. Os callbacks rodam nessa thread em segundo plano, então proteja o estado compartilhado de acordo.
• loop(timeout) — processa o tráfego de rede uma vez. Só é necessário se você integrar o paho ao seu próprio event loop.

QoS, mensagens retidas e Last Will

Passe qos=0, 1 ou 2 tanto para publish() quanto para subscribe(); o QoS efetivo é o menor dos dois. retain=True faz o broker guardar a última mensagem para novos assinantes. Defina um Last Will antes de conectar para que o broker anuncie uma desconexão inesperada:

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)

Saiba mais em níveis de QoS do MQTT, mensagens retidas (retained) e Last Will and Testament.

TLS e autenticação com usuário e senha

Qualquer implantação real deve usar TLS na porta 8883 e credenciais. Ambos são configurados antes de 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()

Evite tls_insecure_set(True) fora de testes locais rápidos — ele desativa a verificação do hostname. Para MQTT 5, crie o cliente com protocol=mqtt.MQTTv5. Para WebSockets, passe transport="websockets" e chame client.ws_set_options(path="/mqtt").

Reconectando de forma confiável

Tanto loop_forever() quanto loop_start() reconectam automaticamente após uma perda de conexão. Ajuste o backoff com reconnect_delay_set():

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

O primeiríssimo connect() é diferente: ele lança uma exceção (por exemplo ConnectionRefusedError ou um timeout de socket) se o broker estiver inacessível. Envolva-o em um loop de retentativas ou use connect_async() seguido de loop_start(), que tenta de novo em segundo plano. Se as conexões continuarem falhando, o guia de erros de conexão MQTT cobre os suspeitos de sempre: porta errada, firewall bloqueando, credenciais inválidas ou client IDs duplicados.

Usando recursos do MQTT 5

Passe protocol=mqtt.MQTTv5 para liberar o MQTT 5: session expiry, user properties, expiração de mensagens e reason codes detalhados. As opções de conexão passam de clean_session (um argumento do construtor no 3.1.1) para clean_start em connect(), e os metadados por mensagem vão em um objeto Properties:

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()

No on_message, as propriedades recebidas ficam disponíveis em msg.properties. Não sabe se precisa da versão 5? Nossa comparação MQTT 5 vs 3.1.1 lista as diferenças que importam na prática.

Erros comuns

• Copiar código da 1.x. mqtt.Client("my-id") passa o client ID onde a 2.x espera a versão da API de callbacks. Use o argumento nomeado client_id=.
• Esquecer o loop. Sem loop_forever() ou loop_start(), nada é enviado nem recebido e os callbacks nunca disparam.
• Sair cedo demais. Um script que chama publish() e termina imediatamente pode encerrar antes de a mensagem sair do socket. Chame wait_for_publish() ou use publish.single().
• Decodificar às cegas. msg.payload é bytes. Use msg.payload.decode("utf-8") para texto e json.loads() para JSON, e trate erros de decodificação — outros clientes no mesmo tópico podem enviar qualquer coisa.
• Assinar # em um broker público. Você vai receber o tráfego de todos os outros usuários. Fique dentro do seu próprio prefixo de tópico; veja tópicos e curingas MQTT.

Publicação avulsa

Para cron jobs e scripts que só enviam uma mensagem, o helper paho.mqtt.publish cuida de conectar, publicar e desconectar em uma única chamada:

import paho.mqtt.publish as publish

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

Quer este código já preenchido com seu host, porta, credenciais e tópico? Use o gerador de código MQTT para gerá-lo com as configurações do seu broker.

Perguntas frequentes