MQTT Contract Testing with JSON Schema — Payload Validation

An MQTT broker routes bytes; it never checks what is inside them. When a firmware team renames temp to temperature, the publish succeeds, the broker delivers it, and the dashboard silently shows nothing. Contract testing MQTT messages with JSON Schema makes that agreement explicit: the payload format lives in a schema file, publishers prove in tests that they produce valid messages, consumers prove they accept them, and the ingestion service rejects anything that does not match.

This guide shows how to write a payload schema, validate it in Python and Node.js, wire it into producer and consumer tests, validate at runtime, and evolve the schema without breaking subscribers.

Why contracts matter more in MQTT

In HTTP, a bad request fails at the server with a 400 and the caller sees it. In publish/subscribe, the publisher never learns that a subscriber choked. The failure shows up later, somewhere else, often as missing data rather than an error.

Without a contract	With a contract
Field rename breaks consumers silently	CI fails on the producer's pull request
Each consumer re-implements type checks	One schema, generated validators
"Which firmware sends what?" is tribal knowledge	Schema files are the documentation
Bad messages crash workers	Bad messages go to a dead-letter topic

Good contracts start with good payloads. See MQTT payload design best practices for field naming, units and timestamps.

Write the schema

Keep schemas in a shared directory (or repository) that both device and backend code depend on. One file per message type and major version:

contracts/telemetry.v1.json

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "https://example.com/contracts/telemetry.v1.json",
  "title": "Telemetry v1",
  "type": "object",
  "required": ["schema", "device", "ts", "temp_c"],
  "properties": {
    "schema": { "const": "telemetry/v1" },
    "device": { "type": "string", "pattern": "^[a-z0-9-]{1,64}$" },
    "ts": { "type": "string", "format": "date-time" },
    "temp_c": { "type": "number", "minimum": -60, "maximum": 150 },
    "battery": { "type": "integer", "minimum": 0, "maximum": 100 }
  },
  "additionalProperties": true
}

Two design decisions matter here:

additionalProperties: true lets producers add optional fields without breaking old consumers. Consumers should ignore what they do not know.
The schema field makes every message self-describing, so a consumer can pick the right validator, and a dead-letter entry tells you exactly which contract it violated.

Validate in Python

The jsonschema package supports draft 2020-12. Note that format keywords are only checked when you pass a format checker, and date-time needs the optional rfc3339-validator package (pip install "jsonschema[format]" installs it).

contracts.py

import json
import pathlib
from jsonschema import Draft202012Validator

CONTRACTS = pathlib.Path(__file__).parent / "contracts"

def load_validator(name: str) -> Draft202012Validator:
    schema = json.loads((CONTRACTS / f"{name}.json").read_text())
    Draft202012Validator.check_schema(schema)        # fail fast on a broken schema
    return Draft202012Validator(schema, format_checker=Draft202012Validator.FORMAT_CHECKER)

VALIDATORS = {"telemetry/v1": load_validator("telemetry.v1")}

def contract_errors(payload: bytes) -> list[str]:
    try:
        doc = json.loads(payload)
    except ValueError as exc:
        return [f"not JSON: {exc}"]
    validator = VALIDATORS.get(doc.get("schema")) if isinstance(doc, dict) else None
    if validator is None:
        return ["unknown or missing schema field"]
    return [
        f"{'/'.join(map(str, e.absolute_path)) or '<root>'}: {e.message}"
        for e in validator.iter_errors(doc)
    ]

iter_errors returns every violation instead of stopping at the first, which makes test failures and dead-letter logs far more useful.

Producer tests: prove you send valid messages

The producer test calls the real code that builds the payload (firmware encoder, simulator, gateway) and validates the output. No broker needed:

test_producer_contract.py

import json
from contracts import contract_errors
from device.telemetry import build_payload   # your real encoder

def test_telemetry_payload_matches_contract():
    payload = build_payload(device="sensor-01", temp_c=21.5, battery=88)
    assert contract_errors(payload) == []

def test_minimal_payload_matches_contract():
    payload = build_payload(device="sensor-01", temp_c=-4.0, battery=None)
    assert contract_errors(payload) == []

Then add one test that goes through the broker, so you also catch encoding problems in the MQTT layer. Publish with MQTT 5 properties that describe the content:

test_contract_over_broker.py

import os, queue, uuid
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties
from contracts import contract_errors
from device.telemetry import build_payload

HOST = "broker.connectmq.com"

def make_client():
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
    c.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    c.tls_set()
    c.connect(HOST, 8883)
    c.loop_start()
    return c

def test_published_message_matches_contract():
    topic = f"testmqtt/contracts/{uuid.uuid4().hex[:8]}/telemetry"
    received = queue.Queue()

    sub = make_client()
    sub.on_message = lambda c, u, m: received.put(m)
    sub.subscribe(topic, qos=1)

    props = Properties(PacketTypes.PUBLISH)
    props.ContentType = "application/json"
    props.UserProperty = [("schema", "telemetry/v1")]

    pub = make_client()
    pub.publish(topic, build_payload(device="sensor-01", temp_c=21.5, battery=88),
                qos=1, properties=props).wait_for_publish()

    msg = received.get(timeout=10)
    assert contract_errors(msg.payload) == []
    assert msg.properties.ContentType == "application/json"
    for c in (pub, sub):
        c.loop_stop(); c.disconnect()

