The Complete Production SOP: How to Deploy and Run Celery Distributed Task Queue & Worker Tuning in Autonomous AI Agents

---
name: "celery-distributed-task-worker-pool"
description: "Run distributed asynchronous background tasks and scheduled crons with Python Celery."
metadata:
  author: "Celery"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "tune celery worker concurrency"
  - "configure celery task routing"
  - "enable celery task_acks_late"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# Celery Distributed Task Queue & Worker Tuning

Configures Celery with Redis/RabbitMQ brokers: configuring prefork vs gevent/eventlet pools, worker concurrency, task routing, late acknowledgments (task_acks_late), and flower monitoring dashboards.

## When to Invoke
Invoke this skill whenever you need to:
- tune celery worker concurrency
- configure celery task routing
- enable celery task_acks_late

## Execution SOP & Rules
1. Initialize Celery app: app = Celery('tasks', broker='redis://localhost:6379/0').
2. Configure app.conf.task_routes = {'tasks.heavy_*': {'queue': 'heavy'}, 'tasks.email_*': {'queue': 'email'}}.
3. Set task_acks_late = True and worker_prefetch_multiplier = 1 for long-running reliable tasks.
4. Run workers specifying queue and pool: celery -A tasks worker -l info -Q heavy -c 4.

## Verification Checklist
- Run local linting, syntax validation, and type-checks before reporting completion.
- Ensure all created or modified files adhere to workspace boundaries and avoid uncommitted secrets.
- Verify tests or build commands exit cleanly with return code 0.

import os
from celery import Celery
from kombu import Queue, Exchange

broker_url = os.getenv('CELERY_BROKER_URL', 'redis://localhost:6379/0')
result_backend = os.getenv('CELERY_RESULT_BACKEND', 'redis://localhost:6379/1')

app = Celery('prod_worker', broker=broker_url, backend=result_backend)

app.conf.update(
    task_default_queue='default',
    task_queues=(
        Queue('default', Exchange('default'), routing_key='default'),
        Queue('heavy', Exchange('heavy'), routing_key='heavy'),
        Queue('email', Exchange('email'), routing_key='email'),
    ),
    task_routes={
        'tasks.heavy_*': {'queue': 'heavy'},
        'tasks.email_*': {'queue': 'email'},
    },
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,
    worker_max_tasks_per_child=50,
    timezone='UTC',
    enable_utc=True,
)