Celery FastAPI

Automatic REST API generation for Celery tasks with FastAPI. This package seamlessly bridges Celery and FastAPI, automatically creating REST endpoints for all your registered Celery tasks.

Why celery-fastapi?

Without it, exposing Celery to other services means hand-writing a FastAPI route, a request model and a status route for every task, then keeping them in sync with the task signatures. celery-fastapi derives all of that from your Celery app:

It is a thin bridge, not a new task queue: workers, brokers and results stay plain Celery.

Features

• 🚀 Automatic endpoint generation — REST APIs created automatically for all Celery tasks
• 🔧 Zero configuration — Works out of the box with sensible defaults
• 📊 Task monitoring — Built-in endpoints for task status, revocation, and worker info
• 🎯 App-scoped operations — Only manages tasks from your specific Celery app, not the entire cluster
• 🖥️ CLI support — Run as a standalone server from command line
• 📦 Modular design — Use as a library or standalone application
• 🔄 Queue-aware routing — Respects Celery queue assignments
• 📝 OpenAPI documentation — Full Swagger/ReDoc support
• ⚡ Full Celery options — All task options (countdown, eta, priority, etc.)
• 🧮 Batch execution — Submit groups of tasks in a single request via /tasks/batch
• 🔗 Workflow primitives — Chain and chord orchestration via /tasks/chain and /tasks/chord
• 🛡️ Input validation — Pydantic-driven validation on task_name/queue at trust boundary
• 📦 Pydantic task params — Tasks annotated with BaseModel subclasses get first-class payload models and OpenAPI schemas
• 🪝 Bridge hooks — pre_hooks/post_hooks run around task dispatch (auth, audit, notify)
• 🗺️ Discovery & mapping — Hide tasks with exclude and rename public names with name_mapping
• 🔀 Custom error mapping — Map exception types to HTTP status codes via error_mapping
• 🧩 Middleware & dependencies — Inject FastAPI middleware and Depends providers
• 💾 Pluggable rate limiting — In-memory by default; bring your own store via BaseRateLimitStorage
• 🔌 WebSocket streaming — Live task status updates via /tasks/{task_id}/ws
• 📡 Server-Sent Events — Proxy-friendly status stream via /tasks/{task_id}/events
• 🔐 Secure by default — Destructive endpoints are opt-in; one api_key protects the rest
• ♻️ Idempotency keys — Retries with the same Idempotency-Key never double-submit
• 📈 Prometheus metrics & OpenTelemetry — Opt-in /metrics and dispatch spans
• 🗓️ Beat schedule listing — Read-only /schedules
• 🤖 MCP server (experimental) — Expose tasks as tools for AI agents
• 🧾 OpenAPI export — celery-fastapi openapi for client generation in CI

Requirements

• Python 3.11 – 3.16 (3.11–3.14 tested in CI; 3.15 (release candidate) and 3.16 run as non-blocking jobs once available)
• FastAPI 0.100.0+
• Celery 5.3.0+ (tested up to 5.6.3)

Installation

# Basic installation
pip install celery-fastapi

# With CLI support
pip install celery-fastapi[cli]

# With uvicorn server
pip install celery-fastapi[server]

# With gunicorn for production
pip install celery-fastapi[gunicorn]

# With Redis broker
pip install celery-fastapi[redis]

# With RabbitMQ broker
pip install celery-fastapi[rabbitmq]

# All extras (recommended for production)
pip install celery-fastapi[all]

Or with Poetry:

poetry add celery-fastapi

Quick Start

As a Python Module

from celery import Celery
from celery_fastapi import CeleryFastAPIBridge, create_app

celery_app = Celery("tasks", broker="redis://localhost:6379/0")


@celery_app.task
def add(x, y):
    return x + y


# Option 1: Using create_app factory
app = create_app(celery_app)

# Option 2: Using the Bridge class for more control
from fastapi import FastAPI

fastapi_app = FastAPI(title="My Task API")
bridge = CeleryFastAPIBridge(celery_app, fastapi_app)
bridge.register_routes()

Run with uvicorn:

uvicorn myapp:app --reload

Try it in five minutes

cd examples/quickstart && docker compose up --build   # then open http://localhost:8000/docs

See examples/quickstart and, for Kubernetes health probes, examples/k8s-sidecar.yaml.

Using the CLI

celery-fastapi serve examples.celery_app:celery_app --port 8000 --reload
celery-fastapi serve examples.celery_app:celery_app -w 4 --host 0.0.0.0
celery-fastapi routes examples.celery_app:celery_app
celery-fastapi tasks examples.celery_app:celery_app
celery-fastapi workers examples.celery_app:celery_app

Security

