Python MQTT JSON: Publicando Dados de Sensores com Timestamps e Versões de Schema

Payloads JSON enviados por MQTT a partir do Python são a forma padrão de levar dados de sensores entre dispositivos, dashboards e bancos de dados, mas pequenos deslizes, como timestamps locais, valores NaN ou campos sem versão, viram grandes dores de cabeça mais adiante. Neste tutorial você vai construir um publisher em Python que serializa as leituras em JSON compacto e válido, com timestamps em UTC, números de sequência e uma versão de schema, e as envia ao broker.connectmq.com com metadados de content type do MQTT 5, além de um subscriber que valida o que recebe. No final, você terá um formato de payload que pode continuar usando à medida que o projeto cresce.

Por que JSON para payloads MQTT?

O MQTT não se importa com o que há dentro de uma mensagem; o payload é apenas bytes, até o limite de tamanho do broker. Essa liberdade significa que você precisa escolher um formato. JSON é a escolha habitual porque:

• toda linguagem e ferramenta consegue interpretá-lo, do Node-RED e do Telegraf a bancos de dados com colunas JSON;
• é legível por humanos, então dá para depurar com mosquitto_sub ou com um explorador de topics;
• é autodescritivo, então adicionar um campo não quebra consumidores mais antigos.

A contrapartida é o tamanho. O JSON repete os nomes dos campos em cada mensagem, o que pesa em enlaces limitados. Para a maioria dos dispositivos Wi-Fi e Ethernet esse custo é desprezível, e os benefícios na depuração vencem. Se mais tarde você precisar de algo mais compacto, formatos binários como o Protobuf (usado pelo Sparkplug B) são uma opção, como mostra o guia de Sparkplug B e Unified Namespace.

Projetando o payload

Um bom payload de sensor responde a cinco perguntas: qual é a versão do formato, qual dispositivo o enviou, quando a medição foi feita, qual é a posição da mensagem na sequência e quais são os valores com as suas unidades. Este é o formato usado no tutorial:

{
  "v": 1,
  "device": "room1",
  "ts": "2026-10-16T08:15:30.123Z",
  "seq": 42,
  "temperature_c": 21.4,
  "humidity_pct": 45.2,
  "battery_v": null
}

As decisões de design por trás de cada campo:

Colocar as unidades nos nomes dos campos evita objetos aninhados como {"value": 21.4, "unit": "C"} para cada leitura, o que mantém os payloads pequenos e as consultas simples. Seja qual for a convenção escolhida, aplique-a de forma consistente em todos os dispositivos.

Construindo um publisher MQTT JSON em Python

Instale o paho-mqtt e salve o publisher como json_publisher.py. Troque k7f3q2 por uma string aleatória sua para ficar separado dos outros usuários do servidor MQTT público gratuito:

pip install "paho-mqtt>=2.0"

import json
import math
import random
import time
import uuid
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from typing import Optional

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

BROKER = "broker.connectmq.com"
PORT = 1883
PREFIX = "connectmq/demo/k7f3q2"
DEVICE = "room1"
SCHEMA_VERSION = 1


@dataclass
class Reading:
    device: str
    ts: str
    seq: int
    temperature_c: Optional[float]
    humidity_pct: Optional[float]
    battery_v: Optional[float] = None
    v: int = SCHEMA_VERSION


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")


def clean(value: Optional[float], digits: int = 2) -> Optional[float]:
    """Arredonda floats e transforma NaN ou infinito em None, que vira null no JSON."""
    if value is None or math.isnan(value) or math.isinf(value):
        return None
    return round(value, digits)


def read_sensor() -> tuple:
    # Substitua pelo código real do sensor. De vez em quando simula uma leitura com falha.
    if random.random() < 0.05:
        return float("nan"), float("nan")
    return random.uniform(20, 25), random.uniform(40, 55)


def to_payload(reading: Reading) -> bytes:
    return json.dumps(asdict(reading), separators=(",", ":"), allow_nan=False).encode("utf-8")


