Enviar JSON por MQTT en Python

MQTT no sabe qué es JSON: un payload son solo bytes. Para enviar JSON por MQTT en Python, serializa tu objeto con json.dumps() antes de publish() y llama a json.loads(msg.payload) en on_message. Todo lo demás en esta guía trata de hacerlo de forma segura: validar lo que llega, sobrevivir a mensajes malos, manejar fechas y mantener el payload pequeño.

Todos los ejemplos usan paho-mqtt 2.x y el broker de pruebas broker.connectmq.com.

El viaje de ida y vuelta mínimo

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)  # entran bytes, sale un dict
    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()

Tres hechos explican la mayoría de los errores de principiante:

• json.dumps() devuelve un str; paho codifica los strings en UTF-8 antes de enviarlos. Pasar el dict directamente falla con TypeError: payload must be a string, bytearray, int, float or None.
• msg.payload siempre es bytes. json.loads() acepta bytes UTF-8, así que no necesitas .decode() antes.
• JSON no tiene tuplas, sets, datetime ni Decimal. Las tuplas vuelven como listas; los demás lanzan TypeError: Object of type datetime is not JSON serializable.

Puedes ver los mensajes con el cliente MQTT online suscribiéndote a testmqtt/json-demo/#. Muestra el payload como texto, así que ves exactamente lo que tu código puso en la red.

Que un payload malo nunca tumbe tu suscriptor

Cualquiera con permiso de publicación en un topic puede enviarte cualquier cosa: un string plano, un mensaje truncado, texto Latin-1 de un dispositivo antiguo. Si json.loads() lanza una excepción dentro de on_message, esta se propaga y tu bucle de red se detiene. Captura los dos errores que importan:

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)

La comprobación isinstance importa más de lo que parece: 42, "ok" y null son JSON válido, y el código que hace data["temperature"] fallará con ellos con un TypeError. Los mensajes típicos que verás en los logs, como Expecting value: line 1 column 1 (char 0) para un payload vacío o que no es JSON, se explican en errores de parseo JSON en MQTT.

Valida la estructura con pydantic

Capturar errores de sintaxis no basta: {"temperature": "hot"} se parsea sin problema. Un esquema hace explícito el contrato, y pydantic v2 puede parsear y validar bytes en un solo paso:

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 es un datetime real, reading.temperature un float

En el lado del publicador, el mismo modelo se serializa solo, incluido el 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)

Compartir un módulo de modelos entre publicador y suscriptor es la forma más barata de contract testing: si alguien renombra un campo, la suite de tests falla antes que los dispositivos.

Fechas, decimales y otros tipos incómodos

Si no usas pydantic, dile a json.dumps() cómo tratar los tipos que no conoce:

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 fila de NaN es una trampa real: Python produce sin quejarse {"t": NaN}, pero JSON.parse de JavaScript, Go y la mayoría de parsers estrictos lo rechazan. allow_nan=False lo convierte en un ValueError en el publicador, donde es fácil de corregir.

Codificación compacta

Cada byte cuenta en enlaces celulares y dispositivos con batería, y MQTT te cobra el payload en cada salto. Esta es la misma lectura codificada de cuatro formas (tamaños medidos con Python 3):

Las tres primeras usan las claves completas device_id, timestamp (un string ISO con microsegundos y offset), temperature, humidity, battery y status. La última es {"id":"sensor-42","ts":1792229400,"t":21.46,"h":48.2,"b":0.87}. Los cambios que marcan la diferencia:

• Separadores compactos eliminan los espacios tras , y :.
• Claves cortas son la mayor ganancia en mensajes pequeños; documéntalas en tu esquema.
• Timestamps Unix en lugar de strings ISO ahorran unos 20 bytes cada uno.
• Redondea los floats a la precisión que tu sensor realmente tiene. 21.456789 lleva ruido, no información.
• ensure_ascii=False mantiene el texto no ASCII como UTF-8 ("Zürich") en lugar de escapes ü, que ocupan 6 bytes en vez de 2.

Para velocidad y comodidad, orjson es un reemplazo directo: produce salida compacta por defecto, devuelve bytes (que paho acepta directamente) y serializa datetime de forma nativa.

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)

Comprueba el tamaño completo del paquete, con topic y cabeceras, en la calculadora de tamaño de paquetes MQTT. Si el JSON compacto sigue siendo demasiado grande, el siguiente paso son los formatos binarios; consulta CBOR y MessagePack sobre MQTT y payloads Protobuf.

Indica a los suscriptores que es JSON (MQTT 5)

MQTT 5 permite al publicador etiquetar el payload, para que las herramientas genéricas y otros servicios no tengan que adivinar:

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

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

En el lado receptor, lee msg.properties.ContentType (compruébalo antes con hasattr, porque solo existe si el publicador lo definió). Con el payload format indicator a 1, un broker MQTT 5 puede rechazar payloads que no sean UTF-8 válido; más sobre esto en errores de codificación UTF-8 en MQTT.

Prueba tu manejo de JSON

Antes de fiarte de un suscriptor, envíale los mensajes que rompen cosas. Con mosquitto_pub (o el cliente online):

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'                          # no es JSON
mosquitto_pub -h $H -t $T -n                                 # payload vacío
mosquitto_pub -h $H -t $T -m '[1,2,3]'                        # JSON válido, forma incorrecta
mosquitto_pub -h $H -t $T -m '{"temperature":"hot"}'          # tipo incorrecto
printf '{"city":"Z\xfcrich"}' | mosquitto_pub -h $H -t $T -s  # Latin-1, no UTF-8

El suscriptor debe guardar el primer mensaje y registrar un aviso claro para cada uno de los demás, sin desconectarse. Pon los mismos casos en un test unitario que llame directamente a tu función on_message con un objeto de mensaje falso, y detectarás regresiones sin necesidad de broker.

Inspecciona los payloads JSON en el cliente online

JSON sobre MQTT es sencillo en cuanto tratas el payload como bytes no fiables: serializa de forma explícita, parsea a la defensiva, valida contra un esquema y mantenlo compacto. Para las preguntas de diseño más amplias, como sobres, timestamps y estructura de topics, sigue con buenas prácticas de diseño de payloads MQTT.

Preguntas frecuentes