queue-pilot

Queue Pilot

MCP server for message queue development — combines message inspection with JSON Schema validation. Supports RabbitMQ and Kafka.

Designed for integration projects where multiple teams communicate via message brokers: inspect queues/topics, view messages, and validate payloads against agreed-upon schemas — all from your AI assistant.

Features

• Multi-broker support — RabbitMQ and Apache Kafka via a unified adapter interface
• Message Inspection — Browse queues/topics, peek at messages without consuming them
• Schema Validation — Validate message payloads against JSON Schema definitions
• Combined Inspection — inspect_queue peeks messages AND validates each against its schema
• Validated Publishing — publish_message validates against a schema before sending — invalid messages never hit the broker
• Queue Management — Create queues/topics, bindings, and purge messages for dev/test workflows
• Broker Info — List exchanges, bindings, consumer groups, and partition details

Prerequisites

• Node.js >= 22 — Required runtime (check with node --version)
• A message broker:
  • RabbitMQ with the management plugin enabled (HTTP API on port 15672), or
  • Apache Kafka (requires @confluentinc/kafka-javascript as peer dependency)
• An MCP-compatible client — Claude Code, Claude Desktop, Cursor, VS Code (Copilot), Windsurf, etc.

Quick Start

1. Define your schemas

Create JSON Schema files in a directory:

schemas/order.created.json:

{
  "$id": "order.created",
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Order Created",
  "description": "Emitted when a new order is placed",
  "version": "1.0.0",
  "type": "object",
  "required": ["orderId", "amount"],
  "properties": {
    "orderId": { "type": "string" },
    "amount": { "type": "number" }
  }
}

2. Add to your MCP client

Generate the config for your client with queue-pilot init:

npx queue-pilot init --schemas /absolute/path/to/your/schemas --client <name>

Supported clients: claude-code, claude-desktop, vscode, cursor, windsurf. Omit --client for generic JSON.

For Kafka, add --broker kafka. The generated config automatically includes the required @confluentinc/kafka-javascript peer dependency.

Non-default credentials are included as environment variables to avoid exposing secrets in ps output:

npx queue-pilot init --schemas ./schemas --rabbitmq-user admin --rabbitmq-pass secret

Run npx queue-pilot init --help for all options including Kafka SASL authentication.

Windows note: If npx fails to resolve the package, try cmd /c npx queue-pilot init ....

Manual configuration (without init)

Add the following server configuration to your MCP client:

RabbitMQ:

{
  "mcpServers": {
    "queue-pilot": {
      "command": "npx",
      "args": [
        "-y",
        "queue-pilot",
        "--schemas", "/absolute/path/to/your/schemas"
      ]
    }
  }
}

Kafka:

{
  "mcpServers": {
    "queue-pilot": {
      "command": "npx",
      "args": [
        "-y",
        "--package=@confluentinc/kafka-javascript",
        "--package=queue-pilot",
        "queue-pilot",
        "--schemas", "/absolute/path/to/your/schemas",
        "--broker", "kafka"
      ],
      "env": {
        "KAFKA_BROKERS": "localhost:9092"
      }
    }
  }
}

Schema path tip: Use an absolute path for --schemas. Relative paths resolve from the MCP client's working directory, which may not be your project root.

Client	Config file
Claude Code	.mcp.json (project) or ~/.claude.json (user)
Claude Desktop	claude_desktop_config.json
Cursor	.cursor/mcp.json
VS Code (Copilot)	.vscode/mcp.json (uses "servers" instead of "mcpServers")
Windsurf	~/.codeium/windsurf/mcp_config.json

Development (running from source)

{
  "mcpServers": {
    "queue-pilot": {
      "command": "npx",
      "args": [
        "tsx",
        "src/index.ts",
        "--schemas", "./schemas"
      ],
      "cwd": "/path/to/queue-pilot"
    }
  }
}

3. Use it

Ask your assistant things like:

• "Which queues are there and how many messages do they have?"
• "Show me the messages in the orders queue"
• "Inspect the registration queue and check if all messages are valid"
• "What schemas are available?"
• "Validate this message against the order.created schema"
• "Publish an order.created event to the events exchange"
• "Create a queue called dead-letters and bind it to the events exchange"
• "Purge all messages from the orders queue"
• "List all consumer groups" (Kafka)
• "Show me the partition details for the orders topic" (Kafka)

