Les payloads Python MQTT JSON sont le moyen par défaut de faire circuler des données de capteurs entre appareils, tableaux de bord et bases de données, mais de petites erreurs comme des horodatages locaux, des valeurs NaN ou des champs non versionnés causent de gros maux de tête par la suite. Dans ce tutoriel, vous allez construire un publisher Python qui sérialise les mesures en JSON compact et valide avec horodatage UTC, numéros de séquence et version de schéma, les envoie à broker.connectmq.com avec les métadonnées content type de MQTT 5, ainsi qu'un subscriber qui valide ce qu'il reçoit. Vous terminerez avec un format de payload que vous pourrez continuer d'utiliser à mesure que votre projet grandit.
Pourquoi JSON pour les payloads MQTT ?
MQTT ne se soucie pas du contenu d'un message : le payload n'est qu'une suite d'octets, dans la limite de taille du broker. Cette liberté vous oblige à choisir un format. JSON est le choix habituel, car :
- tous les langages et outils savent le lire, de Node-RED et Telegraf aux bases de données dotées de colonnes JSON ;
- il est lisible par un humain, ce qui permet de déboguer avec
mosquitto_subou un explorateur de topics ; - il est autodescriptif : ajouter un champ ne casse pas les anciens consommateurs.
La contrepartie est la taille. JSON répète les noms de champs dans chaque message, ce qui compte sur les liaisons contraintes. Pour la plupart des appareils Wi-Fi et Ethernet, ce coût est négligeable et les avantages pour le débogage l'emportent. Si vous avez besoin plus tard d'un format plus compact, les formats binaires comme Protobuf (utilisé par Sparkplug B) sont une option, présentée dans le guide Sparkplug B et Unified Namespace.
Concevoir le payload
Un bon payload de capteur répond à cinq questions : quelle est la version du format, quel appareil l'a envoyé, quand la mesure a-t-elle été prise, quelle est la place du message dans la séquence, et quelles sont les valeurs avec leurs unités. Voici le format utilisé dans ce tutoriel :
{
"v": 1,
"device": "room1",
"ts": "2026-10-16T08:15:30.123Z",
"seq": 42,
"temperature_c": 21.4,
"humidity_pct": 45.2,
"battery_v": null
}
Les choix de conception derrière chaque champ :
| Champ | Rôle | Règle |
|---|---|---|
v |
Version du schéma | Entier ; à incrémenter uniquement lors de changements cassants |
device |
Identité de l'émetteur | Correspond au segment appareil du topic |
ts |
Instant de la mesure | ISO 8601 en UTC avec le suffixe Z |
seq |
Numéro de séquence | Permet aux consommateurs de détecter trous et doublons |
temperature_c, humidity_pct |
Valeurs | Unités dans le nom du champ, nombres simples |
battery_v |
Valeur facultative | null si indisponible, jamais NaN |
Mettre les unités dans les noms de champs évite des objets imbriqués {"value": 21.4, "unit": "C"} pour chaque mesure, ce qui garde des payloads légers et des requêtes simples. Quelle que soit la convention choisie, appliquez-la de façon cohérente sur tous les appareils.
Construire un publisher Python MQTT JSON
Installez paho-mqtt et enregistrez le publisher sous json_publisher.py. Remplacez k7f3q2 par votre propre chaîne aléatoire pour rester séparé des autres utilisateurs du serveur MQTT public gratuit :
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]:
"""Arrondir les flottants et convertir NaN ou l'infini en None, qui devient null en JSON."""
if value is None or math.isnan(value) or math.isinf(value):
return None
return round(value, digits)
def read_sensor() -> tuple:
# Remplacer par le vrai code du capteur. Simule de temps en temps une lecture ratée.
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 privé : 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 # le payload est du texte 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()
Lancez python json_publisher.py. Toutes les deux secondes, une ligne comme celle-ci s'affiche, et environ une mesure sur vingt montre des valeurs null là où le capteur simulé a échoué :
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}
Pourquoi chaque détail compte
Des séparateurs compacts
separators=(",", ":") supprime les espaces que json.dumps ajoute par défaut. Sur un payload de cette taille, cela économise environ dix pour cent, gratuitement. L'indentation lisible a sa place dans vos outils de débogage, pas sur le réseau.
allow_nan=False et le piège du NaN
Par défaut, json.dumps de Python écrit sans broncher NaN et Infinity, mais ce ne sont pas des valeurs JSON valides. De nombreux consommateurs, y compris les parseurs stricts en JavaScript, Go et dans les bases de données, rejettent alors le message entier. Les capteurs produisent sans cesse des NaN quand une lecture échoue : la fonction clean() les convertit donc en None, et allow_nan=False fait qu'une valeur qui passerait entre les mailles lève une erreur de votre côté au lieu de corrompre les données en aval.
Horodatage UTC avec millisecondes
Un horodatage sans fuseau horaire est ambigu, et l'heure locale saute deux fois par an avec l'heure d'été. L'ISO 8601 en UTC avec le suffixe Z se trie correctement en tant que chaîne et se parse partout. Si votre appareil n'a pas d'horloge temps réel, synchronisez-le par NTP avant de publier, ou omettez ts et laissez le consommateur enregistrer l'heure de réception.
Numéros de séquence
seq coûte quelques octets et donne aux consommateurs un moyen de repérer les messages manquants (un saut de 41 à 43) et les doublons (la QoS 1 peut livrer un message deux fois). Réinitialisez-le au redémarrage et laissez les consommateurs interpréter un retour à zéro comme un redémarrage.
Content type MQTT 5
Les propriétés ContentType et PayloadFormatIndicator indiquent à tout consommateur MQTT 5 que le payload est du JSON UTF-8, sans qu'il ait à le deviner. Elles sont facultatives, mais facilitent le travail des outils génériques et des arborescences de topics aux formats mixtes. Notre article sur les fonctionnalités MQTT 5 présente les autres propriétés.
Versionner votre schéma
Votre payload va évoluer : nouveaux capteurs, champs renommés, unités différentes. Préparez-vous dès le premier message :
- Les changements additifs (un nouveau champ facultatif) conservent le même
v. Des consommateurs bien écrits ignorent les champs qu'ils ne connaissent pas. - Les changements cassants (renommage, suppression, changement d'unité ou de type) font passer
và 2. - Pendant une migration, publiez les deux versions ou gardez des consommateurs capables de lire les deux, en aiguillant selon
v. - Ne réutilisez jamais un numéro de version pour un autre format.
- Documentez chaque version à côté de votre structure de topics pour que tout le monde utilise le même contrat.
Un subscriber qui valide
Les consommateurs ne devraient jamais faire aveuglément confiance aux payloads. Ce subscriber parse, vérifie la version et les champs obligatoires, et rejette tout le reste sans planter :
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()
Inspecter les messages JSON
Pour des vérifications rapides sur le serveur MQTT public gratuit, redirigez le client en ligne de commande vers jq pour obtenir une sortie mise en forme et colorée :
mosquitto_sub -h broker.connectmq.com -p 1883 -i json-inspect-k7f3q2 -t 'connectmq/demo/k7f3q2/sensors/+/telemetry' | jq .
Sur un broker privé, l'explorateur de topics en direct du portail ConnectMQ affiche chaque topic de votre arborescence avec son dernier payload : vous pouvez ainsi vérifier les noms de champs, la gestion des null et les horodatages sans écrire de consommateur.
Limites de taille et envoi par lots
Surveillez la taille des messages. Le serveur MQTT public gratuit accepte jusqu'à 64 Ko par message et le plan gratuit Serverless jusqu'à 256 Ko. Une mesure isolée est très loin de ces limites, mais si vous regroupez des données, par exemple en envoyant une minute d'échantillons sous forme de tableau, vérifiez len(payload) avant de publier et découpez les gros lots. Un autre moyen de réduire le trafic consiste à ne publier que lorsqu'une valeur varie au-delà d'un seuil défini, avec un heartbeat périodique pour que les consommateurs distinguent un capteur silencieux d'un capteur hors service. Cette approche réduit souvent fortement le volume de messages pour des valeurs qui évoluent lentement, comme la température d'une pièce, sans perdre d'information utile.
Pour aller plus loin
Stockez ces payloads avec le tutoriel MQTT vers SQLite en Python, dont les requêtes json_extract fonctionnent directement avec ce format, ou tracez-les en direct avec le tableau de bord MQTT Streamlit. Pour les services asynchrones, le même code de payload s'intègre dans un client aiomqtt asyncio. Le panorama des bibliothèques clientes MQTT montre comment d'autres langages parsent le même JSON, et la documentation du module json de Python couvre toutes les options de l'encodeur.
Démarrez avec ConnectMQ
ConnectMQ est un broker MQTT managé qui prend en charge MQTT 5 pour des propriétés comme le content type, avec un explorateur de topics en direct pour inspecter les payloads JSON et TLS sur le port 8883. Prototypez sur broker.connectmq.com, puis migrez vos appareils vers un broker MQTT cloud privé. Inscrivez-vous au plan gratuit Serverless, sans carte bancaire.
Questions fréquentes
Comment envoyer du JSON via MQTT en Python ?
Sérialisez vos données avec json.dumps(), encodez-les en octets UTF-8 et passez-les comme payload à client.publish(). Côté réception, appelez json.loads(msg.payload).
L'horodatage doit-il figurer dans le payload ou être ajouté par le consommateur ?
Incluez un horodatage de mesure en UTC dans le payload lorsque l'appareil dispose d'une horloge fiable. Dans le cas contraire, laissez le consommateur enregistrer l'heure de réception et omettez ts.
Pourquoi mon payload JSON contient-il NaN ?
Par défaut, json.dumps de Python écrit NaN pour les valeurs flottantes NaN, ce qui est du JSON invalide. Convertissez NaN en None avant l'encodage et définissez allow_nan=False pour intercepter celles qui restent.
Où placer la version du schéma, dans le payload ou dans le topic ?
Les deux fonctionnent. Un champ v dans le payload est plus simple et garde des topics stables ; un segment de version dans le topic permet aux consommateurs de ne s'abonner qu'aux versions qu'ils comprennent.
