Type-Safe MQTT TypeScript Client: Typed Topics and Validated Payloads

Writing MQTT in TypeScript gives you autocompletion and compile-time checks, but only if your topics and payloads are actually typed. Out of the box, MQTT.js hands you a topic string and a Buffer, and everything after that is guesswork. In this tutorial you will define one contract file for topics, QoS and payload schemas, wrap MQTT.js in a small typed client, and validate every message at runtime, so publishers and subscribers cannot drift apart without the compiler or the validator telling you.

Why MQTT Code Breaks Silently

MQTT brokers do not know or care what is inside a payload. That flexibility is a strength, but it means the most common bugs in MQTT systems never raise an error:

Static types catch the first two at compile time when both sides share the same code. Runtime validation catches the last two, because data from devices, other teams or older firmware never passed through your compiler. You need both.

Project Setup

mkdir mqtt-typescript-demo && cd mqtt-typescript-demo
npm init -y
npm pkg set type=module
npm install mqtt zod
npm install -D typescript tsx @types/node
npx tsc --init --strict --target es2022 --module nodenext --moduleResolution nodenext

tsx runs TypeScript files directly, and tsc --noEmit type-checks the project. MQTT.js v5 ships its own type definitions. zod provides schemas that double as TypeScript types, so you write each payload shape once.

Step 1: The Contract File

Create src/contracts.ts. This is the single source of truth that every publisher and subscriber imports:

import { z } from "zod";

export const PREFIX = process.env.TOPIC_PREFIX ?? "connectmq/demo/ts-7f3a";

export const Telemetry = z.object({
  deviceId: z.string(),
  temperature: z.number(),
  humidity: z.number().min(0).max(100),
  ts: z.number().int(),
});

export const Command = z.object({
  action: z.enum(["on", "off", "reboot"]),
  requestId: z.string().uuid(),
});

export const Status = z.enum(["online", "offline"]);

interface Contract<S extends z.ZodType, A extends unknown[]> {
  filter: string;
  topic: (...args: A) => string;
  schema: S;
  qos: 0 | 1 | 2;
  retain: boolean;
}

export const contracts = {
  telemetry: {
    filter: `${PREFIX}/devices/+/telemetry`,
    topic: (deviceId: string) => `${PREFIX}/devices/${deviceId}/telemetry`,
    schema: Telemetry,
    qos: 0,
    retain: false,
  },
  command: {
    filter: `${PREFIX}/devices/+/cmd`,
    topic: (deviceId: string) => `${PREFIX}/devices/${deviceId}/cmd`,
    schema: Command,
    qos: 1,
    retain: false,
  },
  status: {
    filter: `${PREFIX}/devices/+/status`,
    topic: (deviceId: string) => `${PREFIX}/devices/${deviceId}/status`,
    schema: Status,
    qos: 1,
    retain: true,
  },
} as const satisfies Record<string, Contract<z.ZodType, any[]>>;

export type Contracts = typeof contracts;
export type ContractKey = keyof Contracts;
export type PayloadOf<K extends ContractKey> = z.infer<Contracts[K]["schema"]>;
export type TopicArgs<K extends ContractKey> = Parameters<Contracts[K]["topic"]>;

The as const satisfies combination does two jobs. satisfies checks every entry against the Contract shape, so forgetting a schema or using QoS 3 is a compile error. as const keeps the exact literal types, so PayloadOf<"command"> resolves to { action: "on" | "off" | "reboot"; requestId: string } rather than a vague object.

Topic builders are functions rather than strings with placeholders, which means contracts.command.topic(deviceId) is checked for the right number and type of arguments, and nobody ever concatenates topic strings by hand again.

Step 2: A Typed MQTT Client

Create src/typed-client.ts:

import mqtt, { IClientOptions, MqttClient } from "mqtt";
import { contracts, ContractKey, PayloadOf, TopicArgs } from "./contracts.js";

type Handler<K extends ContractKey> = (payload: PayloadOf<K>, topic: string) => void;

export function topicMatches(filter: string, topic: string): boolean {
  const f = filter.split("/");
  const t = topic.split("/");
  for (let i = 0; i < f.length; i++) {
    if (f[i] === "#") return true;
    if (i >= t.length) return false;
    if (f[i] !== "+" && f[i] !== t[i]) return false;
  }
  return f.length === t.length;
}

export class TypedMqttClient {
  private readonly client: MqttClient;
  private readonly handlers = new Map<ContractKey, Set<Handler<any>>>();