MCP Tools

Universal tools (all brokers)

Tool	Description
list_schemas	List all loaded message schemas
get_schema	Get the full definition of a specific schema
validate_message	Validate a JSON message against a schema
list_queues	List all queues/topics with message counts
peek_messages	View messages in a queue/topic without consuming them
inspect_queue	Peek messages + validate each against its schema
get_overview	Get broker cluster overview
check_health	Check broker health status
get_queue	Get detailed information about a specific queue/topic
list_consumers	List consumers (RabbitMQ) or consumer groups (Kafka)
publish_message	Publish a message with optional schema validation gate
purge_queue	Remove all messages from a queue/topic
create_queue	Create a new queue/topic
delete_queue	Delete a queue/topic

RabbitMQ-specific tools

Tool	Description
list_exchanges	List all RabbitMQ exchanges
create_exchange	Create a new exchange
delete_exchange	Delete an exchange
list_bindings	List bindings between exchanges and queues
create_binding	Bind a queue to an exchange with a routing key
delete_binding	Delete a binding
list_connections	List all client connections to the broker

Kafka-specific tools

Tool	Description
list_consumer_groups	List all consumer groups with their state
describe_consumer_group	Show members, assignments, and state of a consumer group
list_partitions	Show partition details for a topic (leader, replicas, ISR)
get_offsets	Show earliest/latest offsets per partition

MCP Prompts & Resources

Prompts

Pre-built workflow templates that guide your AI assistant through multi-step operations.

Prompt	Parameters	Description
debug-flow	exchange, queue	Trace bindings from exchange to queue, peek messages, and validate each against its schema
health-report	(none)	Check broker health, get cluster overview, flag queues with backed-up messages
schema-compliance	queue (optional)	Peek messages and validate each against its schema — for one queue or all queues

Usage example (in any MCP-compatible client):

"Use the debug-flow prompt for exchange 'events' and queue 'orders'"

Resources

Each loaded schema is exposed as a readable MCP resource at schema:///<schema-name>.

Clients that support MCP resources can read schema definitions directly without calling tools. For example, a schema loaded from order.created.json is available at schema:///order.created.

Schema Format

Schemas follow JSON Schema draft-07 with a few conventions:

• $id — Message type identifier (matches the type property on messages)
• version — Schema version (custom field, not validated by JSON Schema)
• Standard JSON Schema validation including required, properties, format etc.

Schema matching: when inspecting a queue, the message's type property is used to find the corresponding schema by $id.

Configuration

CLI arguments take priority over environment variables, which take priority over defaults.

Setting	CLI flag	Env var	Default
Schema directory	--schemas	—	(required)
Broker type	--broker	—	rabbitmq
RabbitMQ URL	--rabbitmq-url	RABBITMQ_URL	http://localhost:15672
RabbitMQ user	--rabbitmq-user	RABBITMQ_USER	guest
RabbitMQ password	--rabbitmq-pass	RABBITMQ_PASS	guest
Kafka brokers	--kafka-brokers	KAFKA_BROKERS	localhost:9092
Kafka client ID	--kafka-client-id	KAFKA_CLIENT_ID	queue-pilot
SASL mechanism	--kafka-sasl-mechanism	KAFKA_SASL_MECHANISM	(none)
SASL username	--kafka-sasl-username	KAFKA_SASL_USERNAME	(none)
SASL password	--kafka-sasl-password	KAFKA_SASL_PASSWORD	(none)

Use environment variables in MCP client env blocks to avoid exposing credentials in ps output.

Development

npm install
npm test                    # Unit tests
npm run test:coverage       # Coverage report
npm run build               # TypeScript compilation
npm run typecheck           # Type check

# Integration tests (requires RabbitMQ)
docker compose up -d --wait
npm run test:integration

Tech Stack

• TypeScript (strict mode, ESM)
• MCP SDK v1.26.0
• Ajv for JSON Schema validation
• Zod for MCP tool parameter definitions
• Vitest for testing
• RabbitMQ Management HTTP API
• Confluent Kafka JavaScript (optional, for Kafka support)

License

MIT