Scenario

A FastAPI web tier accepts HTTP requests and offloads slow work to background tasks. Each request dispatches a Celery task with task.delay(...) and immediately returns a task id, so the client never blocks on the work itself. A separate endpoint lets the client poll task status and fetch the result once it is ready.

The twist: there is no Redis and no RabbitMQ anywhere in the stack. Both the Celery broker and the result backend are KubeMQ, reached through the kubemq:// transport. The web pods and the worker pods talk to a single KubeMQ broker, and a KEDA ScaledObject watches the queue depth on that broker to scale the worker Deployment up during request bursts and back down when the queue drains.

This scenario walks the full path: a shared Celery app, the FastAPI endpoints, an optional WebSocket progress stream, the worker tuning profile for API workloads, containerization with docker-compose for local runs, the Kubernetes Deployment, the KEDA trigger, and the reliability settings that keep fast API tasks honest.

Architecture

The web tier and the workers are separate Deployments that share one KubeMQ broker. FastAPI pods only ever produce tasks and read results; the worker Deployment consumes tasks, and KEDA scales it from the queue depth that KubeMQ reports.

Shared Celery app

The web tier and the workers import the same Celery app module. Importing kubemq_celery registers the kubemq:// transport with Kombu, and both the broker and the result backend are read from environment variables so the identical image runs locally and in-cluster. This is the app from examples/integrations/fastapi_integration.py.

tasks.py

from __future__ import annotations

import os
import time
from typing import Any

from celery import Celery

import kubemq_celery  # noqa: F401  registers the kubemq:// transport

celery_app = Celery(
    "fastapi_integration",
    broker=os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
    result_backend=os.environ.get("CELERY_RESULT_BACKEND", "kubemq://localhost:50000"),
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    result_expires=3600,
    task_track_started=True,
    worker_prefetch_multiplier=1,
)


@celery_app.task(bind=True)
def process_data(self, data: str, options: dict[str, Any] | None = None) -> dict:
    """Process data with configurable transformations."""
    options = options or {}
    self.update_state(state="PROGRESS", meta={"step": "processing", "percent": 50})
    time.sleep(1)

    result = data
    if options.get("uppercase"):
        result = result.upper()
    if options.get("reverse"):
        result = result[::-1]

    return {
        "input": data,
        "output": result,
        "options_applied": list(options.keys()),
    }

task_track_started=True makes Celery emit a STARTED state the moment a worker picks up the task, which gives the status endpoint a useful intermediate state between PENDING and SUCCESS. import kubemq_celery must run before the app connects, or Kombu will not recognize the kubemq:// scheme.

FastAPI endpoints

FastAPI is the producer. The POST endpoint dispatches the task and returns its id without waiting; the GET endpoint builds an AsyncResult and maps the Celery state onto a response model. This is examples/integrations/fastapi_integration.py.

api.py

from typing import Any

from celery.result import AsyncResult
from fastapi import FastAPI
from pydantic import BaseModel

from tasks import celery_app, process_data


class TaskRequest(BaseModel):
    data: str
    options: dict[str, Any] | None = None


class TaskResponse(BaseModel):
    task_id: str
    status: str
    detail: str


class TaskStatus(BaseModel):
    task_id: str
    status: str
    result: Any | None = None
    error: str | None = None
    progress: dict[str, Any] | None = None


api = FastAPI(
    title="KubeMQ Celery API",
    description="FastAPI + Celery with KubeMQ transport",
    version="1.0.0",
)


@api.post("/tasks/process", response_model=TaskResponse)
async def submit_task(request: TaskRequest) -> TaskResponse:
    """Submit a data processing task to the KubeMQ broker."""
    result = process_data.delay(request.data, request.options)
    return TaskResponse(
        task_id=result.id,
        status="PENDING",
        detail="Task submitted to KubeMQ broker",
    )


