Guide · Job Queues

MCP Tools for Celery — task routing, chord/group/chain, ETA scheduling, result backends

Three Celery behaviours cause silent failures in MCP integrations: task results expire and are silently discarded after the result backend's TTL — Redis result backend defaults to 1 day (result_expires=86400), so an MCP status-check tool that polls for a result days after task completion will get None with no indication the task ever ran; chord callbacks fire even when group tasks fail unless chord_propagates=True (Celery 4+) or the chord is created with chord(group, callback).set(link_error=...) — a chord body that expects all parallel results but receives partial data silently produces incorrect output; and apply_async with eta requires the broker and worker clocks to be synchronized — if the worker's system clock is behind the broker, tasks with eta in the past are executed immediately rather than raising an error, meaning time-sensitive MCP tools may fire early without warning.

TL;DR

App: app = Celery('name', broker=REDIS_URL, backend=REDIS_URL). Define: @app.task(bind=True, max_retries=3, queue='tasks'). Dispatch: task.apply_async(args, kwargs, countdown=10, eta=datetime, queue='tasks'). Chain: task_a.si(x) | task_b.s() | task_c.s(). Group: group(task.s(x) for x in items)(). Chord: chord(group(...))(callback.s()). Status: AsyncResult(task_id).state. Health: app.control.inspect().active().

App configuration and broker setup

A Celery app connects to a broker (transport layer — Redis or RabbitMQ) and a result backend (where task results are stored). The broker and backend can be the same Redis instance with different key prefixes. For production, always configure task_serializer='json' — the default pickle serializer allows arbitrary code execution if the broker is compromised. Always set result_expires explicitly; without it the default varies by backend and expired results silently return None.

from celery import Celery
import os

app = Celery(
    'mcp_tasks',
    broker=os.environ['CELERY_BROKER_URL'],    # redis://localhost:6379/0 or amqp://...
    backend=os.environ['CELERY_RESULT_BACKEND'], # redis://localhost:6379/1
)

app.conf.update(
    # Serialization — use JSON to avoid pickle deserialization vulnerabilities
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',

    # Timezone — always set explicitly; Celery default is UTC
    timezone='UTC',
    enable_utc=True,

    # Result expiry — how long task results are kept in the backend
    result_expires=3600 * 24 * 7,  # 7 days

    # Task acknowledgement — 'late' acks after task completes (safer; enables requeue on crash)
    task_acks_late=True,
    task_reject_on_worker_lost=True,

    # Rate limit — max tasks per second per worker (None = unlimited)
    task_default_rate_limit='100/s',

    # Routing — define named queues for task segregation
    task_queues={
        'default': {'exchange': 'default', 'routing_key': 'default'},
        'high_priority': {'exchange': 'high_priority', 'routing_key': 'high_priority'},
        'slow': {'exchange': 'slow', 'routing_key': 'slow'},
    },
    task_default_queue='default',
)

For RabbitMQ brokers, use the AMQP URL format: amqp://user:password@host:5672/vhost. RabbitMQ provides better message durability guarantees than Redis (messages survive broker restart if the queue is durable), but Redis is simpler to operate and is the more common choice for new Celery deployments. The result backend is always separate from the broker — use Redis for the backend even with RabbitMQ as the broker.

Defining and dispatching tasks from MCP tools

Celery tasks are defined with the @app.task decorator. MCP tools typically call apply_async to dispatch work and return the task ID for polling. Use bind=True to get access to the task instance (self) for retrying and accessing task metadata. Tasks should be defined in a module that both the MCP server and Celery workers import — keep task definitions in a shared tasks.py module, not inline in the MCP handler.

from celery import Task
from datetime import datetime, timedelta, timezone

