Python MQTT JSON: Publishing Sensor Data with Timestamps and Schema Versions

Python MQTT JSON payloads are the default way to move sensor data between devices, dashboards and databases, but small mistakes like local timestamps, NaN values or unversioned fields cause big headaches later. In this tutorial you will build a Python publisher that serializes readings to compact, valid JSON with UTC timestamps, sequence numbers and a schema version, sends them to broker.connectmq.com with MQTT 5 content-type metadata, and a subscriber that validates what it receives. You will finish with a payload format you can keep using as your project grows.

Why JSON for MQTT Payloads?

MQTT does not care what is inside a message; the payload is just bytes, up to the broker's size limit. That freedom means you must pick a format. JSON is the usual choice because:

• every language and tool can parse it, from Node-RED and Telegraf to databases with JSON columns;
• it is human-readable, so you can debug with mosquitto_sub or a topic explorer;
• it is self-describing, so adding a field does not break older consumers.

The trade-off is size. JSON repeats field names in every message, which matters on constrained links. For most Wi-Fi and Ethernet devices that cost is negligible, and the debugging benefits win. If you later need something more compact, binary formats such as Protobuf (which Sparkplug B uses) are an option, as covered in the Sparkplug B and Unified Namespace guide.

Designing the Payload

A good sensor payload answers five questions: what version of the format is this, which device sent it, when was it measured, which message in the sequence is it, and what are the values with their units. Here is the format this tutorial uses:

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

The design decisions behind each field:

Putting units in field names avoids nested {"value": 21.4, "unit": "C"} objects for every reading, which keeps payloads small and queries simple. Whichever convention you choose, apply it consistently across all devices.

Building a Python MQTT JSON Publisher

Install paho-mqtt and save the publisher as json_publisher.py. Change k7f3q2 to your own random string so you stay separate from other users of the free public MQTT broker:

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]:
    """Round floats and turn NaN or infinity into None, which becomes JSON null."""
    if value is None or math.isnan(value) or math.isinf(value):
        return None
    return round(value, digits)


def read_sensor() -> tuple:
    # Replace with real sensor code. Occasionally simulate a failed read.
    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,
)
# Private ConnectMQ broker: client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"); client.tls_set(); port 8883
client.connect(BROKER, PORT, keepalive=60)
client.loop_start()

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

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

Run python json_publisher.py. Every two seconds a line like this prints, and about one in twenty readings shows null values where the simulated sensor failed:

 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}

Why Each Detail Matters

Compact separators

separators=(",", ":") removes the spaces json.dumps adds by default. On a payload this size that saves around ten percent, for free. Pretty-printing belongs in your debugging tools, not on the wire.

allow_nan=False and the NaN trap

Python's json.dumps happily writes NaN and Infinity by default, but those are not valid JSON. Many consumers, including strict parsers in JavaScript, Go and databases, reject the whole message. Sensors produce NaN all the time when a read fails, so the clean() helper converts them to None, and allow_nan=False makes any value that slips through raise an error on your side instead of corrupting data downstream.

UTC timestamps with milliseconds

Timestamps without a time zone are ambiguous, and local times jump twice a year with daylight saving. ISO 8601 in UTC with a Z suffix sorts correctly as a string and parses everywhere. If your device has no real-time clock, sync it with NTP before publishing, or leave ts out and let the consumer record the receive time.

Sequence numbers

seq costs a few bytes and gives consumers a way to spot missing messages (a jump from 41 to 43) and duplicates (QoS 1 can deliver a message twice). Reset it on reboot and let consumers treat a drop to zero as a restart.

MQTT 5 content type

The ContentType and PayloadFormatIndicator properties tell any MQTT 5 consumer that the payload is UTF-8 JSON without guessing. They are optional, but they make generic tools and mixed-format topic trees easier to handle. Our MQTT 5 features article covers the other properties.

Versioning Your Schema

Your payload will change: new sensors, renamed fields, different units. Plan for it from the first message:

1. Additive changes (a new optional field) keep the same v. Well-written consumers ignore fields they do not know.
2. Breaking changes (renaming, removing, changing units or types) bump v to 2.
3. During a migration, publish both versions or keep consumers able to read both, dispatching on v.
4. Never reuse a version number for a different format.
5. Document each version next to your topic structure so everyone uses the same contract.

A Validating Subscriber

Consumers should never trust payloads blindly. This subscriber parses, checks the version and required fields, and rejects anything else without crashing:

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

Inspecting JSON Messages

For quick checks against the public broker, pipe the command-line client into jq for pretty, colourised output:

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

On a private broker, the ConnectMQ portal's live topic explorer shows each topic in your tree with its latest payload, so you can confirm field names, null handling and timestamps without writing a consumer.

Size Limits and Batching

Keep an eye on message size. The public broker accepts up to 64 KB per message and the free Serverless plan up to 256 KB. Single readings are far below that, but if you batch, for example sending a minute of samples as an array, check len(payload) before publishing and split large batches. Another way to keep traffic down is to publish only when a value changes by more than a set threshold, plus a periodic heartbeat so consumers can tell a quiet sensor from a dead one. That approach often cuts message volume dramatically for slow-moving values such as room temperature, without losing any meaningful information.

Where to Go Next

Store these payloads with the MQTT to SQLite in Python tutorial, whose json_extract queries work directly with this format, or chart them live with the Streamlit MQTT dashboard. For async services, the same payload code drops into an aiomqtt asyncio client. The MQTT client libraries overview shows how other languages parse the same JSON, and the Python json module documentation covers every encoder option.

Get Started with ConnectMQ

ConnectMQ is a hosted MQTT broker with MQTT 5 support for properties like content type, a live topic explorer for inspecting JSON payloads, and TLS on port 8883. Prototype on broker.connectmq.com, then move your devices to a private cloud MQTT broker. Sign up for the free Serverless plan, no credit card required.

Frequently Asked Questions

How do I send JSON over MQTT in Python?

Serialize your data with json.dumps(), encode it to UTF-8 bytes and pass it as the payload to client.publish(). On the receiving side, call json.loads(msg.payload).

Should timestamps be in the payload or added by the consumer?

Include a UTC measurement timestamp in the payload when the device has a reliable clock. If it does not, let the consumer record the receive time and leave ts out.

Why does my JSON payload contain NaN?

Python's json.dumps writes NaN for float NaN values by default, which is invalid JSON. Convert NaN to None before encoding and set allow_nan=False to catch any that remain.

Where should the schema version go, payload or topic?

Either works. A v field in the payload is simpler and keeps topics stable; a version segment in the topic lets consumers subscribe to only the versions they understand.