Queue Pilot

🚀 Queue Pilot

Queue Pilot 是一款用于消息队列开发的 MCP 服务器，它将消息检查与 JSON Schema 验证相结合，支持 RabbitMQ 和 Kafka。专为多个团队通过消息代理进行通信的集成项目而设计，你可以通过 AI 助手检查队列/主题、查看消息，并根据商定的模式验证有效负载。

🚀 快速开始

1. 定义你的模式

在一个目录中创建 JSON Schema 文件，例如 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. 添加到你的 MCP 客户端

使用 queue-pilot init 为你的客户端生成配置：

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

支持的客户端有：claude-code、claude-desktop、vscode、cursor、windsurf。如果省略 --client，则生成通用的 JSON 配置。

对于 Kafka，添加 --broker kafka。生成的配置会自动包含所需的 @confluentinc/kafka-javascript 对等依赖项。

非默认凭证以环境变量的形式包含，以避免在 ps 输出中暴露机密信息：

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

运行 npx queue-pilot init --help 以查看包括 Kafka SASL 认证在内的所有选项。

⚠️ 重要提示

Windows 用户注意：如果 npx 无法解析包，请尝试 cmd /c npx queue-pilot init ...。

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

{
  "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"
      }
    }
  }
}

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

3. 使用它

向你的助手询问以下问题：

• "有哪些队列，每个队列有多少条消息？"
• "显示订单队列中的消息"
• "检查注册队列，查看所有消息是否有效"
• "有哪些可用的模式？"
• "根据 order.created 模式验证此消息"
• "向 events 交换器发布一个 order.created 事件"
• "创建一个名为 dead-letters 的队列，并将其绑定到 events 交换器"
• "清空订单队列中的所有消息"
• "列出所有消费者组"（Kafka）
• "显示订单主题的分区详细信息"（Kafka）

✨ 主要特性

• 多代理支持 — 通过统一的适配器接口支持 RabbitMQ 和 Apache Kafka。
• 消息检查 — 浏览队列/主题，查看消息而不消费它们。
• 模式验证 — 根据 JSON Schema 定义验证消息有效负载。
• 组合检查 — inspect_queue 查看消息并根据其模式验证每条消息。
• 验证发布 — publish_message 在发送前根据模式进行验证，无效消息不会到达代理。
• 队列管理 — 创建队列/主题、绑定，并为开发/测试工作流清除消息。
• 代理信息 — 列出交换器、绑定、消费者组和分区详细信息。

📦 安装指南

前提条件

• Node.js >= 22 — 必需的运行时（使用 node --version 检查）。
• 消息代理：
  • RabbitMQ，启用 管理插件（HTTP API 端口为 15672）。
  • Apache Kafka（需要 @confluentinc/kafka-javascript 作为对等依赖项）。
• 兼容 MCP 的客户端 — Claude Code、Claude Desktop、Cursor、VS Code (Copilot)、Windsurf 等。

📚 详细文档

模式格式

模式遵循 JSON Schema draft-07，并遵循以下约定：

• $id — 消息类型标识符（与消息上的 type 属性匹配）。
• version — 模式版本（自定义字段，JSON Schema 不验证）。
• 标准 JSON Schema 验证，包括 required、properties、format 等。

模式匹配：检查队列时，消息的 type 属性用于通过 $id 查找相应的模式。

配置

CLI 参数优先于环境变量，环境变量优先于默认值。

在 MCP 客户端的 env 块中使用环境变量，以避免在 ps 输出中暴露凭证。

🔧 技术细节

开发

npm install
npm test                    # 单元测试
npm run test:coverage       # 覆盖率报告
npm run build               # TypeScript 编译
npm run typecheck           # 类型检查

# 集成测试（需要 RabbitMQ）
docker compose up -d --wait
npm run test:integration

技术栈

• TypeScript（严格模式，ESM）
• MCP SDK v1.26.0
• Ajv 用于 JSON Schema 验证
• Zod 用于 MCP 工具参数定义
• Vitest 用于测试
• RabbitMQ 管理 HTTP API
• Confluent Kafka JavaScript（可选，用于 Kafka 支持）

📄 许可证

本项目采用 MIT 许可证。