  constructor(
    url: string,
    options: IClientOptions,
    private readonly onInvalid: (topic: string, error: unknown) => void = (topic, error) =>
      console.warn(`Invalid payload on ${topic}:`, error)
  ) {
    this.client = mqtt.connect(url, { protocolVersion: 5, reconnectPeriod: 2000, ...options });

    this.client.on("connect", () => {
      for (const key of this.handlers.keys()) {
        const c = contracts[key];
        this.client.subscribe(c.filter, { qos: c.qos });
      }
    });

    this.client.on("message", (topic, buffer) => this.dispatch(topic, buffer));
    this.client.on("error", (err) => console.error("MQTT error:", err.message));
  }

  async publish<K extends ContractKey>(key: K, args: TopicArgs<K>, payload: PayloadOf<K>): Promise<void> {
    const c = contracts[key];
    const validated = c.schema.parse(payload); // never send what we would reject ourselves
    const topic = (c.topic as (...a: unknown[]) => string)(...args);
    await this.client.publishAsync(topic, JSON.stringify(validated), { qos: c.qos, retain: c.retain });
  }

  subscribe<K extends ContractKey>(key: K, handler: Handler<K>): () => void {
    let set = this.handlers.get(key);
    if (!set) {
      set = new Set();
      this.handlers.set(key, set);
      if (this.client.connected) {
        this.client.subscribe(contracts[key].filter, { qos: contracts[key].qos });
      }
    }
    set.add(handler);
    return () => {
      set!.delete(handler);
      if (set!.size === 0) {
        this.handlers.delete(key);
        this.client.unsubscribe(contracts[key].filter);
      }
    };
  }

  async close(): Promise<void> {
    await this.client.endAsync();
  }

  private dispatch(topic: string, buffer: Buffer): void {
    for (const [key, set] of this.handlers) {
      const c = contracts[key];
      if (!topicMatches(c.filter, topic)) continue;
      let raw: unknown;
      try {
        raw = JSON.parse(buffer.toString());
      } catch (err) {
        this.onInvalid(topic, err);
        continue;
      }
      const result = c.schema.safeParse(raw);
      if (!result.success) {
        this.onInvalid(topic, result.error.issues);
        continue;
      }
      for (const handler of set) handler(result.data, topic);
    }
  }
}

Three details make this robust:

• Outgoing payloads are validated too. TypeScript already checks literals, but values built at runtime (from sensors, forms or other services) can still violate constraints like humidity <= 100. schema.parse throws before a bad message leaves the process.
• Incoming payloads use safeParse. A malformed message from one device is reported through onInvalid instead of crashing the subscriber or flowing into business logic as a wrongly typed object.
• Subscriptions are restored on every connect, using each contract's QoS, so reconnects are invisible to calling code.

Step 3: Use It

Create src/demo.ts:

import { randomUUID } from "node:crypto";
import { contracts } from "./contracts.js";
import { TypedMqttClient } from "./typed-client.js";

const mq = new TypedMqttClient("mqtts://broker.connectmq.com:8883", {
  clientId: `ts-demo-${randomUUID().slice(0, 8)}`,
});

// payload is fully typed: { deviceId: string; temperature: number; humidity: number; ts: number }
mq.subscribe("telemetry", (payload, topic) => {
  console.log(`${payload.deviceId}: ${payload.temperature.toFixed(1)} C (${topic})`);
});

mq.subscribe("command", (cmd) => {
  console.log(`Command ${cmd.action} (${cmd.requestId})`);
});

await mq.publish("status", ["sensor-42"], "online");

const timer = setInterval(() => {
  mq.publish("telemetry", ["sensor-42"], {
    deviceId: "sensor-42",
    temperature: 20 + Math.random() * 5,
    humidity: 40 + Math.random() * 10,
    ts: Date.now(),
  }).catch((err) => console.error("Rejected by schema:", err.message));
}, 2000);

setTimeout(async () => {
  await mq.publish("command", ["sensor-42"], { action: "reboot", requestId: randomUUID() });
}, 5000);

// These lines do not compile, which is the point:
// mq.publish("command", ["sensor-42"], { action: "explode", requestId: "x" });
// mq.publish("telemetry", [], { deviceId: "a", temperature: "21", humidity: 1, ts: 0 });

process.on("SIGINT", async () => {
  clearInterval(timer);
  await mq.publish("status", ["sensor-42"], "offline");
  await mq.close();
  process.exit(0);
});

