# Celery (/integrations/celery)



[`kubemq-celery`](https://pypi.org/project/kubemq-celery/) is a KubeMQ transport and queue-peek result backend for [Celery](https://docs.celeryq.dev/) task queues. It lets you run Celery on KubeMQ with a one-line configuration change — `broker="kubemq://localhost:50000"` — making KubeMQ the only Kubernetes-native, in-cluster Celery broker. The transport plugs into Kombu (Celery's messaging layer) and registers the `kubemq://` URL scheme, so your existing tasks, workers, and tooling keep working unchanged.

New to the idea? See [what is an integration](/integrations#what-an-integration-is) for how SDK-level integrations differ from the server-side [connectors](/connectors).

## Supported versions [#supported-versions]

| Requirement   | Supported versions                         |
| ------------- | ------------------------------------------ |
| Language      | Python `>= 3.10`                           |
| Celery        | `>= 5.4`                                   |
| Kombu         | `>= 5.4`                                   |
| KubeMQ SDK    | `kubemq >= 4.1.5`                          |
| KubeMQ broker | Reachable over native gRPC on port `50000` |
| Package       | `kubemq-celery` `1.1.0` (MIT)              |

## Why KubeMQ over Redis/RabbitMQ? [#why-kubemq-over-redisrabbitmq]

KubeMQ replaces visibility-timeout heuristics and broker plugins with native gRPC primitives, and runs as a first-class workload inside your Kubernetes cluster.

|                          | Redis                | RabbitMQ          | KubeMQ                      |
| ------------------------ | -------------------- | ----------------- | --------------------------- |
| **Acknowledgment**       | Visibility timeout   | Native AMQP ack   | Native gRPC ack             |
| **Delayed delivery**     | Client-side polling  | Plugin required   | Native `delay_in_seconds`   |
| **Kubernetes**           | External StatefulSet | External + Erlang | K8s-native, auto-clustering |
| **Connection stability** | TCP reset under load | Stable            | gRPC keep-alive             |
| **Setup complexity**     | Moderate             | High              | Low                         |

* **No visibility-timeout bugs** — messages are explicitly acknowledged or rejected over a gRPC stream, instead of relying on a timeout window that causes duplicates when set too low and stalls reprocessing when set too high.
* **Native delayed delivery** — `countdown` and `eta` map directly to KubeMQ's `delay_in_seconds` with zero polling overhead and no broker plugin.
* **Kubernetes-native** — an in-cluster broker with auto-clustering, gRPC keep-alive for long-lived connections, and KEDA-driven autoscaling, instead of an external StatefulSet or an Erlang-backed service.

## Install [#install]

Install from PyPI with `pip` or [uv](https://docs.astral.sh/uv/):

```bash
# pip
pip install kubemq-celery

# uv (recommended)
uv add kubemq-celery
```

Switching an existing app is a one-line change — `kubemq-celery` is a drop-in replacement for Redis or RabbitMQ:

```python title="tasks.py"
import kubemq_celery  # registers the kubemq:// transport
from celery import Celery

app = Celery("myapp", broker="kubemq://localhost:50000")

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

<Callout type="info">
  `import kubemq_celery` must run before Celery resolves the broker URL — it registers the `kubemq://` scheme with Kombu. Without it, Celery raises an "unknown transport" error.
</Callout>

Start a worker and send a task:

```bash
celery -A tasks worker --loglevel=info
```

```python
from tasks import add

add.delay(4, 6)
```

For the full walkthrough — including a local Docker broker and the result backend — see [Getting Started](/integrations/celery/tutorials/getting-started).

## Features [#features]

| Feature                | Description                                                                      |
| ---------------------- | -------------------------------------------------------------------------------- |
| **One-line setup**     | `broker="kubemq://host:50000"` — drop-in replacement for Redis/RabbitMQ          |
| **Native ack/nack**    | Messages are explicitly acknowledged over gRPC — no visibility-timeout bugs      |
| **Delayed delivery**   | `countdown` and `eta` map to KubeMQ's native `delay_in_seconds` (no polling)     |
| **Per-message TTL**    | `message_expiration` transport option for automatic message expiration           |
| **Dead letter queue**  | Built-in DLQ via `max_receive_count` + `dead_letter_queue` transport options     |
| **Batch receive**      | `max_batch_size` fetches multiple messages per gRPC call for higher throughput   |
| **Queue-peek results** | Optional result backend using non-destructive peek — no external Redis/DB needed |
| **Full monitoring**    | Flower, `celery inspect`, `celery control` — all work via KubeMQ Events fanout   |
| **TLS + mTLS**         | `kubemq+tls://` for encrypted gRPC connections, mTLS for mutual authentication   |
| **Async transport**    | `kubemq+async://` for native asyncio I/O with `--pool=asyncio` workers           |
| **KEDA autoscaling**   | Kubernetes-native broker with auto-clustering and KEDA-driven scaling            |
| **Auto-registration**  | `import kubemq_celery` registers the `kubemq://` URL scheme automatically        |

## Architecture [#architecture]

A Celery worker talks to KubeMQ through Kombu's virtual transport layer. `kubemq-celery` implements a `Channel` that extends Kombu's `virtual.Channel` (with `supports_fanout = True`) and maps Kombu's storage primitives onto KubeMQ's gRPC API on port `50000`. Task messages travel over KubeMQ **Queues** (with native ack/nack and `delay_in_seconds`), while pidbox and monitoring traffic — Flower, `celery inspect`, `celery control` — use KubeMQ **Events** fanout. When the optional result backend is enabled, results are written as Queue messages and read back with a non-destructive peek, so multiple callers can fetch the same result without consuming it.

<Mermaid
  chart="`
flowchart LR
App[&#x22;Celery App / Worker<br/>(@app.task)&#x22;]
Kombu[&#x22;Kombu Virtual Transport<br/>(Channel, supports_fanout)&#x22;]
KMQ{{&#x22;KubeMQ Broker<br/>gRPC :50000&#x22;}}
Q[&#x22;Queues<br/>(tasks + peek results)&#x22;]
E[&#x22;Events<br/>(pidbox / monitoring fanout)&#x22;]

App -- &#x22;kubemq:// URL&#x22; --> Kombu
Kombu -- &#x22;gRPC :50000&#x22; --> KMQ
KMQ --> Q
KMQ --> E

class App external
class Kombu aiway
class KMQ,Q,E broker
`"
/>

*A Celery worker speaks `kubemq://` through Kombu; tasks and peek-read results ride KubeMQ Queues, while pidbox and monitoring traffic fan out over KubeMQ Events.*

## Usage [#usage]

<Cards>
  <Card title="Canvas Workflows" href="/integrations/celery/how-to/canvas-workflows" description="Compose tasks with chains, groups, chords, maps, and chunks on KubeMQ." />

  <Card title="Scheduling & Delayed Delivery" href="/integrations/celery/how-to/scheduling" description="Use countdown, eta, and Celery Beat — backed by KubeMQ's native delay_in_seconds." />

  <Card title="Result Backend" href="/integrations/celery/how-to/result-backend" description="Store and retrieve task results with the queue-peek backend — no external Redis or database." />
</Cards>

## Quick Links [#quick-links]

<Cards>
  <Card title="Getting Started" href="/integrations/celery/tutorials/getting-started" description="Run a Celery worker on KubeMQ in under 5 minutes." />

  <Card title="Concepts" href="/integrations/celery/concepts" description="How the transport maps Celery and Kombu onto KubeMQ Queues and Events." />

  <Card title="Configuration" href="/integrations/celery/how-to/configuration" description="All broker transport options, TLS/mTLS, async, and result backend settings." />

  <Card title="API Reference" href="/integrations/celery/reference/configuration" description="URL schemes, transport options, and result backend reference." />
</Cards>

<Callout type="info">
  **Requirements** — Python `>= 3.10`, Celery `>= 5.4`, Kombu `>= 5.4`, and the `kubemq` SDK `>= 4.1.5`. Current version: `1.1.0`. You also need a reachable KubeMQ broker (Docker, Kubernetes, or standalone).
</Callout>

New to KubeMQ? Start with the [KubeMQ Getting Started guide](/deploy) for core concepts like Queues and Events before wiring up Celery.