Envoyer du JSON via MQTT en Python

MQTT ne sait pas ce qu'est le JSON : un payload n'est qu'une suite d'octets. Pour envoyer du JSON via MQTT en Python, sérialisez votre objet avec json.dumps() avant publish() et appelez json.loads(msg.payload) dans on_message. Tout le reste de ce guide consiste à le faire en toute sécurité : valider ce qui arrive, survivre aux messages invalides, gérer les dates et garder un payload léger.

Tous les exemples utilisent paho-mqtt 2.x et le broker de test broker.connectmq.com.

L'aller-retour minimal

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)  # des bytes en entrée, un dict en sortie
    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()

Trois faits expliquent la plupart des bugs de débutant :

• json.dumps() renvoie un str ; paho encode les chaînes en UTF-8 avant l'envoi. Passer le dict lui-même échoue avec TypeError: payload must be a string, bytearray, int, float or None.
• msg.payload est toujours de type bytes. json.loads() accepte des bytes UTF-8, donc inutile d'appeler .decode() avant.
• JSON ne connaît ni tuples, ni sets, ni datetime, ni Decimal. Les tuples reviennent sous forme de listes ; les autres lèvent TypeError: Object of type datetime is not JSON serializable.

Vous pouvez observer les messages avec le client MQTT en ligne en vous abonnant à testmqtt/json-demo/#. Il affiche le payload en texte : vous voyez exactement ce que votre code a envoyé sur le réseau.

Un payload invalide ne doit jamais tuer votre abonné

Quiconque a le droit de publier sur un topic peut vous envoyer n'importe quoi : une simple chaîne, un message tronqué, du texte Latin-1 venant d'un vieil appareil. Si json.loads() lève une exception dans on_message, elle se propage et votre boucle réseau s'arrête. Interceptez les deux erreurs qui comptent :

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)

Le test isinstance compte plus qu'il n'y paraît : 42, "ok" et null sont du JSON valide, et un code qui fait data["temperature"] plantera dessus avec un TypeError. Les messages typiques que vous verrez dans les logs, comme Expecting value: line 1 column 1 (char 0) pour un payload vide ou non JSON, sont expliqués dans erreurs de parsing JSON MQTT.

Valider la structure avec pydantic

Intercepter les erreurs de syntaxe ne suffit pas : {"temperature": "hot"} se parse sans problème. Un schéma rend le contrat explicite, et pydantic v2 sait parser et valider des bytes en une seule étape :

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 est un vrai datetime, reading.temperature un float

Côté publication, le même modèle se sérialise lui-même, datetime compris :

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)

Partager un module de modèles entre publisher et abonné est la forme la moins coûteuse de contract testing : si quelqu'un renomme un champ, la suite de tests échoue avant les appareils.

Dates, décimaux et autres types gênants

Si vous n'utilisez pas pydantic, indiquez à json.dumps() comment traiter les types qu'il ne connaît pas :

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)

La ligne NaN est un vrai piège : Python produit sans broncher {"t": NaN}, mais JSON.parse en JavaScript, Go et la plupart des parsers stricts le rejettent. allow_nan=False le transforme en ValueError côté publisher, là où il est facile à corriger.

Encodage compact

Chaque octet compte sur les liaisons cellulaires et les appareils sur batterie, et MQTT vous facture le payload à chaque saut. Voici la même mesure encodée de quatre façons (tailles mesurées avec Python 3) :

Les trois premiers utilisent les clés complètes device_id, timestamp (une chaîne ISO avec microsecondes et décalage), temperature, humidity, battery et status. Le dernier est {"id":"sensor-42","ts":1792229400,"t":21.46,"h":48.2,"b":0.87}. Ce qui fait la différence :

• Les séparateurs compacts suppriment les espaces après , et :.
• Les clés courtes sont le gain le plus important pour les petits messages ; documentez-les dans votre schéma.
• Les timestamps Unix au lieu des chaînes ISO économisent environ 20 octets chacun.
• Arrondissez les floats à la précision réelle de votre capteur. 21.456789 transporte du bruit, pas de l'information.
• ensure_ascii=False conserve le texte non ASCII en UTF-8 ("Zürich") au lieu d'échappements ü, qui prennent 6 octets au lieu de 2.

Pour la vitesse et le confort, orjson est un remplaçant direct : sortie compacte par défaut, renvoie des bytes (que paho accepte directement) et sérialise datetime nativement.

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)

Vérifiez la taille complète du paquet, topic et en-têtes compris, avec le calculateur de taille de paquet MQTT. Si le JSON compact reste trop gros, les formats binaires sont l'étape suivante ; voir CBOR et MessagePack sur MQTT et payloads Protobuf.

Indiquer aux abonnés que c'est du JSON (MQTT 5)

MQTT 5 permet au publisher d'étiqueter le payload, pour que les outils génériques et les autres services n'aient pas à deviner :

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 = texte UTF-8

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

Côté réception, lisez msg.properties.ContentType (vérifiez d'abord avec hasattr, car la propriété n'existe que si le publisher l'a définie). Avec un payload format indicator à 1, un broker MQTT 5 peut rejeter les payloads qui ne sont pas de l'UTF-8 valide ; voir erreurs d'encodage UTF-8 MQTT.

Tester votre gestion du JSON

Avant de faire confiance à un abonné, envoyez-lui les messages qui cassent tout. Avec mosquitto_pub (ou le client en ligne) :

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'                          # pas du JSON
mosquitto_pub -h $H -t $T -n                                 # payload vide
mosquitto_pub -h $H -t $T -m '[1,2,3]'                        # JSON valide, mauvaise forme
mosquitto_pub -h $H -t $T -m '{"temperature":"hot"}'          # mauvais type
printf '{"city":"Z\xfcrich"}' | mosquitto_pub -h $H -t $T -s  # Latin-1, pas UTF-8

L'abonné doit enregistrer le premier message et journaliser un avertissement clair pour chacun des autres, tout en restant connecté. Mettez les mêmes cas dans un test unitaire qui appelle directement votre fonction on_message avec un faux objet message : vous détecterez les régressions sans broker.

Inspecter les payloads JSON dans le client en ligne

Le JSON sur MQTT est simple dès que vous traitez le payload comme des octets non fiables : sérialisez explicitement, parsez défensivement, validez contre un schéma et restez compact. Pour les questions de conception plus larges, comme les enveloppes, les timestamps et la structure des topics, poursuivez avec les bonnes pratiques de conception de payload MQTT.

Questions fréquentes