@app.task(
    bind=True,
    name='tasks.generate_report',
    max_retries=3,
    default_retry_delay=60,      # seconds between retries (overridden by exponential below)
    queue='slow',                # always route to the 'slow' queue
    soft_time_limit=300,         # raises SoftTimeLimitExceeded after 5 min (catchable)
    time_limit=360,              # hard kill after 6 min (not catchable)
)
def generate_report(self: Task, user_id: str, report_type: str) -> dict:
    try:
        records = fetch_records(user_id)
        report = build_report(records, report_type)
        url = upload_report(report)
        return {'url': url, 'record_count': len(records)}
    except TemporaryError as exc:
        # Retry with exponential backoff: 60s, 120s, 240s
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))
    except SoftTimeLimitExceeded:
        # Clean up before hard kill
        raise

# MCP tool handler — dispatches task and returns job ID
def mcp_generate_report_tool(user_id: str, report_type: str) -> dict:
    task = generate_report.apply_async(
        args=[user_id, report_type],
        queue='slow',            # explicit queue overrides task default
        countdown=0,             # dispatch immediately
    )
    return {'task_id': task.id, 'status': 'queued'}

# Dispatch with delay (countdown = seconds from now)
def mcp_schedule_reminder(user_id: str, delay_seconds: int) -> dict:
    task = send_reminder.apply_async(
        args=[user_id],
        countdown=delay_seconds,
    )
    return {'task_id': task.id, 'fires_at': (datetime.now(timezone.utc) + timedelta(seconds=delay_seconds)).isoformat()}

# Dispatch at specific UTC datetime (eta = datetime object)
def mcp_schedule_at(user_id: str, run_at: datetime) -> dict:
    task = send_reminder.apply_async(
        args=[user_id],
        eta=run_at,              # must be timezone-aware datetime; broker clock must match
    )
    return {'task_id': task.id, 'fires_at': run_at.isoformat()}

The si() (immutable signature) and s() (signature) distinction matters in workflows: task.s(arg) appends the previous task's return value as the first argument, while task.si(arg) ignores the previous result entirely. In chain workflows, use si() for tasks that should receive only their own arguments and not be modified by the previous step's output.

Chord, group, and chain — parallel and sequential workflows

Celery's group runs tasks in parallel, chain runs tasks sequentially (each receives the previous result), and chord runs a group of tasks in parallel and then calls a callback with all results. These primitives enable MCP orchestration patterns where multiple data sources are fetched in parallel and then synthesized in a final step.

from celery import group, chain, chord
from celery.result import GroupResult

# Chain: sequential — task_b receives task_a's return value
pipeline = chain(
    fetch_data.s(source_id),         # returns raw data
    normalize_data.s(),              # receives raw data, returns normalized
    store_result.s(destination_id),  # receives normalized data
)
result = pipeline.apply_async()
# result.get() returns the final chain result (store_result's return value)

# Group: parallel — all tasks run concurrently, return list of results
parallel = group(
    fetch_github.s(repo) for repo in ['repo1', 'repo2', 'repo3']
)
group_result: GroupResult = parallel.apply_async()
# group_result.get() blocks until all tasks complete
# group_result.join_native() is faster for large groups (uses Redis BLPOP)

# Chord: parallel group + callback — callback receives list of all group results
analysis = chord(
    group(
        analyze_source.s(source) for source in ['web', 'docs', 'github']
    ),
    synthesize_findings.s(),   # called with [result1, result2, result3] as first arg
)
chord_async = analysis.apply_async()
# chord_async is the result of synthesize_findings

# Error handling in chords — chord_propagates ensures callback is not called if a task fails
app.conf.chord_propagates = True  # raises ChordError on chord_async.get() if any task failed

# Safer pattern: collect errors explicitly in callback
@app.task
def safe_synthesize(results: list) -> dict:
    # With chord_propagates=False, failed tasks pass their exception as the result
    successful = [r for r in results if not isinstance(r, Exception)]
    failed_count = len(results) - len(successful)
    return {'synthesized': merge(successful), 'failed_count': failed_count}

Chord results require the result backend to store intermediate group results. The chord callback is triggered by a chord unlock task that polls the group completion state — with many small parallel tasks, this can create polling overhead. For groups with more than 100 tasks, consider using chord_unlock_max_retries and chord_unlock_retry_interval to tune the polling frequency.

Task result polling from MCP status tools