client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id=f"json-pub-{uuid.uuid4().hex[:8]}",
    protocol=mqtt.MQTTv5,
)
# Broker ConnectMQ privado: client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"); client.tls_set(); porta 8883
client.connect(BROKER, PORT, keepalive=60)
client.loop_start()

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

topic = f"{PREFIX}/sensors/{DEVICE}/telemetry"
seq = 0
try:
    while True:
        temperature, humidity = read_sensor()
        reading = Reading(
            device=DEVICE,
            ts=utc_now_iso(),
            seq=seq,
            temperature_c=clean(temperature),
            humidity_pct=clean(humidity, 1),
        )
        payload = to_payload(reading)
        client.publish(topic, payload, qos=1, properties=props)
        print(f"{len(payload):>4} bytes  {payload.decode()}")
        seq += 1
        time.sleep(2)
except KeyboardInterrupt:
    pass
finally:
    client.disconnect()
    client.loop_stop()

Execute python json_publisher.py. A cada dois segundos é impressa uma linha como esta, e cerca de uma em cada vinte leituras mostra valores null onde o sensor simulado falhou:

 116 bytes  {"device":"room1","ts":"2026-10-16T08:15:30.123Z","seq":42,"temperature_c":21.4,"humidity_pct":45.2,"battery_v":null,"v":1}

Por que cada detalhe importa

Separadores compactos

separators=(",", ":") remove os espaços que o json.dumps adiciona por padrão. Em um payload desse tamanho, isso economiza cerca de dez por cento, de graça. Formatação bonita é coisa das ferramentas de depuração, não do que trafega na rede.

allow_nan=False e a armadilha do NaN

Por padrão, o json.dumps do Python escreve NaN e Infinity sem reclamar, mas isso não é JSON válido. Muitos consumidores, incluindo parsers estritos em JavaScript, Go e bancos de dados, rejeitam a mensagem inteira. Sensores produzem NaN o tempo todo quando uma leitura falha, então a função auxiliar clean() os converte em None, e allow_nan=False faz qualquer valor que escape gerar um erro do seu lado, em vez de corromper dados mais adiante.

Timestamps em UTC com milissegundos

Timestamps sem fuso horário são ambíguos, e horários locais pulam duas vezes por ano com o horário de verão. ISO 8601 em UTC com sufixo Z ordena corretamente como string e é interpretado em qualquer lugar. Se o seu dispositivo não tem relógio de tempo real, sincronize-o com NTP antes de publicar, ou omita ts e deixe o consumidor registrar o horário de recebimento.

Números de sequência

O seq custa poucos bytes e dá aos consumidores uma forma de identificar mensagens perdidas (um salto de 41 para 43) e duplicadas (o QoS 1 pode entregar uma mensagem duas vezes). Reinicie-o no boot e deixe os consumidores tratarem uma queda para zero como uma reinicialização.

Content type do MQTT 5

As properties ContentType e PayloadFormatIndicator informam a qualquer consumidor MQTT 5 que o payload é JSON em UTF-8, sem precisar adivinhar. Elas são opcionais, mas facilitam o trabalho com ferramentas genéricas e árvores de topics com formatos mistos. Nosso artigo sobre recursos do MQTT 5 cobre as demais properties.

Versionando o seu schema

O seu payload vai mudar: novos sensores, campos renomeados, unidades diferentes. Planeje isso desde a primeira mensagem:

1. Mudanças aditivas (um novo campo opcional) mantêm o mesmo v. Consumidores bem escritos ignoram campos que não conhecem.
2. Mudanças incompatíveis (renomear, remover, trocar unidades ou tipos) incrementam v para 2.
3. Durante uma migração, publique as duas versões ou mantenha os consumidores capazes de ler ambas, despachando conforme v.
4. Nunca reutilize um número de versão para um formato diferente.
5. Documente cada versão junto da sua estrutura de topics, para que todos usem o mesmo contrato.

Um subscriber que valida os dados

Consumidores nunca devem confiar cegamente nos payloads. Este subscriber faz o parse, verifica a versão e os campos obrigatórios e rejeita todo o resto sem travar:

