Everything kubemq-celery needs to connect lives in two places: the broker URL (scheme, host, port, and an optional token) and broker_transport_options (a dictionary of fine-grained settings). This page is the complete reference for both, plus TLS/mTLS, async transport, and environment-variable patterns for containerized deployments.

Prerequisites

kubemq-celery installed (pip install kubemq-celery) alongside Celery in your app
A running KubeMQ broker reachable from the app (see the Docker snippet below)
import kubemq_celery present before Celery resolves the broker URL, so the kubemq*:// schemes are registered with Kombu

Broker URL

Format

The broker URL selects the transport scheme and points at your KubeMQ broker. All four schemes share the same [:token@]host[:port] shape:

kubemq://[:token@]host[:port]
kubemq+tls://[:token@]host[:port]
kubemq+async://[:token@]host[:port]
kubemq+async+tls://[:token@]host[:port]

Component	Description	Default
Scheme	`kubemq://`, `kubemq+tls://`, `kubemq+async://`, or `kubemq+async+tls://`	Required
Token	Authentication token (after `:`, before `@`)	None
Host	KubeMQ broker hostname	`localhost`
Port	KubeMQ gRPC port	`50000`

The token sits in the URL's password field — there is no username — so an authenticated URL reads kubemq://:my-token@host:50000. If you would rather keep the token out of the URL, pass it as the auth_token transport option instead (see Transport options).

Examples

broker_urls.py

# Basic connection (localhost, default port 50000)
app.conf.broker_url = "kubemq://localhost:50000"

# In-cluster Kubernetes service
app.conf.broker_url = "kubemq://kubemq.default.svc:50000"

# With authentication token (token in the password field)
app.conf.broker_url = "kubemq://:my-secret-token@kubemq.default.svc:50000"

# With TLS
app.conf.broker_url = "kubemq+tls://kubemq.default.svc:50000"

# TLS + authentication
app.conf.broker_url = "kubemq+tls://:my-token@kubemq.default.svc:50000"

# Async transport (for asyncio worker pools)
app.conf.broker_url = "kubemq+async://localhost:50000"

# Async + TLS
app.conf.broker_url = "kubemq+async+tls://kubemq.default.svc:50000"

import kubemq_celery must run before Celery resolves the broker URL — it registers all four kubemq*:// schemes with Kombu. Without it, Celery raises an "unknown transport" error regardless of the URL you set.

Async transport

The kubemq+async:// and kubemq+async+tls:// schemes use the native async KubeMQ clients — AsyncQueuesClient and AsyncPubSubClient — for non-blocking I/O. This is the right choice when your tasks are themselves I/O-bound (HTTP calls, database queries, other async frameworks like Starlette or Litestar) and you run Celery's asyncio worker pool.

Pair the async scheme with the --pool=asyncio worker flag:

terminal

celery -A myapp worker --pool=asyncio --loglevel=info

For synchronous, CPU-bound, or prefork-pool workloads, stick with the plain kubemq:// scheme — the async clients add no benefit there.

Transport options

All connection tuning beyond the URL goes through broker_transport_options:

transport_options.py

app.conf.broker_transport_options = {
    "wait_timeout": 1,
    "auth_token": "my-token",
    "dead_letter_queue": "celery-dead-letters",
    "max_receive_count": 3,
    "client_id_prefix": "celery",
    "tls_enabled": False,
    "tls_cert_file": "/path/to/cert.pem",
    "tls_key_file": "/path/to/key.pem",
    "tls_ca_file": "/path/to/ca.pem",
    "max_send_size": 4_194_304,
    "max_receive_size": 4_194_304,
    "message_expiration": 3600,
    "max_batch_size": 10,
    "fanout_max_retries": 5,
    "grpc_keepalive_time": 30,
    "grpc_keepalive_timeout": 10,
    "grpc_permit_without_calls": True,
}

Option reference

