Enviando JSON via MQTT em Python

O MQTT não sabe o que é JSON: um payload é só bytes. Para enviar JSON via MQTT em Python, serialize seu objeto com json.dumps() antes do publish() e chame json.loads(msg.payload) no on_message. Todo o resto deste guia é sobre fazer isso com segurança: validar o que chega, sobreviver a mensagens ruins, lidar com datas e manter o payload pequeno.

Todos os exemplos usam paho-mqtt 2.x e o broker de testes broker.connectmq.com.

A ida e volta mínima

import json
import time

import paho.mqtt.client as mqtt

TOPIC = "testmqtt/json-demo/sensor-42/telemetry"


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(TOPIC, qos=1)


def on_message(client, userdata, msg):
    data = json.loads(msg.payload)  # entram bytes, sai um dict
    print(msg.topic, data["temperature"], type(data["temperature"]))


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883)
client.loop_start()
time.sleep(1)

reading = {"device_id": "sensor-42", "temperature": 21.5, "humidity": 48.2}
info = client.publish(TOPIC, json.dumps(reading), qos=1)
info.wait_for_publish()

time.sleep(1)
client.loop_stop()
client.disconnect()

Três fatos explicam a maioria dos bugs de iniciante:

• json.dumps() retorna um str; o paho codifica strings em UTF-8 antes de enviar. Passar o próprio dict falha com TypeError: payload must be a string, bytearray, int, float or None.
• msg.payload é sempre bytes. json.loads() aceita bytes UTF-8, então você não precisa de .decode() antes.
• JSON não tem tuplas, sets, datetime nem Decimal. Tuplas voltam como listas; os outros lançam TypeError: Object of type datetime is not JSON serializable.

Você pode acompanhar as mensagens com o cliente MQTT online assinando testmqtt/json-demo/#. Ele mostra o payload como texto, então você vê exatamente o que seu código colocou na rede.

Nunca deixe um payload ruim derrubar seu subscriber

Qualquer um com permissão de publish em um topic pode mandar qualquer coisa: uma string simples, uma mensagem truncada, texto Latin-1 de um dispositivo antigo. Se json.loads() lançar exceção dentro do on_message, ela se propaga e seu loop de rede para. Capture os dois erros que importam:

import json
import logging

log = logging.getLogger("mqtt")


def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload)
    except UnicodeDecodeError:
        log.warning("payload on %s is not UTF-8: %r", msg.topic, msg.payload[:50])
        return
    except json.JSONDecodeError as exc:
        log.warning("invalid JSON on %s: %s (%r)", msg.topic, exc, msg.payload[:50])
        return

    if not isinstance(data, dict):
        log.warning("expected an object on %s, got %s", msg.topic, type(data).__name__)
        return

    handle_reading(msg.topic, data)

A verificação isinstance importa mais do que parece: 42, "ok" e null são JSON válido, e código que faz data["temperature"] vai quebrar com eles com um TypeError. As mensagens típicas que você verá nos logs, como Expecting value: line 1 column 1 (char 0) para um payload vazio ou que não é JSON, são explicadas em erros de parsing de JSON no MQTT.

Valide a estrutura com pydantic

Capturar erros de sintaxe não basta: {"temperature": "hot"} faz parse sem problema. Um schema torna o contrato explícito, e o pydantic v2 consegue fazer parse e validar bytes em um único passo:

from datetime import datetime

from pydantic import BaseModel, Field, ValidationError


class Reading(BaseModel):
    device_id: str = Field(min_length=1, max_length=64)
    ts: datetime
    temperature: float = Field(ge=-50, le=150)
    humidity: float | None = Field(default=None, ge=0, le=100)


def on_message(client, userdata, msg):
    try:
        reading = Reading.model_validate_json(msg.payload)
    except ValidationError as exc:
        log.warning("rejected message on %s: %s", msg.topic, exc.errors(include_url=False))
        return
    store(reading)  # reading.ts é um datetime de verdade, reading.temperature um float

Do lado do publisher, o mesmo model se serializa sozinho, incluindo o datetime:

from datetime import datetime, timezone

reading = Reading(device_id="sensor-42", ts=datetime.now(timezone.utc), temperature=21.5)
client.publish("testmqtt/json-demo/sensor-42/telemetry", reading.model_dump_json(), qos=1)

Compartilhar um módulo de models entre publisher e subscriber é a forma mais barata de contract testing: se alguém renomear um campo, a suíte de testes falha antes dos dispositivos.

Datas, decimais e outros tipos chatos

Se você não usa pydantic, diga ao json.dumps() como tratar os tipos que ele não conhece:

import json
from datetime import datetime
from decimal import Decimal


