rabbit-mcp

rabbit-mcp

English · ქართული

基于消息契约（.NET 10）构建的 RabbitMQ MCP 服务。每个契约都会成为一个类型化工具，其输入架构就是契约本身。消息采用 MassTransit 线路格式，因此 MassTransit 服务可以直接接收；不使用 MassTransit 的服务则会收到普通 JSON。服务本身不依赖 MassTransit。

• 发送： publish_<contract> 和 request_<contract>，发送前先根据契约进行验证。
• 查看： 队列、消息（包括故障详情），以及各个契约的消费方。
• 恢复： 将消息从 _error 队列移回、修复后重新发送，或搁置有问题的消息。消息绝不会被删除。
• 答复： 在响应方模式下，模型会回答服务发来的请求。
• 防护： contracts diff 可在 CI 中捕获会破坏线路兼容性的更改；支持团队使用的角色、速率限制和审计。

安装

方式	说明
二进制文件	Releases：适用于 Linux、Windows 和 macOS 的独立版本。macOS：xattr -d com.apple.quarantine rabbit-mcp
.NET 工具	dotnet tool install -g RabbitMcp.Server，或无需安装：dnx RabbitMcp.Server --yes（从首个版本开始提供）
Docker	ghcr.io/lazogio/rabbit-mcp — HTTP 模式，参见团队使用
源码	dotnet publish src/RabbitMcp.Server -c Release -o out（切勿使用 dotnet run：它会写入 stdout）

claude mcp add rabbit-mcp \
  -e RabbitMcp__Uri=amqp://guest:guest@localhost:5672/ \
  -e RabbitMcp__EnablePublish=true \
  -- /full/path/to/rabbit-mcp

Claude Desktop：配置方式与 claude_desktop_config.json 中的 "command" 和 "env" 相同。本地 RabbitMQ：docker compose up -d。

试用： samples/Shop 可通过一条命令启动 RabbitMQ、PostgreSQL、MassTransit 商店和 rabbit-mcp，并演示订单、错误恢复和审批流程。

工具

工具	功能	启用条件
list_contracts、validate_message	契约及其详情；检查 JSON，但不发送	始终启用
queue_stats、peek_messages	队列计数；查看消息但不移除（负载、契约检查、故障、x-death）	始终启用
find_message	按 ID 或文本查找消息，并告知其所在队列	始终启用
read_stream	从 first、last、偏移量或时间（10m）读取 RabbitMQ 流	始终启用
list_queues、contract_consumers	队列；查看契约会到达哪些队列以及由谁消费	ManagementUri
recent_activity	审计： “谁昨天重新发布了消息？”	AuditLogPath
publish_<contract>、request_<contract>	发布事件／发送请求并返回答复	EnablePublish
republish_from_error	将消息从错误队列或死信队列移回：先预览，再设置 confirm=true。支持按筛选条件（exceptionType、contract、since …）处理事故	EnableRepublish
dead_letter_from_error	将永远无法成功的消息搁置到 <queue>_dead，并附上原因	EnableRepublish
resend_corrected	发送修正后的失败消息副本，并搁置原消息	EnableRepublish + EnablePublish
pending_requests、reply_<response>、reject_request	查看等待答复的服务请求	RespondTo

结果采用结构化格式（outputSchema）；错误信息会说明接下来该怎么做。提示词 diagnose_errors、investigate_message、review_approvals 和 contract_health 可引导完成常见任务，每个契约也都是一个资源 contract://<Name>。

合约

// Event → publish_payment_captured
[MessageContract("Payment was captured by the PSP.", Version = 1)]
public sealed record PaymentCaptured
{
    [Description("Order the payment belongs to.")]
    public required Guid OrderId { get; init; }

    [Range(typeof(decimal), "0.01", "1000000")]
    public required decimal Amount { get; init; }
}

// Request → request_get_order_status
[MessageContract("Asks for the current status of an order.",
    Responses = new[] { typeof(OrderStatus), typeof(OrderNotFound) })]
public sealed record GetOrderStatus
{
    public required Guid OrderId { get; init; }
}

• required、可空性、[Description] 和 DataAnnotations 会写入架构并得到强制执行。
• 名称遵循 MassTransit 规则（urn:message:Namespace:Type，交换机为 Namespace:Type）；Urn 和 Exchange 可覆盖这些名称。
• 你的合约：引用 RabbitMcp.Contracts 包的类库，通过 RabbitMcp__ContractAssemblies=/path/Billing.Contracts.dll 加载。

不使用 MassTransit 的服务 — Format = MessageFormat.Raw：消息体为纯 JSON，元数据写入 AMQP 属性，并设置 Exchange 和/或 RoutingKey。请求使用 RabbitMQ RPC（reply_to + correlation_id；回复类型位于 AMQP type 中）。

[MessageContract("Asks the shipping desk for a quote.", Format = MessageFormat.Raw, RoutingKey = "shipping-quotes",
    Responses = new[] { typeof(ShippingQuote), typeof(ShippingUnavailable) })]
public sealed record GetShippingQuote { … }

