Type-Safe MQTT in TypeScript with MQTT.js and Zod

Type-safe MQTT in TypeScript means two things: the compiler stops you from publishing the wrong payload to the wrong topic, and every incoming message is validated before your code treats it as typed data. MQTT.js 5 ships its own types, but payloads arrive as a Buffer and topics are plain strings. This guide closes that gap with a typed topic map, payload types inferred from zod schemas and a small wrapper around the MQTT.js client.

The examples use MQTT.js 5, zod 4 and TypeScript 5 in strict mode, against broker.connectmq.com with topics under testmqtt/ts/....

Install

npm install mqtt zod
npm install -D typescript @types/node

Where the types break down

Plain MQTT.js code compiles happily even when it is wrong:

untyped.ts

client.publish("testmqtt/ts/devices/dev-1/telemtry", JSON.stringify({ temp: "21" }));

client.on("message", (topic, payload) => {
  const data = JSON.parse(payload.toString()); // any
  console.log(data.temperature.toFixed(1));     // crashes at runtime
});

A misspelled topic silently goes nowhere, a string where a number was expected reaches the dashboard, and JSON.parse returns any, which switches type checking off for everything downstream. There are two separate problems: outgoing messages need compile-time checks, incoming messages need runtime checks, because the broker forwards whatever bytes any client sends.

Schemas first, types derived

Write each payload once as a zod schema and let TypeScript infer the type from it. That way the runtime check and the static type cannot drift apart:

src/schemas.ts

import { z } from "zod";

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(["reboot", "identify", "set-interval"]),
  intervalSec: z.number().int().positive().optional(),
});

export const Status = z.object({
  online: z.boolean(),
  firmware: z.string().optional(),
});

export type Telemetry = z.infer<typeof Telemetry>;

Keep the schemas in a package that firmware simulators, backend services and tests can all import. Field naming and versioning are covered in MQTT payload design.

A typed topic map

Next, describe every topic once: its pattern with named parameters and its schema. Template literal types can pull the parameter names out of the pattern string, so the compiler knows that telemetry needs a deviceId:

src/topics.ts

import { z } from "zod";
import { Command, Status, Telemetry } from "./schemas";

type ParamNames<T extends string> =
  T extends `${string}{${infer P}}${infer Rest}` ? P | ParamNames<Rest> : never;
type Params<T extends string> = Record<ParamNames<T>, string>;

export const topics = {
  telemetry: { pattern: "testmqtt/ts/devices/{deviceId}/telemetry", schema: Telemetry },
  command: { pattern: "testmqtt/ts/devices/{deviceId}/command", schema: Command },
  status: { pattern: "testmqtt/ts/devices/{deviceId}/status", schema: Status },
} as const satisfies Record<string, { pattern: string; schema: z.ZodType }>;

export type TopicKey = keyof typeof topics;
export type PayloadOf<K extends TopicKey> = z.infer<(typeof topics)[K]["schema"]>;
export type ParamsOf<K extends TopicKey> = Params<(typeof topics)[K]["pattern"]>;