@api.get("/tasks/{task_id}", response_model=TaskStatus)
async def get_task_status(task_id: str) -> TaskStatus:
    """Check the status of a submitted task."""
    result = AsyncResult(task_id, app=celery_app)

    if result.state == "PROGRESS":
        return TaskStatus(task_id=task_id, status="PROGRESS", progress=result.info)
    elif result.state == "SUCCESS":
        return TaskStatus(task_id=task_id, status="SUCCESS", result=result.result)
    elif result.state == "FAILURE":
        return TaskStatus(task_id=task_id, status="FAILURE", error=str(result.result))
    else:
        return TaskStatus(task_id=task_id, status=result.state)

The dispatch-and-poll flow looks like this from the client side:

dispatch and poll

# Submit a task -> returns a task id
curl -X POST http://localhost:8000/tasks/process \
  -H "Content-Type: application/json" \
  -d '{"data": "hello world", "options": {"uppercase": true}}'
# {"task_id": "a1b2c3...", "status": "PENDING", "detail": "Task submitted to KubeMQ broker"}

# Poll status with the returned id
curl http://localhost:8000/tasks/a1b2c3...
# {"task_id": "a1b2c3...", "status": "SUCCESS", "result": {"input": "hello world", "output": "HELLO WORLD", ...}}

Real-time progress

Polling is fine for short tasks, but a longer task can stream its progress to the browser over a WebSocket instead. The task reports fine-grained progress with self.update_state(...), and the WebSocket endpoint reads those PROGRESS states from the result backend and forwards them to the client until the task reaches a terminal state. This combines examples/integrations/fastapi_websocket_progress.py with the progress task from examples/monitoring/progress_tracking.py.

ws_progress.py

import asyncio

from celery.result import AsyncResult
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

from tasks import celery_app


@celery_app.task(bind=True)
def long_running_task(self, steps: int = 10) -> dict:
    """Task that reports progress at each step."""
    import time

    for i in range(1, steps + 1):
        self.update_state(
            state="PROGRESS",
            meta={
                "current": i,
                "total": steps,
                "percent": int(i / steps * 100),
                "message": f"Processing step {i}/{steps}",
            },
        )
        time.sleep(1)
    return {"steps_completed": steps, "status": "done"}


api = FastAPI()


@api.post("/tasks/process")
async def submit_task(steps: int = 5) -> dict:
    result = long_running_task.delay(steps)
    return {"task_id": result.id, "status": "PENDING", "ws_url": f"/ws/progress/{result.id}"}


@api.websocket("/ws/progress/{task_id}")
async def websocket_progress(websocket: WebSocket, task_id: str):
    """Stream task progress updates over WebSocket."""
    await websocket.accept()
    max_iterations = 300
    try:
        for _ in range(max_iterations):
            result = AsyncResult(task_id, app=celery_app)

            if result.state == "PROGRESS":
                await websocket.send_json({"status": "PROGRESS", "progress": result.info})
            elif result.state == "SUCCESS":
                await websocket.send_json({"status": "SUCCESS", "result": result.result})
                break
            elif result.state == "FAILURE":
                await websocket.send_json({"status": "FAILURE", "error": str(result.result)})
                break
            else:
                await websocket.send_json({"status": result.state})

            await asyncio.sleep(0.5)
    except WebSocketDisconnect:
        pass

Each update_state() overwrites the task's entry on the KubeMQ result channel, and the client reads the latest value by peeking that channel — the backend does not keep a history of progress snapshots. The endpoint caps its read loop at max_iterations so a never-finishing task cannot hold the socket open forever.

Worker config for API workload

Tasks dispatched from web requests need to start fast, not batch efficiently. The low-latency profile from docs/performance.md raises concurrency so more requests run in parallel, keeps prefetch_multiplier=1 so no worker hoards queued tasks while it is busy, and sets max_batch_size=1 to pull one message at a time for the lowest possible dispatch latency.

worker_config.py

celery_app.conf.update(
    worker_concurrency=8,
    worker_prefetch_multiplier=1,
    broker_transport_options={
        "wait_timeout": 1,
        "max_batch_size": 1,  # minimize latency
    },
)

Keep wait_timeout below Celery's drain_events timeout (default 2s) to avoid transport deadlocks. wait_timeout: 1 satisfies this.

Containerize and run locally