After dispatching a Celery task, MCP tools need to check its status. AsyncResult(task_id) gives access to the task state and result. States: PENDING (default — task not started or unknown), STARTED (worker has picked it up, only if task_track_started=True), RETRY, FAILURE, SUCCESS. PENDING is ambiguous — it means either the task is waiting in the queue or the task ID is unknown (expired or never existed).

from celery.result import AsyncResult

def get_task_status(task_id: str) -> dict:
    result = AsyncResult(task_id, app=app)
    state = result.state

    response: dict = {'task_id': task_id, 'state': state}

    if state == 'SUCCESS':
        response['result'] = result.get()  # result.get() raises on FAILURE
    elif state == 'FAILURE':
        # result.result holds the exception; convert to string for JSON serialization
        response['error'] = str(result.result)
        response['traceback'] = result.traceback
    elif state == 'PENDING':
        # Distinguish "waiting in queue" from "unknown task ID"
        # One approach: store task IDs in your own database when dispatching
        response['note'] = 'Task is queued or task ID is unknown/expired'
    elif state == 'RETRY':
        response['retries'] = result.info.get('retries') if isinstance(result.info, dict) else None

    return response

# Wait for result with timeout (blocking — use only for short tasks or in async context)
def wait_for_task(task_id: str, timeout: float = 30.0):
    result = AsyncResult(task_id, app=app)
    try:
        return result.get(timeout=timeout, propagate=True)
    except Exception as exc:
        raise RuntimeError(f"Task {task_id} failed: {exc}") from exc

# Revoke (cancel) a queued task — only works if task has not started
def cancel_task(task_id: str) -> bool:
    app.control.revoke(task_id, terminate=False)  # terminate=True sends SIGTERM to running task
    return True

The PENDING ambiguity is a known Celery design limitation. To distinguish "queued" from "expired task ID" in MCP status tools, store task IDs in your own database at dispatch time with a dispatched_at timestamp. At status-check time, if the state is PENDING and the task ID is not in your database, return "not found". If it is in your database with dispatched_at less than result_expires ago, it is still queued.

Health probe — inspect API and worker liveness

Celery workers expose a control/inspect API via the broker. app.control.inspect().active() returns the currently running tasks per worker node, and app.control.inspect().registered() returns all tasks each worker has registered. Both calls use the broker as a transport and have a configurable timeout.

def check_celery_health(timeout: float = 5.0) -> dict:
    try:
        inspector = app.control.inspect(timeout=timeout)

        # Ping all workers — returns {worker_name: 'pong'} for each live worker
        ping_result = inspector.ping()
        if not ping_result:
            return {
                'alive': False,
                'error': 'No workers responded to ping within timeout',
            }

        # Active tasks per worker
        active = inspector.active() or {}
        # Queued tasks per worker (reserved — fetched but not started)
        reserved = inspector.reserved() or {}
        # Queue depths via broker inspection
        queue_lengths = {}
        try:
            from kombu import Connection
            with Connection(app.conf.broker_url) as conn:
                for queue_name in ['default', 'high_priority', 'slow']:
                    try:
                        queue_lengths[queue_name] = conn.default_channel.queue_declare(
                            queue=queue_name, passive=True
                        ).message_count
                    except Exception:
                        queue_lengths[queue_name] = -1
        except Exception:
            pass

        workers = list(ping_result.keys())
        active_task_count = sum(len(tasks) for tasks in active.values())
        reserved_count = sum(len(tasks) for tasks in reserved.values())

        return {
            'alive': True,
            'ready': len(workers) > 0,
            'worker_count': len(workers),
            'workers': workers,
            'active_tasks': active_task_count,
            'reserved_tasks': reserved_count,
            'queue_depths': queue_lengths,
        }
    except Exception as e:
        return {'alive': False, 'error': str(e)}

The inspect API sends a broadcast message via the broker and collects replies — it's not a direct TCP health check. If the broker is down, inspect() raises a connection error. If no workers are listening (but the broker is up), inspect().ping() returns None or an empty dict. Both are failure conditions for an MCP server that depends on task execution. Monitor both broker connectivity and worker liveness as separate health signals.