Define your AMQP contracts once — get type safety, autocompletion, and runtime validation everywhere.
- 🔒 End-to-end type safety — TypeScript knows your message shapes
- 🔄 Reliable retry — Built-in exponential backoff with Dead Letter Queue support
- 📄 AsyncAPI compatible — Generate documentation from your contracts
// contract.ts
import {
defineContract,
defineEventConsumer,
defineEventPublisher,
defineExchange,
defineMessage,
defineQueue,
defineQueueBinding,
} from "@amqp-contract/contract";
import { z } from "zod";
const ordersExchange = defineExchange("orders");
const ordersDlx = defineExchange("orders-dlx");
// Every consumed queue needs a dead-letter exchange (or an explicit
// `onPoison: "drop"`), and that exchange must route somewhere — defineContract
// rejects a contract that would silently lose rejected messages.
const orderQueue = defineQueue("order-processing", { deadLetter: { exchange: ordersDlx } });
const orderDlq = defineQueue("order-processing-dlq");
const orderMessage = defineMessage(z.object({ orderId: z.string(), amount: z.number() }));
const orderCreated = defineEventPublisher(ordersExchange, orderMessage, {
routingKey: "order.created",
});
export const contract = defineContract({
publishers: { orderCreated },
consumers: { processOrder: defineEventConsumer(orderCreated, orderQueue) },
queues: { orderDlq },
bindings: { orderDlq: defineQueueBinding(orderDlq, ordersDlx, { routingKey: "#" }) },
});Then use that contract — the worker consumes, the client publishes:
import { TypedAmqpClient } from "@amqp-contract/client";
import { TypedAmqpWorker } from "@amqp-contract/worker";
import { OkAsync } from "unthrown";
import { contract } from "./contract.js";
const worker = await TypedAmqpWorker.create({
contract,
handlers: {
processOrder: ({ input: { payload } }) => {
console.log(payload.orderId); // typed from the schema
return OkAsync(undefined);
},
},
urls: ["amqp://localhost"],
}).getOrThrow();
const client = await TypedAmqpClient.create({ contract, urls: ["amqp://localhost"] }).getOrThrow();
// Validated against the schema before it is sent. publish() returns an AsyncResult;
// .getOrThrow() awaits and unwraps it here — a service would .match() on it instead.
await client.publish("orderCreated", { orderId: "ORD-123", amount: 99.99 }).getOrThrow();
await client.close().get();
await worker.close().get();▶ For the full runnable version (including the RabbitMQ Docker command), follow the fifteen-minute tutorial.
Note
This README describes amqp-contract 3.x, which is published under the beta npm tag until 3.0 is stable — latest is still 2.x, and the root of the documentation site documents 2.x (the 3.x docs are at /beta/). Install 3.x with:
pnpm add @amqp-contract/contract@beta @amqp-contract/client@beta @amqp-contract/worker@beta unthrown zodRequires Node.js 22.22+.
pnpm add @amqp-contract/contract @amqp-contract/client @amqp-contract/worker unthrown zodunthrown is exposed in the public types (AsyncResult<void, HandlerError>), so consumers need it directly to construct handler results. zod can be swapped for any Standard Schema library (Valibot, ArkType, …).
Need a local RabbitMQ to try it against?
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4-management- Get Started — Get running in fifteen minutes
- Core Concepts — Understand the fundamentals
- Examples — Real-world usage patterns
| Package | Description |
|---|---|
| @amqp-contract/contract | Contract builder and type definitions |
| @amqp-contract/client | Type-safe client for publishing |
| @amqp-contract/worker | Type-safe worker with retry support |
| @amqp-contract/core | Shared runtime: topology setup, connections, telemetry |
| @amqp-contract/asyncapi | AsyncAPI 3.1 generator |
| @amqp-contract/testing | Vitest utilities with a RabbitMQ testcontainer |
See CONTRIBUTING.md.
MIT