Package the shared app into one image and run the same image as both the web tier and the worker, differing only by command. The Dockerfile from examples/kubernetes/Dockerfile installs dependencies with uv.

Dockerfile

FROM python:3.12-slim

WORKDIR /app

# Install uv and dependencies
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
COPY pyproject.toml .
RUN uv pip install --system --no-cache .

# Copy application code
COPY src/ src/
COPY examples/ examples/

ENV PYTHONPATH=/app/src:/app/examples

# Health check
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD python -c "import kubemq_celery; print('ok')" || exit 1

CMD ["celery", "-A", "basic_task", "worker", "--loglevel=info"]

Before touching Kubernetes, validate the whole stack with docker-compose. This compose file (adapted from examples/kubernetes/docker-compose.yaml) brings up a KubeMQ broker with a health check and a Celery worker that waits for the broker to be ready.

docker-compose.yaml

services:
  kubemq:
    image: europe-docker.pkg.dev/kubemq/images/kubemq:next
    ports:
      - "50000:50000"  # gRPC
      - "9090:9090"    # Shared HTTP server (REST/health)
      - "8080:8080"    # Management API dashboard
    environment:
      - KUBEMQ_LOG_LEVEL=info
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:9090/health"]
      interval: 10s
      timeout: 5s
      retries: 3
      start_period: 10s

  worker:
    build:
      context: .
      dockerfile: Dockerfile
    command: celery -A tasks worker --loglevel=info -Q celery --concurrency=8
    environment:
      - CELERY_BROKER_URL=kubemq://kubemq:50000
      - CELERY_RESULT_BACKEND=kubemq://kubemq:50000
    depends_on:
      kubemq:
        condition: service_healthy
    restart: unless-stopped

  api:
    build:
      context: .
      dockerfile: Dockerfile
    command: uvicorn api:api --host 0.0.0.0 --port 8000
    ports:
      - "8000:8000"
    environment:
      - CELERY_BROKER_URL=kubemq://kubemq:50000
      - CELERY_RESULT_BACKEND=kubemq://kubemq:50000
    depends_on:
      kubemq:
        condition: service_healthy
    restart: unless-stopped

run locally

docker compose up -d
# KubeMQ dashboard: http://localhost:8080  (Management API — watch queue depth here)
# FastAPI:          http://localhost:8000/docs

Deploy to Kubernetes

In-cluster, the broker runs as a StatefulSet and the workers as a separate Deployment. Workers reach the broker through its Kubernetes service DNS name, and the readiness and liveness probes use celery inspect ping, which exercises the transport's verify_connection() against the broker. This worker Deployment is from docs/kubernetes.md.

worker-deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: celery-worker
  namespace: default
  labels:
    app: celery-worker
spec:
  replicas: 3
  selector:
    matchLabels:
      app: celery-worker
  template:
    metadata:
      labels:
        app: celery-worker
    spec:
      containers:
        - name: worker
          image: myapp:latest
          command:
            - celery
            - -A
            - tasks
            - worker
            - --loglevel=info
            - --concurrency=8
          env:
            - name: CELERY_BROKER_URL
              value: "kubemq://kubemq.default.svc:50000"
            - name: CELERY_RESULT_BACKEND
              value: "kubemq://kubemq.default.svc:50000"
          resources:
            requests:
              cpu: 250m
              memory: 256Mi
            limits:
              cpu: 1000m
              memory: 512Mi
          readinessProbe:
            exec:
              command:
                - celery
                - -A
                - tasks
                - inspect
                - ping
                - --timeout=5
            initialDelaySeconds: 30
            periodSeconds: 30
            timeoutSeconds: 10
          livenessProbe:
            exec:
              command:
                - celery
                - -A
                - tasks
                - inspect
                - ping
                - --timeout=5
            initialDelaySeconds: 60
            periodSeconds: 60
            timeoutSeconds: 10

The broker itself runs as a StatefulSet exposing gRPC on 50000, the shared HTTP server (REST/health) on 9090, and the Management API dashboard on 8080. Inside the cluster, workers and web pods address it as kubemq.default.svc:50000.

kubemq-statefulset.yaml

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kubemq
  namespace: default
