mcp-server-celery

mcp-server-celery

一个 MCP 服务，可让 AI Agent 调试正在运行的 Celery 部署。它会检查哪些 worker 处于活动状态、每个 worker 消费什么，以及消息堆积在哪里——这些检查通常需要在发生故障时手动运行 celery inspect 和 rabbitmqctl。

默认情况下，它是只读的。每个响应中的 Broker 密码都会被遮蔽，包括 Celery 自带的 inspect stats 以明文返回的 Broker 信息。

支持 RabbitMQ（amqp://）和 Redis（redis://、rediss://）。

工具

工具	可回答的问题
list_workers	哪些 worker 会响应、它们的 pool 和并发数、运行时长、预取数量、已处理的任务，以及每个 worker 监听的队列。
queue_depths	每个队列中有多少条等待消息，以及哪些 worker 在消费该队列。不带参数时，会检查已配置的队列，以及所有在线 worker 正在消费的队列。

即将推出：active_tasks、reserved_tasks、scheduled_tasks、task_status 和 diagnose_stuck。这些工具会交叉参考上述信息，标记没有 worker 监听的队列、路由到没有消费者的队列的任务，以及停止响应的 worker。请参阅路线图。

安装

需要 Python 3.10+ 和 uv。

uvx mcp-server-celery

配置

所有配置都通过环境变量进行。

变量	默认值	用途
CELERY_BROKER_URL	amqp://guest:guest@localhost:5672//	要检查的 Broker。
CELERY_RESULT_BACKEND	—	结果后端（将在下一个版本中用于 task_status）。
CELERY_APP	—	应用的导入路径，例如 proj.celery:app。让服务能够读取你的 task_queues 和路由。如果两者都已设置，CELERY_BROKER_URL 仍会覆盖应用的 Broker。
CELERY_INSPECT_TIMEOUT	1.0	等待 worker 回复的秒数。对于大型或响应较慢的集群，请调高此值。

CELERY_APP 必须可导入，因此请从项目目录和环境中运行服务：

uv run --with mcp-server-celery mcp-server-celery

连接

Claude Code

claude mcp add celery -e CELERY_BROKER_URL=redis://localhost:6379/0 -- uvx mcp-server-celery

Cursor

在 .cursor/mcp.json 中：

{
  "mcpServers": {
    "celery": {
      "command": "uvx",
      "args": ["mcp-server-celery"],
      "env": { "CELERY_BROKER_URL": "redis://localhost:6379/0" }
    }
  }
}

Claude Desktop

在 claude_desktop_config.json 中，将与 Cursor 相同的 command、args 和 env 块放在 mcpServers 下。

示例

询问 Agent“为什么我的邮件任务没有运行？”，它就可以调用 queue_depths：

{
  "broker": "amqp://app:****@rabbit:5672//",
  "queues": [
    { "queue": "celery", "exists": true, "messages": 0, "consumers": 2, "workers_consuming": ["w1@web", "w2@web"] },
    { "queue": "emails", "exists": true, "messages": 1843, "consumers": 0, "workers_consuming": [] }
  ]
}

emails 队列有积压，但没有 worker 消费它：启动 worker 时没有指定 -Q emails。

数据说明

• Redis 没有消费者的概念，因此 consumers 为 null；请使用来自 worker 自身的 workers_consuming。Redis 还会在队列清空后删除其键，因此已清空的队列会显示 "messages": 0, "exists": null。
• RabbitMQ 检查使用被动声明，因此服务绝不会创建队列。不存在的队列会显示 "exists": false。
• 忙到无法在 CELERY_INSPECT_TIMEOUT 时间内回复的 worker 不会出现在 list_workers 中。如果预期的 worker 不见了，请先尝试调大 timeout 参数，再判断它是否已停止运行。

安全

• 每个工具都标注为 readOnlyHint: true。写入操作（撤销、重试、清空）只有在显式启用选项后才会开放。
• 所有输出和错误消息中的 URL 凭据以及 password/secret/token 字段都会被遮蔽。

路线图

v0.1 — list_workers、queue_depths ✅

v1 — active_tasks、reserved_tasks、scheduled_tasks、task_status(task_id)、diagnose_stuck

v2 — summarize_failures：按任务名称和异常类型汇总近期故障。Celery 默认不保留故障历史，因此此功能需要事件监听器。

开发

uv sync
uv run pytest tests/unit
uv run ruff check . && uv run ruff format --check .

集成测试会启动一个真实的工作进程，需要一个消息代理：

CELERY_BROKER_URL=redis://localhost:6379/0 uv run pytest tests/integration -m integration

每次推送时，CI 都会使用 Redis 和 RabbitMQ 运行这些测试。

发布

更新 pyproject.toml 中的版本，以及 server.json 中两个 version 字段的版本，提交后推送 vX.Y.Z 标签。Release 工作流会检查这三个版本是否与标签一致，将软件包发布到 PyPI，然后将 server.json 发布到 MCP 注册表。

许可证

MIT