celery-fastapi is secure by default, because an HTTP API in front of a task queue is a powerful thing to expose:

• Admin endpoints are off. POST /purge, POST /trigger (runs any task by name), DELETE /tasks/{id} and POST /tasks/batch/revoke only exist when you pass enable_admin_endpoints=True (--enable-admin on the CLI).
• One key protects everything. api_key="..." (CELERY_FASTAPI_API_KEY) requires the X-API-Key header (configurable) on every endpoint except the /healthz and /ping probes. WebSockets also accept ?api_key=. Keys are compared in constant time.
• Your own auth still works. dependencies=[...] are applied to every HTTP endpoint except the probes (use them for OAuth/JWT).
• Only your tasks. Per-task, batch, chain and chord endpoints refuse task names outside the app; task and queue names are validated.
• Rate limiting covers every endpoint that submits work.

Always run it behind TLS, and prefer environment variables over command-line arguments for secrets. See SECURITY.md for reporting issues.

Upgrading from 0.1.5: if you call /purge, /trigger or the revoke endpoints, add enable_admin_endpoints=True. Everything else keeps working.

API Endpoints

All endpoints are prefixed with the configured prefix (empty by default).

Task Execution

POST /{task_name_with_slashes} — Execute a task.

{
  "args": [1, 2],
  "kwargs": {},
  "countdown": 60,
  "priority": 5,
  "queue": "high_priority"
}

{
  "task_id": "abc123-def456-...",
  "status": "PENDING"
}

Send an Idempotency-Key header to make retries safe: the same key with the same body returns the original task_id (response header Idempotent-Replayed: true); the same key with a different body returns 409. Keys are remembered in memory for idempotency_ttl seconds (24h by default), per process.

POST /trigger (admin) — Trigger any task by name (queue required here).

{
  "task_name": "myapp.add",
  "queue": "celery",
  "args": [1, 2]
}

Pydantic task parameters

When a task is annotated with a pydantic.BaseModel subclass, the generated payload model uses that model directly, so nested schemas appear in OpenAPI:

class Item(BaseModel):
    name: str
    qty: int = 1


@celery_app.task(name="myapp.process")
def process(payload: Item) -> dict:
    return payload.model_dump()

{"payload": {"name": "widget", "qty": 3}}

Workflow Primitives

POST /tasks/chain — Run tasks sequentially, passing results forward.

{
  "tasks": [
    {"task_name": "myapp.add", "args": [1, 2]},
    {"task_name": "myapp.add", "args": [3, 4]}
  ]
}

POST /tasks/chord — Run a header group, then a callback once all complete.

{
  "header": [
    {"task_name": "myapp.add", "args": [1, 2]},
    {"task_name": "myapp.add", "args": [3, 4]}
  ],
  "callback": "myapp.greet"
}

Batch Execution

POST /tasks/batch — Submit a group of tasks.

POST /tasks/batch/revoke (admin) — Revoke tasks by list of IDs.

Task Status

• GET /tasks/{task_id} — Full task status (state, result, traceback, date_done).
• GET /tasks/{task_id}/result — Task result only.
• GET /tasks — List active, scheduled, reserved, revoked tasks (filtered to this app). Query: worker, name (substring), limit and offset (per worker list).
• GET /tasks/{task_id}/events — Server-Sent Events status stream.
• DELETE /tasks/{task_id} (admin) — Revoke a single task.

A task that is unknown or has not started reports state: "PENDING" (HTTP 200), which matches Celery; GET /tasks/{id}/result answers 202 until it finishes.

Discovery & Management

• GET /available-tasks — List tasks registered in THIS app (respects name_mapping/exclude).
• GET /workers — List active workers, filtered to this app's tasks (includes active_queues).
• GET /schedules — Celery Beat entries (beat_schedule) for exposed tasks. Read-only: enabling/disabling entries depends on your scheduler and is not managed here.
• POST /purge (admin) — Purge all pending tasks.

Health Check

• GET /healthz — Health check for local Celery worker.
• GET /ping — Ping local Celery worker.

Observability

• GET /metrics (opt-in: enable_metrics=True) — Prometheus text format: celery_fastapi_tasks_submitted_total{task}, ..._dispatch_errors_total{task}, ..._rate_limited_total, ..._broker_up, ..._workers_online.
• OpenTelemetry (opt-in: enable_tracing=True, pip install celery-fastapi[otel]) — each dispatch runs in a celery_fastapi.send_task span with celery.task_name and celery.queue attributes, using whatever tracer provider your app configured.

MCP (experimental)