export function topicFor<K extends TopicKey>(key: K, params: ParamsOf<K>): string {
  return topics[key].pattern.replace(/\{(\w+)\}/g, (_, name: string) => {
    const value = (params as Record<string, string>)[name];
    if (!value || /[+#\/]/.test(value)) {
      throw new Error(`invalid topic parameter ${name}=${value}`);
    }
    return value;
  });
}

export function filterFor(key: TopicKey): string {
  return topics[key].pattern.replace(/\{\w+\}/g, "+");
}

export function matchTopic(topic: string) {
  for (const key of Object.keys(topics) as TopicKey[]) {
    const names: string[] = [];
    const source = topics[key].pattern.replace(/\{(\w+)\}/g, (_, name: string) => {
      names.push(name);
      return "([^/]+)";
    });
    const m = new RegExp(`^${source}$`).exec(topic);
    if (m) {
      const params = Object.fromEntries(names.map((n, i) => [n, m[i + 1]]));
      return { key, params };
    }
  }
  return null;
}

Three details are worth noting:

as const satisfies … keeps the literal pattern strings (so ParamNames works) while still checking that each entry has a pattern and a schema.
topicFor rejects parameter values containing /, + or #. Without that check, a device ID such as dev-1/# taken from user input would turn a publish into something you did not intend.
filterFor("telemetry") turns {deviceId} into +, giving the subscription filter testmqtt/ts/devices/+/telemetry. Patterns here contain no regex special characters; escape them if yours do. You can check filters against sample topics with the topic matcher.

A typed wrapper around MqttClient

The wrapper owns the single message listener. It maps each incoming topic back to a key, parses JSON, validates with the matching schema and only then calls the typed handlers:

src/typed-mqtt.ts

import type { IClientPublishOptions, MqttClient } from "mqtt";
import type { z } from "zod";
import {
  filterFor, matchTopic, topicFor, topics,
  type ParamsOf, type PayloadOf, type TopicKey,
} from "./topics";

type AnyHandler = (payload: unknown, params: Record<string, string>) => void;

export class TypedMqtt {
  private readonly client: MqttClient;
  private readonly handlers = new Map<TopicKey, AnyHandler[]>();

  constructor(client: MqttClient) {
    this.client = client;
    client.on("message", (topic, buf) => this.dispatch(topic, buf));
  }

  async publish<K extends TopicKey>(
    key: K,
    params: ParamsOf<K>,
    payload: PayloadOf<K>,
    opts: IClientPublishOptions = { qos: 1 },
  ): Promise<void> {
    const schema: z.ZodType = topics[key].schema;
    const data = schema.parse(payload); // strips unknown fields, catches casts
    await this.client.publishAsync(topicFor(key, params), JSON.stringify(data), opts);
  }

  async on<K extends TopicKey>(
    key: K,
    handler: (payload: PayloadOf<K>, params: ParamsOf<K>) => void,
    qos: 0 | 1 | 2 = 1,
  ): Promise<void> {
    const list = this.handlers.get(key) ?? [];
    list.push(handler as unknown as AnyHandler); // the one place types are erased
    this.handlers.set(key, list);
    await this.client.subscribeAsync(filterFor(key), { qos });
  }

  private dispatch(topic: string, buf: Buffer): void {
    const hit = matchTopic(topic);
    if (!hit) return;

    let json: unknown;
    try {
      json = JSON.parse(buf.toString("utf8"));
    } catch {
      console.warn(`[mqtt] not JSON on ${topic}`);
      return;
    }

    const schema: z.ZodType = topics[hit.key].schema;
    const result = schema.safeParse(json);
    if (!result.success) {
      console.warn(`[mqtt] invalid payload on ${topic}`, result.error.issues);
      return;
    }
    for (const h of this.handlers.get(hit.key) ?? []) h(result.data, hit.params);
  }
}

Invalid messages are logged and dropped instead of throwing inside the event handler, where an exception would surface far from its cause. In production, count them or forward them to a dead-letter topic so you notice a misbehaving firmware version. The common failure modes are listed in MQTT JSON parse errors.

Using it

src/main.ts

import mqtt from "mqtt";
import { TypedMqtt } from "./typed-mqtt";

const client = await mqtt.connectAsync("mqtts://broker.connectmq.com:8883", {
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
});
const bus = new TypedMqtt(client);

await bus.on("telemetry", (t, { deviceId }) => {
  // t is { deviceId: string; temperature: number; humidity: number; ts: number }
  console.log(deviceId, t.temperature.toFixed(1), `${t.humidity}%`);
});

await bus.publish(
  "telemetry",
  { deviceId: "dev-1" },
  { deviceId: "dev-1", temperature: 21.4, humidity: 48, ts: Date.now() },
);

await bus.publish("command", { deviceId: "dev-1" }, { action: "set-interval", intervalSec: 30 });

Now the mistakes from the first example fail at compile time:

Mistake	Compiler error
`bus.publish("telemtry", …)`	`Argument of type '"telemtry"' is not assignable to parameter of type '"telemetry" \| "command" \| "status"'`
`{ temperature: "21" }`	`Type 'string' is not assignable to type 'number'`
`{ action: "format" }`	`Type '"format"' is not assignable to type '"reboot" \| "identify" \| "set-interval"'`
`bus.publish("status", {}, …)`	`Property 'deviceId' is missing in type '{}' but required in type 'ParamsOf<"status">'`

Add @ts-expect-error lines for these cases to a type test file and run tsc --noEmit in CI; if someone loosens a schema, the build fails.

MQTT 5 extras

With protocolVersion: 5, you can put the schema version in the publish properties so consumers can pick the right parser before validating:

Publish with MQTT 5 properties

await client.publishAsync("testmqtt/ts/devices/dev-1/telemetry", JSON.stringify(data), {
  qos: 1,
  properties: {
    contentType: "application/json",
    userProperties: { schema: "telemetry/v2" },
  },
});

The receiving side reads them from the third argument of the message handler (packet.properties?.userProperties). The MQTT code generator produces a matching connection snippet for other languages if devices are not written in TypeScript.

Testing the validation

Run main.ts with npx tsx src/main.ts, then open the online MQTT client, connect to broker.connectmq.com and publish three messages to testmqtt/ts/devices/dev-9/telemetry:

{"deviceId":"dev-9","temperature":19.5,"humidity":40,"ts":1760000000000} — the handler logs it.
{"deviceId":"dev-9","temperature":"hot"} — you see invalid payload with zod's issue list (expected number, received string and the missing fields).
not json — you see not JSON on testmqtt/ts/devices/dev-9/telemetry.

The same three cases make a good unit test: hand TypedMqtt a fake client, emit message events and assert which handlers ran, as shown in testing MQTT.js with Jest and Vitest.

Send test payloads to the typed client

A typed topic map plus schema validation at the boundary turns MQTT from "any bytes on any string" into an interface you can refactor with confidence. Define topics and schemas once, derive every type from them, validate everything that arrives, and let the compiler catch the rest.

Frequently asked questions

Does MQTT.js have TypeScript types?

Yes. MQTT.js 5 is written in TypeScript and ships its own declarations, so MqttClient, IClientOptions and IClientPublishOptions are available without an @types package. Payloads are typed as Buffer, though; typing their contents is up to you.

Why validate MQTT payloads at runtime if I already use TypeScript?

TypeScript types disappear at compile time and the broker accepts any bytes on any topic. Firmware bugs, old device versions and manual test messages all arrive as untyped Buffers. A zod schema checks each message once at the boundary so the rest of your code can trust the type.

Can TypeScript check MQTT wildcard topics?

It can check the topics you build, using template literal types, so a typo in a topic key or a missing parameter is a compile error. Matching incoming topics against + and # filters still happens at runtime, which is why the wrapper maps each message back to a typed topic key.

Type-Safe MQTT in TypeScript