Celery Flower MCP

🌸 celery-flower-mcp

---

O que é isso?

celery-flower-mcp é um servidor Model Context Protocol que expõe a API REST completa do Celery Flower como ferramentas MCP. Aponte-o para sua instância do Flower e seu assistente de IA (Claude, Cursor, Windsurf, etc.) poderá:

• Monitorar workers, tarefas e filas em tempo real
• Controlar pools de workers — aumentar, reduzir, autoscale, reiniciar, desligar
• Gerenciar tarefas — aplicar, revogar, abortar, definir timeouts e limites de taxa
• Inspecionar filas — verificar profundidade, adicionar/remover consumidores

Todos os 21 endpoints da API do Flower são cobertos.

Recursos

• Cobertura completa da API — cada endpoint REST do Flower exposto como ferramenta MCP
• Injeção de dependência via dishka — arquitetura limpa e testável
• Pydantic Settings — configuração tipada com suporte a arquivo .env
• Assíncrono em tudo — construído sobre httpx + FastMCP
• 65 testes — 49 testes unitários (99% de cobertura) + 16 testes de integração contra um Flower real
• Tipagem estrita — modo estrito do mypy, totalmente anotado

Início Rápido

Instalar via uvx

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

Instalar a partir do código-fonte

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

Claude Desktop

Adicione ao ~/Library/Application Support/Claude/claude_desktop_config.json:

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

Configuração

A configuração é lida de variáveis de ambiente ou de um arquivo .env na raiz do projeto. Copie o .env.example para começar:

cp .env.example .env

Variável	Padrão	Descrição
FLOWER_URL	http://localhost:5555	URL base da sua instância do Flower
FLOWER_USERNAME	—	Nome de usuário para autenticação básica
FLOWER_PASSWORD	—	Senha para autenticação básica
FLOWER_API_TOKEN	—	Token Bearer (tem prioridade sobre a autenticação básica)

Ferramentas Disponíveis

Workers (8 ferramentas)

Ferramenta	Descrição
list_workers	Listar todos os workers — opcionalmente filtrar por nome, atualizar estatísticas ao vivo ou obter apenas o status
shutdown_worker	Desligar um worker de forma graciosa
restart_worker_pool	Reiniciar o pool de processos de um worker
grow_worker_pool	Adicionar N processos ao pool de um worker
shrink_worker_pool	Remover N processos do pool de um worker
autoscale_worker_pool	Configurar limites mínimos/máximos de autoscale
add_queue_consumer	Fazer um worker começar a consumir de uma fila
cancel_queue_consumer	Fazer um worker parar de consumir de uma fila

Tarefas (11 ferramentas)

Ferramenta	Descrição
list_tasks	Listar tarefas com filtros: estado, worker, nome, intervalo de datas, busca, paginação
list_task_types	Listar todos os tipos de tarefa registrados nos workers
get_task_info	Obter detalhes completos de uma tarefa por UUID
get_task_result	Recuperar o resultado de uma tarefa (com timeout opcional)
apply_task	Executar uma tarefa de forma síncrona e aguardar o resultado
async_apply_task	Enviar uma tarefa de forma assíncrona, retorna o UUID da tarefa
send_task	Enviar uma tarefa pelo nome — sem necessidade de registro no lado do worker
abort_task	Abortar uma tarefa em execução
revoke_task	Revogar uma tarefa; opcionalmente encerrar com um sinal
set_task_timeout	Definir limites de tempo suave e/ou rígido para uma tarefa em um worker
set_task_rate_limit	Definir limite de taxa para uma tarefa em um worker (ex.: 100/m)

Filas e Saúde (2 ferramentas)

Ferramenta	Descrição
get_queue_lengths	Obter a profundidade atual de todas as filas configuradas
healthcheck	Verificar se a instância do Flower está acessível e saudável

Arquitetura

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

O dishka gerencia o ciclo de vida do FlowerClient: criado uma vez na inicialização, fechado de forma limpa no encerramento via um provider gerador assíncrono.

Desenvolvimento

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

Testes

A suíte de testes é dividida em duas camadas:

Testes unitários (tests/) — rápidos, sem dependências externas, usam pytest-httpx para simular chamadas HTTP:

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

Testes de integração (tests/integration/) — executados contra uma instância real do Flower com Redis e um worker Celery ativo, tudo gerenciado pelo Docker Compose:

make integration

Este comando:

1. Constrói e inicia a pilha do Docker Compose (docker-compose.test.yml) — Redis → worker Celery → Flower
2. Aguarda o endpoint /healthcheck do Flower retornar OK
3. Executa os 16 testes de integração contra http://localhost:5555
4. Encerra a pilha ao final

A pilha é definida em docker-compose.test.yml. As imagens do worker e do Flower são construídas a partir de tests/integration/Dockerfile.worker e tests/integration/Dockerfile.flower.

Para iniciar a pilha manualmente para testes exploratórios:

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

Os testes de integração usam pytest.mark.asyncio(loop_scope="session") para que todos os testes compartilhem um único event loop — isso evita RuntimeError: Event loop is closed quando os transports do httpx são limpos entre os limites dos testes no Python 3.14.

Consulte CONTRIBUTING.md para detalhes sobre como adicionar novas ferramentas ou enviar um PR.

Changelog

Consulte CHANGELOG.md.

Licença

MIT