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 許可證。