enable_mcp=True adds POST /mcp, a minimal Model Context Protocol endpoint (JSON-RPC over HTTP: initialize, tools/list, tools/call). Every exposed task becomes a tool whose input schema comes from the task signature, plus a get_task_status tool. It honours api_key, dependencies, hooks and the rate limit. Streaming/SSE transports and resources/prompts are not implemented.

Typed clients

celery-fastapi openapi myapp.celery:app -o openapi.json
npx @openapitools/openapi-generator-cli generate -i openapi.json -g typescript-fetch -o client

WebSocket Streaming

WS /tasks/{task_id}/ws — Stream task status updates as JSON frames.

Configuration

CeleryFastAPIBridge Options

from celery_fastapi import (
    CeleryFastAPIBridge,
    BaseRateLimitStorage,
)

bridge = CeleryFastAPIBridge(
    celery_app=celery_app,
    fastapi_app=fastapi_app,  # Optional
    prefix="/api/v1",  # URL prefix
    include_status_endpoints=True,
    task_filter=lambda name: not name.startswith("internal."),
    rate_limit=100,  # req/min per client
    rate_limit_storage=None,  # BaseRateLimitStorage instance
    enable_admin_endpoints=False,  # /purge, /trigger, revoke
    api_key=None,  # require X-API-Key everywhere except /healthz, /ping
    enable_metrics=False,  # /metrics (Prometheus)
    enable_mcp=False,  # /mcp (experimental)
    enable_tracing=False,  # OpenTelemetry spans
    idempotency_ttl=86400,  # seconds an Idempotency-Key is remembered
    exclude={"internal.secret"},  # Hide from API
    name_mapping={"my.add": "public_add"},  # Rename in listings
    middleware=[my_http_middleware],
    dependencies=[auth_provider],
    pre_hooks=[audit_hook],  # receive payload
    post_hooks=[notify_hook],  # receive response
    error_mapping={ValueError: 422},
)

Pluggable Rate Limit Storage

Rate limiting is backed by BaseRateLimitStorage. The default is in-memory (single process). For several workers use the bundled Redis store, or implement the ABC for another shared store:

from redis import Redis
from celery_fastapi import CeleryFastAPIBridge, RedisRateLimitStorage

bridge = CeleryFastAPIBridge(
    celery_app,
    rate_limit=100,
    rate_limit_storage=RedisRateLimitStorage(Redis()),
)

pip install celery-fastapi[redis] provides the redis package.

Integration with Existing FastAPI App

from fastapi import FastAPI
from celery_fastapi import CeleryFastAPIBridge

app = FastAPI()


@app.get("/health")
def health_check():
    return {"status": "healthy"}


bridge = CeleryFastAPIBridge(celery_app, app, prefix="/celery")
bridge.register_routes()

CLI Reference

celery-fastapi serve        Start the server (uvicorn)
celery-fastapi serve-gunicorn  Start with Gunicorn
celery-fastapi openapi      Print the OpenAPI schema (no server)
celery-fastapi routes       List all generated routes
celery-fastapi tasks        List registered Celery tasks
celery-fastapi workers      Show active workers

serve, serve-gunicorn, routes and openapi accept --enable-admin, --api-key (or CELERY_FASTAPI_API_KEY), --metrics and --mcp.

celery-fastapi serve examples.celery_app:celery_app \
    --host 0.0.0.0 --port 8000 --reload --workers 4 \
    --log-level info --ssl-keyfile key.pem --ssl-certfile cert.pem

Development

git clone https://github.com/karailker/celery-fastapi.git
cd celery-fastapi
poetry install --extras all
poetry run pytest
poetry run ruff check .
poetry run mypy celery_fastapi

Integration test suite (broker/backend matrix)

The repo ships a docker-compose.yml that brings up every broker/backend the test matrix exercises:

docker compose up -d                              # redis, rabbitmq, postgres, mysql, memcached, mongodb
poetry run pytest tests/test_integration_broker_backend.py

Then start one worker for the end-to-end test and run it:

poetry run celery -A tests.broker_workers:live_app worker --loglevel=info
poetry run pytest tests/test_integration_broker_backend.py::test_live_redis_broker_backend

The matrix covers all stable Celery brokers (Redis, RabbitMQ) crossed with nine result backends (redis, rpc, cache+memory, cache+memcached, db+sqlite, db+postgresql, db+mysql, mongodb, filesystem) — 18 combinations, each verified for bridge construction, broker dispatch, and result-backend round-trip. The rpc backend skips the round-trip layer (it needs a live reply consumer); the single live worker test uses Redis end to end. Missing drivers skip individually.

Contributing & project

• CONTRIBUTING.md — dev setup, checks, PR conventions
• SECURITY.md — how to report a vulnerability
• ROADMAP.md — what is shipped and what is next
• CHANGELOG.md

License

MIT License — see LICENSE.