spec:
  serviceName: kubemq
  replicas: 3  # HA cluster
  selector:
    matchLabels:
      app: kubemq
  template:
    metadata:
      labels:
        app: kubemq
    spec:
      containers:
        - name: kubemq
          image: europe-docker.pkg.dev/kubemq/images/kubemq:next
          ports:
            - containerPort: 50000
              name: grpc
            - containerPort: 9090
              name: http
            - containerPort: 8080
              name: dashboard
          readinessProbe:
            tcpSocket:
              port: 50000
            initialDelaySeconds: 5
            periodSeconds: 10
          livenessProbe:
            tcpSocket:
              port: 50000
            initialDelaySeconds: 15
            periodSeconds: 20

The worker probes use celery inspect ping, which connects through the broker. Give the broker StatefulSet a generous initialDelaySeconds (the worker readiness probe waits 30s) so the broker is reachable before workers start failing health checks during a cold start.

Autoscale with KEDA

HPA scales on CPU and memory, which lag behind a queue that is filling from a request burst. KEDA scales on the metric that actually matters here — the KubeMQ queue depth. The kubemq trigger watches the celery channel and adds workers when more than 10 messages are pending, scaling between 1 and 20 replicas. This is examples/kubernetes/keda-scaler.yaml.

keda-scaler.yaml

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: celery-worker-scaler
  namespace: default
  labels:
    app: celery-worker
    component: autoscaling
spec:
  scaleTargetRef:
    name: celery-worker          # must match Deployment name
  minReplicaCount: 1             # always keep at least 1 worker
  maxReplicaCount: 20            # maximum workers
  pollingInterval: 10            # check queue depth every 10 seconds
  cooldownPeriod: 60             # wait 60s before scaling down
  triggers:
    - type: kubemq
      metadata:
        # KubeMQ broker address (cluster-internal)
        address: "kubemq.default.svc:50000"
        # Queue channel to monitor (Celery default queue)
        channel: "celery"
        # Scale threshold: add workers when queue depth > 10
        queueLength: "10"

Install KEDA and apply the scaler:

install KEDA

helm repo add kedacore https://kedacore.github.io/charts
helm install keda kedacore/keda --namespace keda --create-namespace
kubectl apply -f keda-scaler.yaml

When a request burst hits the FastAPI tier, the celery channel depth climbs, KEDA notices within pollingInterval (10s) and scales the worker Deployment up; once the queue drains it scales back down after cooldownPeriod (60s). Watch the depth live in the KubeMQ Management API dashboard at port 8080 (or query the REST /queue/info endpoint on the shared HTTP server at 9090).

Reliability notes

A few defaults matter for fast, request-driven tasks:

acks_early is the right default. The message is acknowledged before the task runs, so a quick API task is not re-delivered if a worker dies mid-execution. For at-least-once delivery on a critical task, opt in per task with acks_late=True and reject_on_worker_lost=True — but be aware that KubeMQ's transaction timeout can expire for long-running tasks (> 60s), so use acks_early or make the task idempotent for those.
reliability.py
```
celery_app.conf.task_acks_late = False  # default — good for fast API tasks

@celery_app.task(acks_late=True, reject_on_worker_lost=True)
def critical_payment(payment_id: str) -> dict:
    """Process payment — must not be lost. Idempotent via payment_id dedup."""
    ...
```
Configure a dead letter queue for poison messages. The DLQ operates at the broker level (receive count) and is independent of Celery's task-level max_retries; the two work together so a message that repeatedly fails to process lands in a DLQ channel instead of looping forever.
Keep result_expires at or below 86400. For a polling API, results only need to live long enough for the client to fetch them. The shared app sets result_expires=3600 (one hour); the documented maximum is 86400 (24 hours). Bounding this keeps the result backend from accumulating stale entries.

FastAPI Task Dispatch with KEDA Autoscaling

Scenario

Architecture

Shared Celery app

FastAPI endpoints

Real-time progress

Worker config for API workload

Containerize and run locally

Deploy to Kubernetes

Autoscale with KEDA

Reliability notes

Next steps

Kubernetes Deployment

Error Handling

API Reference

On this page