def default(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    if isinstance(obj, Decimal):
        return str(obj)
    if isinstance(obj, set):
        return sorted(obj)
    raise TypeError(f"Cannot serialize {type(obj).__name__}")


payload = json.dumps(data, default=default, allow_nan=False)

A linha do NaN é uma armadilha real: o Python gera {"t": NaN} sem reclamar, mas o JSON.parse do JavaScript, o Go e a maioria dos parsers estritos rejeitam. allow_nan=False transforma isso em um ValueError no publisher, onde é fácil de corrigir.

Codificação compacta

Cada byte conta em links celulares e dispositivos a bateria, e o MQTT cobra o payload em cada salto. Esta é a mesma leitura codificada de quatro formas (tamanhos medidos com Python 3):

As três primeiras usam as chaves completas device_id, timestamp (uma string ISO com microssegundos e offset), temperature, humidity, battery e status. A última é {"id":"sensor-42","ts":1792229400,"t":21.46,"h":48.2,"b":0.87}. As mudanças que fazem diferença:

• Separadores compactos removem os espaços depois de , e :.
• Chaves curtas são o maior ganho em mensagens pequenas; documente-as no seu schema.
• Timestamps Unix em vez de strings ISO economizam cerca de 20 bytes cada.
• Arredonde os floats para a precisão que seu sensor realmente tem. 21.456789 carrega ruído, não informação.
• ensure_ascii=False mantém texto não ASCII como UTF-8 ("Zürich") em vez de escapes ü, que usam 6 bytes em vez de 2.

Para velocidade e conveniência, o orjson é uma troca direta: gera saída compacta por padrão, retorna bytes (que o paho aceita diretamente) e serializa datetime nativamente.

import orjson

client.publish(TOPIC, orjson.dumps(reading_dict), qos=1)

def on_message(client, userdata, msg):
    data = orjson.loads(msg.payload)  # raises orjson.JSONDecodeError (a ValueError subclass)

Confira o tamanho total do pacote, incluindo topic e cabeçalhos, com a calculadora de tamanho de pacote MQTT. Se o JSON compacto ainda for grande demais, o próximo passo são os formatos binários; veja CBOR e MessagePack sobre MQTT e payloads Protobuf.

Avise aos subscribers que é JSON (MQTT 5)

O MQTT 5 permite que o publisher rotule o payload, para que ferramentas genéricas e outros serviços não precisem adivinhar:

import json

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)
client.connect("broker.connectmq.com", 1883)
client.loop_start()

props = Properties(PacketTypes.PUBLISH)
props.ContentType = "application/json"
props.PayloadFormatIndicator = 1  # 1 = texto UTF-8

info = client.publish("testmqtt/json-demo/sensor-42/telemetry",
                      json.dumps({"temperature": 21.5}), qos=1, properties=props)
info.wait_for_publish()

Do lado de quem recebe, leia msg.properties.ContentType (verifique antes com hasattr, porque só existe quando o publisher o definiu). Com o payload format indicator em 1, um broker MQTT 5 pode rejeitar payloads que não sejam UTF-8 válido; mais sobre isso em erros de codificação UTF-8 no MQTT.

Teste seu tratamento de JSON

Antes de confiar em um subscriber, mande para ele as mensagens que quebram coisas. Com mosquitto_pub (ou o cliente online):

T=testmqtt/json-demo/sensor-42/telemetry
H=broker.connectmq.com
mosquitto_pub -h $H -t $T -m '{"device_id":"sensor-42","ts":"2026-10-17T09:30:00Z","temperature":21.5}'
mosquitto_pub -h $H -t $T -m 'hello'                          # não é JSON
mosquitto_pub -h $H -t $T -n                                 # payload vazio
mosquitto_pub -h $H -t $T -m '[1,2,3]'                        # JSON válido, formato errado
mosquitto_pub -h $H -t $T -m '{"temperature":"hot"}'          # tipo errado
printf '{"city":"Z\xfcrich"}' | mosquitto_pub -h $H -t $T -s  # Latin-1, não UTF-8

O subscriber deve gravar a primeira mensagem e logar um aviso claro para cada uma das outras, sem desconectar. Coloque os mesmos casos em um teste unitário que chame sua função on_message diretamente com um objeto de mensagem falso, e você vai pegar regressões sem precisar de broker.

Inspecione os payloads JSON no cliente online

JSON sobre MQTT é simples quando você trata o payload como bytes não confiáveis: serialize explicitamente, faça parse de forma defensiva, valide contra um schema e mantenha tudo compacto. Para as questões maiores de design, como envelopes, timestamps e estrutura de topics, continue com boas práticas de design de payload MQTT.

Perguntas frequentes