JSON über MQTT in Python senden

MQTT weiß nicht, was JSON ist: Ein Payload sind einfach Bytes. Um JSON über MQTT in Python zu senden, serialisierst du dein Objekt vor publish() mit json.dumps() und rufst in on_message json.loads(msg.payload) auf. Alles andere in diesem Guide dreht sich darum, das sicher zu tun: Eingehendes validieren, kaputte Nachrichten überleben, Datumswerte behandeln und den Payload klein halten.

Alle Beispiele nutzen paho-mqtt 2.x und den Test-Broker broker.connectmq.com.

Der minimale Roundtrip

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)  # Bytes rein, Dict raus
    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()

Drei Fakten erklären die meisten Anfängerfehler:

• json.dumps() liefert einen str; paho kodiert Strings vor dem Senden als UTF-8. Übergibst du das Dict selbst, gibt es TypeError: payload must be a string, bytearray, int, float or None.
• msg.payload ist immer bytes. json.loads() akzeptiert UTF-8-Bytes, ein .decode() vorher ist also unnötig.
• JSON kennt keine Tupel, Sets, datetime oder Decimal. Tupel kommen als Listen zurück; die anderen werfen TypeError: Object of type datetime is not JSON serializable.

Du kannst die Nachrichten mit dem Online-MQTT-Client beobachten, indem du testmqtt/json-demo/# abonnierst. Er zeigt den Payload als Text, du siehst also genau, was dein Code auf die Leitung gelegt hat.

Ein kaputter Payload darf den Subscriber nie umbringen

Jeder mit Publish-Rechten auf einem Topic kann dir alles schicken: einen nackten String, eine abgeschnittene Nachricht, Latin-1-Text von einem alten Gerät. Wirft json.loads() in on_message eine Exception, wird sie weitergereicht und dein Netzwerk-Loop stoppt. Fang die zwei Fehler ab, auf die es ankommt:

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)

Die isinstance-Prüfung ist wichtiger, als sie aussieht: 42, "ok" und null sind gültiges JSON, und Code mit data["temperature"] stürzt damit per TypeError ab. Die typischen Meldungen in den Logs, etwa Expecting value: line 1 column 1 (char 0) bei einem leeren oder Nicht-JSON-Payload, erklärt MQTT-JSON-Parse-Fehler.

Struktur mit pydantic validieren

Syntaxfehler abzufangen reicht nicht: {"temperature": "hot"} lässt sich problemlos parsen. Ein Schema macht den Vertrag explizit, und pydantic v2 kann Bytes in einem Schritt parsen und validieren:

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 ist ein echtes datetime, reading.temperature ein float

Auf der Publisher-Seite serialisiert sich dasselbe Model selbst, inklusive 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)

Ein gemeinsames Model-Modul für Publisher und Subscriber ist die günstigste Form von Contract Testing: Benennt jemand ein Feld um, schlägt die Testsuite fehl, bevor es die Geräte tun.

Datumswerte, Decimals und andere sperrige Typen

Ohne pydantic sagst du json.dumps(), wie es mit unbekannten Typen umgehen soll:

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)

Die NaN-Zeile ist eine echte Falle: Python erzeugt klaglos {"t": NaN}, aber JSON.parse in JavaScript, Go und die meisten strikten Parser lehnen das ab. allow_nan=False macht daraus einen ValueError beim Publisher, wo er leicht zu beheben ist.

Kompakte Kodierung

Auf Mobilfunkstrecken und bei Batteriegeräten zählt jedes Byte, und MQTT berechnet dir den Payload auf jedem Hop. Hier dieselbe Messung auf vier Arten kodiert (Größen mit Python 3 gemessen):

Die ersten drei nutzen die vollen Keys device_id, timestamp (ein ISO-String mit Mikrosekunden und Offset), temperature, humidity, battery und status. Die letzte ist {"id":"sensor-42","ts":1792229400,"t":21.46,"h":48.2,"b":0.87}. Was den Unterschied macht:

• Kompakte Separatoren entfernen die Leerzeichen nach , und :.
• Kurze Keys bringen bei kleinen Nachrichten am meisten; dokumentiere sie in deinem Schema.
• Unix-Timestamps statt ISO-Strings sparen jeweils etwa 20 Bytes.
• Runde Floats auf die Präzision, die dein Sensor wirklich hat. 21.456789 transportiert Rauschen, keine Information.
• ensure_ascii=False lässt Nicht-ASCII-Text als UTF-8 ("Zürich") statt als ü-Escape, das 6 statt 2 Bytes belegt.

Für Tempo und Komfort ist orjson ein direkter Ersatz: Es erzeugt standardmäßig kompakte Ausgabe, liefert bytes (die paho direkt akzeptiert) und serialisiert datetime nativ.

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)

Die vollständige Paketgröße inklusive Topic und Headern prüfst du mit dem MQTT-Paketgrößen-Rechner. Ist kompaktes JSON immer noch zu groß, sind Binärformate der nächste Schritt; siehe CBOR und MessagePack über MQTT und Protobuf-Payloads.

Subscribern sagen, dass es JSON ist (MQTT 5)

Mit MQTT 5 kann der Publisher den Payload kennzeichnen, damit generische Tools und andere Dienste nicht raten müssen:

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

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

Auf der Empfängerseite liest du msg.properties.ContentType (vorher mit hasattr prüfen, denn das Feld existiert nur, wenn der Publisher es gesetzt hat). Ist der Payload Format Indicator 1, darf ein MQTT-5-Broker Payloads ablehnen, die kein gültiges UTF-8 sind; mehr dazu unter MQTT-UTF-8-Kodierungsfehler.

Dein JSON-Handling testen

Bevor du einem Subscriber vertraust, schick ihm die Nachrichten, die Dinge kaputt machen. Mit mosquitto_pub (oder dem Online-Client):

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'                          # kein JSON
mosquitto_pub -h $H -t $T -n                                 # leerer Payload
mosquitto_pub -h $H -t $T -m '[1,2,3]'                        # gültiges JSON, falsche Form
mosquitto_pub -h $H -t $T -m '{"temperature":"hot"}'          # falscher Typ
printf '{"city":"Z\xfcrich"}' | mosquitto_pub -h $H -t $T -s  # Latin-1, kein UTF-8

Der Subscriber sollte die erste Nachricht speichern und für jede weitere eine klare Warnung loggen – und dabei verbunden bleiben. Pack dieselben Fälle in einen Unit-Test, der deine on_message-Funktion direkt mit einem Fake-Message-Objekt aufruft, dann fängst du Regressionen ganz ohne Broker ab.

JSON-Payloads im Online-Client ansehen

JSON über MQTT ist einfach, sobald du den Payload als nicht vertrauenswürdige Bytes behandelst: explizit serialisieren, defensiv parsen, gegen ein Schema validieren und kompakt halten. Für die größeren Designfragen wie Envelopes, Timestamps und Topic-Struktur geht es weiter mit Best Practices für MQTT-Payload-Design.

Häufig gestellte Fragen