import json
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/k7f3q2"
REQUIRED = {"v", "device", "ts", "seq"}


def parse_reading(raw: bytes):
    try:
        data = json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"not valid JSON: {exc}")
    if not isinstance(data, dict) or not REQUIRED <= data.keys():
        raise ValueError("missing required fields")
    if data["v"] != 1:
        raise ValueError(f"unsupported schema version {data['v']}")
    return data


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/sensors/+/telemetry", qos=1)


def on_message(client, userdata, msg):
    try:
        r = parse_reading(msg.payload)
    except ValueError as exc:
        print(f"Rejected message on {msg.topic}: {exc}")
        return
    print(f"{r['device']} #{r['seq']} at {r['ts']}: {r.get('temperature_c')} C, {r.get('humidity_pct')} %")


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"json-sub-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Inspecionando mensagens JSON

Para verificações rápidas no servidor MQTT público, encaminhe a saída do cliente de linha de comando para o jq e obtenha um resultado formatado e colorido:

mosquitto_sub -h broker.connectmq.com -p 1883 -i json-inspect-k7f3q2 -t 'connectmq/demo/k7f3q2/sensors/+/telemetry' | jq .

Em um broker privado, o explorador de topics ao vivo do portal do ConnectMQ mostra cada topic da sua árvore com o payload mais recente, para que você confirme nomes de campos, tratamento de nulos e timestamps sem escrever um consumidor.

Limites de tamanho e envio em lote

Fique de olho no tamanho das mensagens. O servidor MQTT público aceita até 64 KB por mensagem, e o plano Serverless gratuito, até 256 KB. Leituras individuais ficam muito abaixo disso, mas, se você agrupar, por exemplo enviando um minuto de amostras como um array, confira len(payload) antes de publicar e divida lotes grandes. Outra forma de reduzir o tráfego é publicar apenas quando um valor muda mais do que um limiar definido, junto com um heartbeat periódico para que os consumidores distingam um sensor quieto de um sensor morto. Essa abordagem costuma reduzir drasticamente o volume de mensagens para valores que mudam devagar, como a temperatura de um ambiente, sem perder nenhuma informação relevante.

Próximos passos

Armazene esses payloads com o tutorial de MQTT para SQLite em Python, cujas consultas com json_extract funcionam diretamente com este formato, ou crie gráficos ao vivo com o dashboard MQTT em Streamlit. Para serviços assíncronos, o mesmo código de payload se encaixa em um cliente aiomqtt com asyncio. A visão geral de bibliotecas de cliente MQTT mostra como outras linguagens interpretam o mesmo JSON, e a documentação do módulo json do Python cobre todas as opções do encoder.

Comece a usar o ConnectMQ

O ConnectMQ é um broker MQTT gerenciado com suporte ao MQTT 5 para properties como content type, um explorador de topics ao vivo para inspecionar payloads JSON e TLS na porta 8883. Faça o protótipo no servidor MQTT público broker.connectmq.com (apenas para testes) e depois leve os seus dispositivos para um broker MQTT na nuvem privado. Cadastre-se no plano Serverless gratuito, sem cartão de crédito.

Perguntas frequentes

Como envio JSON por MQTT em Python?

Serialize os dados com json.dumps(), codifique-os em bytes UTF-8 e passe-os como payload para client.publish(). Do lado de quem recebe, chame json.loads(msg.payload).

Os timestamps devem ir no payload ou ser adicionados pelo consumidor?

Inclua um timestamp de medição em UTC no payload quando o dispositivo tiver um relógio confiável. Se não tiver, deixe o consumidor registrar o horário de recebimento e omita ts.

Por que o meu payload JSON contém NaN?

O json.dumps do Python escreve NaN para valores float NaN por padrão, o que é JSON inválido. Converta NaN em None antes de codificar e defina allow_nan=False para capturar os que sobrarem.

Onde deve ficar a versão do schema, no payload ou no topic?

As duas opções funcionam. Um campo v no payload é mais simples e mantém os topics estáveis; um segmento de versão no topic permite que os consumidores assinem apenas as versões que entendem.