🌸 celery-flower-mcp

What is this?

celery-flower-mcp is a Model Context Protocol server that exposes the full Celery Flower REST API as MCP tools. Point it at your Flower instance and your AI assistant (Claude, Cursor, Windsurf, etc.) can:

• Monitor workers, tasks, and queues in real time
• Control worker pools — grow, shrink, autoscale, restart, shut down
• Manage tasks — apply, revoke, abort, set timeouts and rate limits
• Inspect queues — check depths, add/remove consumers

All 21 Flower API endpoints are covered.

Features

• Full API coverage — every Flower REST endpoint exposed as an MCP tool
• Dependency injection via dishka — clean, testable architecture
• Pydantic Settings — typed configuration with .env file support
• Async throughout — built on httpx + FastMCP
• 65 tests — 49 unit tests (99% coverage) + 16 integration tests against real Flower
• Strict typing — mypy strict mode, fully annotated

Quick Start

Install via uvx

FLOWER_URL=http://localhost:5555 uvx celery-flower-mcp

Install from source

git clone https://github.com/Darius1223/celery-flower-mcp
cd celery-flower-mcp
uv sync
uv run python -m source.main

Claude Desktop

Add to ~/Library/Application Support/Claude/claude_desktop_config.json:

{
  "mcpServers": {
    "celery-flower": {
      "command": "uvx",
      "args": ["celery-flower-mcp"],
      "env": {
        "FLOWER_URL": "http://localhost:5555"
      }
    }
  }
}

Configuration

Configuration is read from environment variables or a .env file in the project root. Copy .env.example to get started:

cp .env.example .env

| Variable | Default | Description | |---|---|---| | FLOWER_URL | http://localhost:5555 | Base URL of your Flower instance | | FLOWER_USERNAME | — | Basic auth username | | FLOWER_PASSWORD | — | Basic auth password | | FLOWER_API_TOKEN | — | Bearer token (takes priority over basic auth) |

Available Tools

Workers (8 tools)

| Tool | Description | |---|---| | list_workers | List all workers — optionally filter by name, refresh live stats, or get status only | | shutdown_worker | Gracefully shut down a worker | | restart_worker_pool | Restart a worker's process pool | | grow_worker_pool | Add N processes to a worker's pool | | shrink_worker_pool | Remove N processes from a worker's pool | | autoscale_worker_pool | Configure autoscale min/max bounds | | add_queue_consumer | Make a worker start consuming from a queue | | cancel_queue_consumer | Make a worker stop consuming from a queue |

Tasks (11 tools)

| Tool | Description | |---|---| | list_tasks | List tasks with filters: state, worker, name, date range, search, pagination | | list_task_types | List all registered task types across workers | | get_task_info | Get full details for a task by UUID | | get_task_result | Retrieve a task's result (with optional timeout) | | apply_task | Execute a task synchronously and wait for the result | | async_apply_task | Dispatch a task asynchronously, returns task UUID | | send_task | Send a task by name — no registration required on worker side | | abort_task | Abort a running task | | revoke_task | Revoke a task; optionally terminate with a signal | | set_task_timeout | Set soft and/or hard time limits for a task on a worker | | set_task_rate_limit | Set rate limit for a task on a worker (e.g. 100/m) |

Queues & Health (2 tools)

| Tool | Description | |---|---| | get_queue_lengths | Get the current depth of all configured queues | | healthcheck | Check whether the Flower instance is reachable and healthy |

Architecture

source/
├── main.py        # FastMCP server entry point + dishka container wiring
├── settings.py    # Pydantic Settings — typed config from env / .env
├── client.py      # Async HTTP client wrapping Flower REST API
├── providers.py   # dishka Provider — manages FlowerClient lifecycle
└── tools/
    ├── workers.py # 8 worker management tools
    ├── tasks.py   # 11 task management tools
    └── queues.py  # 2 queue / health tools

dishka manages the FlowerClient lifecycle: created once at startup, closed cleanly on shutdown via an async generator provider.

Development

make fmt        # auto-format with ruff
make lint       # lint with ruff
make typecheck  # type-check with mypy (strict)
make test       # run 49 unit tests
make cov        # unit tests + coverage report
make all        # fmt + lint + typecheck

Testing

The test suite is split into two layers:

Unit tests (tests/) — fast, no external dependencies, use pytest-httpx to mock HTTP calls:

make test
# or
uv run pytest tests/ -m "not integration"

Integration tests (tests/integration/) — run against a real Flower instance backed by Redis and a live Celery worker, all managed by Docker Compose:

make integration

This command:

1. Builds and starts the Docker Compose stack (docker-compose.test.yml) — Redis → Celery worker → Flower
2. Waits for Flower's /healthcheck endpoint to return OK
3. Runs the 16 integration tests against http://localhost:5555
4. Tears down the stack when done

The stack is defined in docker-compose.test.yml. The worker and Flower images are built from tests/integration/Dockerfile.worker and tests/integration/Dockerfile.flower.

To start the stack manually for exploratory testing:

docker compose -f docker-compose.test.yml up -d --build
# run tests, explore, etc.
make integration-down   # stop + remove volumes

Integration tests use pytest.mark.asyncio(loop_scope="session") so all tests share one event loop — this avoids RuntimeError: Event loop is closed when httpx transports are cleaned up across test boundaries on Python 3.14.

See CONTRIBUTING.md for details on adding new tools or submitting a PR.

Changelog

See CHANGELOG.md.

License

MIT