Option	Type	Default	Description
`wait_timeout`	`int`	`1`	Blocking receive timeout in seconds. Controls how long an internal receive waits for a message before returning empty. Must be less than Celery's `drain_events` timeout (default 2s). Increase for higher-latency environments.
`auth_token`	`str \| None`	`None`	KubeMQ authentication token. Overrides the token in the broker URL if both are set.
`dead_letter_queue`	`str`	`""`	KubeMQ channel name for dead letter messages. Messages that exceed `max_receive_count` are routed here.
`max_receive_count`	`int`	`0`	Maximum receive attempts before routing to the dead letter queue. Set to `0` to disable DLQ (messages redelivered indefinitely). Requires `dead_letter_queue` to be set.
`client_id_prefix`	`str`	`"celery"`	Prefix for KubeMQ client IDs. Each worker gets a unique ID: `{prefix}-queues-{random8}` and `{prefix}-pubsub-{random8}`.
`tls_enabled`	`bool`	`False`	Enable TLS for gRPC connections. Automatically set to `True` when using a `kubemq+tls://` URL scheme. Set explicitly to override URL-based detection.
`tls_cert_file`	`str`	`""`	Path to the client certificate file for mTLS authentication.
`tls_key_file`	`str`	`""`	Path to the client private key file for mTLS authentication.
`tls_ca_file`	`str`	`""`	Path to the CA certificate file for custom certificate authority verification.
`max_send_size`	`int`	`4_194_304`	Maximum gRPC send message size in bytes (default 4MB). Increase for large task payloads.
`max_receive_size`	`int`	`4_194_304`	Maximum gRPC receive message size in bytes (default 4MB). Increase for large task results.
`message_expiration`	`int`	`0`	Per-message TTL in seconds. Messages older than this are discarded by KubeMQ. Set to `0` to disable (no expiration). Maximum 86400 (24 hours). A task-level `expires` header takes precedence if set.
`max_batch_size`	`int`	`10`	Maximum messages per gRPC receive call. Higher values reduce round-trips but increase memory. Range: 1-100.
`fanout_max_retries`	`int`	`5`	Maximum re-subscription attempts when a fanout subscription (Events) encounters an error. Uses exponential backoff (1s, 2s, 4s, ... max 30s).
`grpc_keepalive_time`	`int`	`30`	Seconds between gRPC keepalive pings. Prevents idle connections from being dropped by load balancers or firewalls.
`grpc_keepalive_timeout`	`int`	`10`	Seconds to wait for a keepalive ping response before considering the connection dead.
`grpc_permit_without_calls`	`bool`	`True`	Send keepalive pings even when there are no active RPCs. Set to `True` for long-lived connections that may be idle between task bursts.

wait_timeout must stay below Celery's drain_events timeout (default 2s). The transport's blocking receive runs inside Celery's event drain loop; if wait_timeout meets or exceeds the drain timeout, the loop can deadlock instead of cycling. The default wait_timeout of 1 is safe — only raise it if you also raise the drain timeout, and keep it strictly below.

A max_send_size/max_receive_size example for large payloads and tuned keepalive:

grpc_options.py

import kubemq_celery  # noqa: F401 — registers the kubemq:// transport
from celery import Celery

app = Celery("grpc_options")
app.conf.update(
    broker_url="kubemq://localhost:50000",
    result_backend="kubemq://localhost:50000",
    broker_transport_options={
        # Keepalive: send a ping every 15 seconds to detect broken connections
        "grpc_keepalive_time": 15,
        # Wait up to 5 seconds for a keepalive response before considering it dead
        "grpc_keepalive_timeout": 5,
        # Allow 8MB messages (default is 4MB)
        "max_send_size": 8_388_608,
        "max_receive_size": 8_388_608,
    },
)

TLS and mTLS

There are two ways to enable TLS: the kubemq+tls:// URL scheme (which sets tls_enabled for you) or the tls_enabled transport option. Add tls_cert_file/tls_key_file for mutual authentication, and tls_ca_file to trust a custom CA.

Encrypts the gRPC channel; no client certificates required. The kubemq+tls:// scheme is all you need.

tls_connection.py

import os

import kubemq_celery  # noqa: F401 — registers the kubemq:// transport
from celery import Celery