不使用 C#（AsyncAPI） — RabbitMcp__ContractFiles=warehouse.yaml：AsyncAPI 3.0 或 2.x 文件中的每条消息都会成为一个工具，并使用文件中的 JSON Schema；路由来自 AMQP 绑定，MassTransit 服务使用 x-rabbit-mcp: { format: masstransit }。示例：samples/asyncapi/warehouse.yaml。rabbit-mcp contracts asyncapi Billing.Contracts.dll 可导出 C# 合约。

兼容性检查 — 在 CI 中比较两个构建（或两个 AsyncAPI 文件）；存在不兼容变更时退出代码为 1：

rabbit-mcp contracts diff old/Billing.Contracts.dll new/Billing.Contracts.dll
# BREAKING OrderPlaced customer: Type changed from string to number.
# WARNING  OrderPlaced: Breaking change without a version bump (still v1).

请求/响应

request_<contract> 会等待回复（超时 1–120 s），并返回 responded、faulted 或 timeout；回复会根据其合约进行检查。

响应方： RabbitMcp__RespondTo=ApproveLargeOrder 可让你的服务发起请求并由模型作答。请求会显示在 pending_requests 中；模型通过 reply_<response> 或 reject_request 作答（请求方会收到 RequestFaultException）。请求保存在内存中；每个 ClientName 仅由一个实例作答，其他实例待命。

面向团队（HTTP）

RabbitMcp__Transport=http 会提供 /mcp（和 /healthz）。未进行身份验证时拒绝启动：

• API 密钥： RabbitMcp__ApiKeys="alice:KEY1;ci:KEY2"（≥ 32 个字符）；客户端发送 X-Api-Key。
• JWT / OIDC： RabbitMcp__JwtAuthority + RabbitMcp__JwtAudience。

docker run -d -p 5080:5080 \
  -e RabbitMcp__Uri=amqp://user:pass@rabbit:5672/ \
  -e RabbitMcp__ApiKeys="team:$(openssl rand -base64 32)" \
  -e RabbitMcp__ContractAssemblies=/contracts/Billing.Contracts.dll -v ./contracts:/contracts:ro \
  ghcr.io/lazogio/rabbit-mcp:1.0

角色（仅限 HTTP；未配置角色时，每位调用者都可以使用所有已启用的工具）：

RabbitMcp__Roles__viewer="read"
RabbitMcp__Roles__operator="read;publish:Order*;request;republish:*_error"
RabbitMcp__UserRoles="alice=operator;ci=viewer"
RabbitMcp__DefaultRoles="viewer"

权限：read、publish、request、republish、respond、*；模式可按合约或队列限制权限。用户通过 API 密钥名称、令牌用户或 JWT roles 声明进行匹配。调用者只能看到自己有权使用的工具，每次调用也会再次检查权限。

配置

环境变量 RabbitMcp__<Option> 或 appsettings.json。列表以 ; 分隔，* 是通配符。

选项	默认值	描述
Uri	amqp://guest:guest@localhost:5672/	AMQP URI，末尾为虚拟主机
ContractAssemblies / ContractFiles	—	合约 DLL / AsyncAPI 文件。为空时使用内置示例
ManagementUri	—	管理 API，例如 http://localhost:15672
EnablePublish / EnableRepublish	false	允许发送 / 移动消息
PublishAllow / PublishDeny	—	允许 / 禁止发送的合约
DeadLetterQueue	{queue}_dead	存放有问题消息的位置
RespondTo / MaxPendingRequests	— / 100	响应器合约 / 等待请求上限
HumanConfirmation / ConfirmTools	auto / 三个恢复工具	用户在客户端对话框中确认更改（无法询问时，required 会拒绝；off 则信任 confirm）
AuditLogPath / AuditIncludePayload	— / false	审计文件（JSON Lines）/ 是否包含消息正文
MaxMessagesPerCall / MaxBulkMessages / MaxPayloadChars	50 / 1000 / 16000	限制
RateLimits	—	每个调用方的更改次数限制，例如 publish:100/h;*:1000/d
Transport / HttpUrls / HttpPath	stdio / http://localhost:5080 / /mcp	传输方式
ApiKeys, JwtAuthority, JwtAudience, JwtScopes	—	HTTP 身份验证
Roles__<role>, UserRoles, DefaultRoles	—	角色
ClientName, HttpPublicUrl, HttpAllowedOrigins, HttpAllowAnonymous, ExchangeCacheSeconds, JwtRolesClaim		很少需要

OpenTelemetry：设置 OTEL_EXPORTER_OTLP_ENDPOINT；跟踪会关联到 MassTransit 消费者的跟踪。

保证

• 不会静默丢失： 无效消息不会发送；发布者确认和 mandatory 会将无法投递的消息转为错误。
• 安全移动： 只有在代理确认新副本后，消息才会离开错误队列；正文绝不会更改。
• 不会删除任何内容： 有问题的消息会连同所有标头一起存放，并且可以恢复。
• 由用户确认： 支持 MCP 询问的客户端（例如 Claude Code）会在移动任何内容之前显示对话框；无状态 HTTP 则回退到 confirm=true。
• 可审计： 记录每次发送、移动和答复，包括被拒绝的操作，并记录用户。
• 不可信内容： 消息标头绝不会重定向消息的去向；密码不会出现在日志或错误中。

许可证

MIT。开发、测试和发布：CONTRIBUTING.md。