console.log("Telemetry topic example:", contracts.telemetry.topic("sensor-42"));

Run it and type-check separately:

npx tsx src/demo.ts
npx tsc --noEmit

Uncomment the two "do not compile" lines and tsc reports an invalid enum value for action, a missing topic argument and a string where a number is required. That is exactly the class of bug that otherwise reaches production.

To see runtime validation in action, publish a broken payload from the command line to the public MQTT test broker:

mosquitto_pub -h broker.connectmq.com -p 1883 -i ts-bad-7f3a \
  -t connectmq/demo/ts-7f3a/devices/sensor-9/telemetry -m '{"deviceId":"sensor-9","temperature":"hot"}'

The demo logs an Invalid payload warning listing the failing fields, and your handler is never called. The public area is readable by anyone, so change ts-7f3a to your own random string.

Sharing Contracts Across Services and Teams

The contract file is most valuable when every component imports it:

1. Monorepo: put contracts.ts in a shared package used by backend services, the React dashboard and test tools.
2. Separate repositories: publish the contracts as a private npm package and version it. A breaking schema change becomes a major version bump that consumers opt into.
3. Non-TypeScript devices: generate JSON Schema from the zod schemas with a converter library and share that with firmware teams, so ESP32 or Python code can validate against the same definition.
4. Evolve safely: add new fields as optional first, deploy subscribers, then update publishers. For payloads that change shape completely, add a version segment to the topic.

Testing Your Contracts

Contracts are code, so test them like code. Three kinds of tests give the most value for the least effort:

1. Schema fixtures. Keep a folder of real payloads captured from devices, including old firmware versions, and assert that each one passes safeParse for its contract. When someone tightens a schema, the test shows immediately which deployed devices would start failing validation.
2. Topic round trips. For every contract, build a topic with topic() and assert that topicMatches(filter, builtTopic) is true. This catches the classic mistake of changing a builder without updating its filter.
3. Type-level tests. Files that should not compile, checked with // @ts-expect-error comments, prove that invalid payloads stay rejected. If someone loosens a type by accident, tsc flags the unused directive.

All three run in milliseconds and need no broker, so they belong in every pull request check rather than a nightly job. Add a single integration test against the public broker with a random prefix to verify the full publish, validate and dispatch path over a real connection.

Running Against a Private Broker

Switching from the public broker to production only changes the constructor. On ConnectMQ, create a broker in the portal and pass the generated credentials:

const mq = new TypedMqttClient(`mqtts://${process.env.MQTT_HOST}:8883`, {
  clientId: process.env.MQTT_CLIENT_ID!,   // YOUR_CLIENT_ID
  username: process.env.MQTT_USERNAME!,    // YOUR_USERNAME
  password: process.env.MQTT_PASSWORD!,    // YOUR_PASSWORD
});

A managed MQTT broker takes care of TLS and broker operations, while your typed contracts take care of what flows through it. The ConnectMQ portal's live topic explorer is a quick way to confirm topics generated by your builders match what you expect.

Related Tutorials

If you are new to the underlying client, start with MQTT.js in Node.js. The same contracts can back the observables in the Angular MQTT service. For other languages and their typing options, see the MQTT client libraries guide. The zod documentation covers schema features such as refinements and transforms you can add to your contracts.

Get Started with ConnectMQ

Run your type-safe MQTT services on a private broker in minutes. Sign up at app.connectmq.com, choose the free Serverless plan with no credit card required, create a broker and pass the credentials to TypedMqttClient. The free plan supports MQTT 5 and QoS up to 1, which covers every contract in this tutorial.

Frequently Asked Questions

Does MQTT.js support TypeScript?

Yes. MQTT.js v5 is written in TypeScript and ships its own declarations, so options, events and packets are typed. Payloads remain Buffer, which is why a contract layer like this one is useful.

Why validate at runtime if TypeScript already checks types?

TypeScript only checks code it compiles. Messages from devices, other services or older firmware arrive as untyped bytes, so runtime validation is the only way to guarantee the object your handler receives matches its declared type.

Can I use something other than zod?

Yes. Any validator that infers TypeScript types works, such as Valibot, TypeBox or ArkType. The contract structure stays the same; only the schema definitions and the parse calls change.

How do I type topics with wildcards?

Keep two fields per contract: a filter with + or # for subscribing, and a typed topic() builder for publishing. Use the filter only on the subscriber side and extract parameters from the topic string when a handler needs the device ID.