app = Celery(
    "tls_connection",
    broker=os.environ.get("CELERY_BROKER_URL", "kubemq+tls://localhost:50000"),
    result_backend=os.environ.get("CELERY_RESULT_BACKEND", "kubemq+tls://localhost:50000"),
)

Mutual authentication: the client presents its own certificate and key, and verifies the broker against the CA. Set the same TLS files on the result backend if you use one.

mtls_connection.py

import os

import kubemq_celery  # noqa: F401 — registers the kubemq:// transport
from celery import Celery

CERT_DIR = os.environ.get("CERT_DIR", "/etc/kubemq/certs")

app = Celery("mtls_connection")
app.config_from_object(
    {
        "broker_url": "kubemq+tls://kubemq.default.svc:50000",
        "result_backend": "kubemq+tls://kubemq.default.svc:50000",
        "broker_transport_options": {
            "tls_cert_file": f"{CERT_DIR}/client.crt",
            "tls_key_file": f"{CERT_DIR}/client.key",
            "tls_ca_file": f"{CERT_DIR}/ca.crt",
        },
        "result_backend_transport_options": {
            "tls_enabled": True,
            "tls_cert_file": f"{CERT_DIR}/client.crt",
            "tls_key_file": f"{CERT_DIR}/client.key",
            "tls_ca_file": f"{CERT_DIR}/ca.crt",
        },
    }
)

When the broker presents a self-signed certificate, enable TLS explicitly and point tls_ca_file at the CA that signed it — no client certificate needed.

custom_ca.py

app.conf.broker_transport_options = {
    "tls_enabled": True,
    "tls_ca_file": "/certs/custom-ca.pem",
}

tls_enabled is set automatically when the URL scheme is kubemq+tls:// or kubemq+async+tls://. Set it explicitly only when you want TLS on a plain kubemq:// URL (for example, the custom-CA case above).

Environment-variable configuration

For containerized deployments, read the broker URL and token from the environment so the same image runs in every cluster. The conventional names are CELERY_BROKER_URL, CELERY_RESULT_BACKEND, and KUBEMQ_AUTH_TOKEN:

env_var_config.py

import os

import kubemq_celery  # noqa: F401 — registers the kubemq:// transport
from celery import Celery

app = Celery("myapp")
app.conf.update(
    broker_url=os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
    result_backend=os.environ.get("CELERY_RESULT_BACKEND", "kubemq://localhost:50000"),
    broker_transport_options={
        "auth_token": os.environ.get("KUBEMQ_AUTH_TOKEN"),
    },
)

Then set the variables in your deployment manifest, shell, or .env file:

terminal

export CELERY_BROKER_URL=kubemq://kubemq.default.svc:50000
export CELERY_RESULT_BACKEND=kubemq://kubemq.default.svc:50000
export KUBEMQ_AUTH_TOKEN=my-secret-token

celery -A myapp worker --loglevel=info

Full configuration example

A complete, production-shaped configuration that combines the broker, the result backend, and task and worker settings:

celeryconfig.py

import kubemq_celery  # noqa: F401 — registers the kubemq:// transport
from celery import Celery

app = Celery("myapp")

app.conf.update(
    # Broker
    broker_url="kubemq://kubemq.default.svc:50000",
    broker_transport_options={
        "wait_timeout": 1,
        "dead_letter_queue": "celery-dead-letters",
        "max_receive_count": 5,
        "client_id_prefix": "myapp",
        "max_send_size": 8_388_608,      # 8MB
        "max_receive_size": 8_388_608,   # 8MB
    },

    # Result backend
    result_backend="kubemq://kubemq.default.svc:50000",
    result_expires=86400,  # 24 hours

    # Task settings
    task_acks_late=False,
    task_default_queue="myapp-tasks",
    task_routes={
        "myapp.tasks.high_priority": {"queue": "high-priority"},
        "myapp.tasks.low_priority": {"queue": "low-priority"},
    },

    # Worker settings
    worker_prefetch_multiplier=1,
    worker_concurrency=4,
)

To try this against a local broker, start KubeMQ in Docker first — gRPC on 50000, the shared HTTP server (REST/health) on 9090:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Configuration