In a real suite, wait for on_subscribe before publishing (or use a fixture as in MQTT integration testing). Content type and user properties are MQTT 5 features; on MQTT 3.1.1, rely on the schema field in the payload.

Consumer tests: prove you accept valid messages

Consumer tests run the opposite direction. Keep a folder of example payloads, validated against the schema in CI, and feed them to the consumer's handler:

test_consumer_contract.py

import pathlib, pytest
from contracts import contract_errors
from backend.ingest import handle_telemetry   # your real handler

EXAMPLES = sorted(pathlib.Path("contracts/examples/telemetry.v1").glob("*.json"))

@pytest.mark.parametrize("path", EXAMPLES, ids=lambda p: p.name)
def test_example_is_valid(path):
    assert contract_errors(path.read_bytes()) == []

@pytest.mark.parametrize("path", EXAMPLES, ids=lambda p: p.name)
def test_consumer_accepts_example(path):
    reading = handle_telemetry(path.read_bytes())
    assert reading.device

The example files become the shared fixtures: when the producer team adds a field, they add an example, and every consumer's test suite runs against it.

Validate at runtime with Ajv

Tests protect the code you ship. Devices in the field still run whatever firmware they have. Validate at the ingestion edge and route violations to a dead-letter topic. In Node.js with MQTT.js 5 and Ajv 8:

ingest.ts

import { readFileSync } from "node:fs";
import mqtt from "mqtt";
import Ajv2020 from "ajv/dist/2020.js";
import addFormats from "ajv-formats";

const ajv = new Ajv2020({ allErrors: true });
addFormats(ajv);
const validate = ajv.compile(JSON.parse(readFileSync("contracts/telemetry.v1.json", "utf8")));

const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  protocolVersion: 5,
});

client.on("connect", () => client.subscribe("testmqtt/contracts/+/telemetry", { qos: 1 }));

client.on("message", (topic, payload) => {
  let doc: unknown;
  try {
    doc = JSON.parse(payload.toString("utf8"));
  } catch {
    doc = undefined;
  }
  if (doc === undefined || !validate(doc)) {
    const reason = doc === undefined ? "not JSON" : ajv.errorsText(validate.errors);
    client.publish("testmqtt/contracts/deadletter", payload, {
      qos: 1,
      properties: { userProperties: { reason, topic } },
    });
    return;
  }
  // doc is valid: store it
});

Compile the schema once at startup; ajv.compile is the expensive step, and the compiled function validates each message quickly.

Versioning without breaking subscribers

Classify every schema change before you merge it:

Change	Compatible?	What to do
Add an optional field	Yes	Same major version, add an example
Widen a range or enum	Usually	Check consumers that switch on the values
Make an optional field required	No for old producers	New major version
Rename or remove a field	No	New major version, run both in parallel
Change a type or unit	No	New major version and a new field name

Then test compatibility mechanically: validate all v1 examples against the v2 schema (can new consumers read old devices?) and all v2 examples against the v1 schema (can old consumers tolerate new devices?). Both checks are a few lines on top of the parametrized tests above.

For a breaking change, signal the version in the schema field, in a topic level (.../telemetry/v2) or in an MQTT 5 user property, and run both validators until no device publishes v1 any more. A topic-level version makes it easy to watch with a wildcard: subscribe to testmqtt/contracts/+/telemetry in the online MQTT client and check which versions are still live.

Inspect contract test traffic

Checklist

One JSON Schema per message type and major version, in a shared location.
A schema field (or user property) on every message.
Producer tests validate real encoder output; consumer tests run on shared examples.
Runtime validation at ingestion, with a dead-letter topic and the error text.
Cross-version compatibility tests in CI (GitHub Actions example).
Combine with fuzz testing so that invalid input is rejected, not just valid input accepted.

A schema file costs an afternoon. It replaces the silent dashboard gap after a firmware release with a failing test on the pull request that caused it, and that is the cheapest place to fix a broken contract.

Frequently asked questions

Why do MQTT systems need contract tests?

MQTT brokers do not inspect payloads, so a publisher can change a field name or type and every subscriber breaks at runtime with no error at publish time. A shared JSON Schema turns that implicit agreement into a file both sides test against.

Should I validate payloads at runtime or only in tests?

Both. Tests catch contract changes before deployment. Runtime validation at the ingestion edge catches devices running old or buggy firmware, and routes rejected messages to a dead-letter topic instead of crashing the consumer.

How do I version an MQTT payload schema?

Make additive changes (new optional fields) within a major version and keep consumers tolerant of unknown fields. For breaking changes, publish a new major version, signal it in a field, a topic level or an MQTT 5 user property, and run both versions side by side until all consumers have migrated.

Contract Testing MQTT Messages with JSON Schema