This page is the configuration reference for the kubemq-celery package: broker URL schemes, the public API surface, the Celery settings it honors, environment variables, monitoring and control commands, known limitations, and the exceptions it raises. The full broker_transport_options and result_backend_transport_options key tables live on Transport Options. For a guided introduction see Getting Started; for worked configuration recipes see the Configuration guide.

Package facts

Field	Value
Package name	`kubemq-celery`
Version	`1.1.0`
License	MIT
Python	`>= 3.10`
Celery	`>= 5.4`
Kombu	`>= 5.4`
KubeMQ SDK	`kubemq >= 4.1.5`

Install from PyPI with pip or uv:

pip install kubemq-celery
# or
uv add kubemq-celery

The three runtime dependencies (kubemq, celery, kombu) are pulled in automatically. Optional extras for the example apps are listed under Optional dependency extras.

Public API

The package exposes a deliberately small surface. Importing kubemq_celery does two things: it makes Transport and KubeMQResultBackend importable, and — as a side effect of the import — it registers the kubemq URL schemes with Kombu and the kubemq backend alias with Celery.

import kubemq_celery

kubemq_celery.__version__          # "1.1.0"
kubemq_celery.Transport            # the Kombu transport class
kubemq_celery.KubeMQResultBackend  # the queue-peek result backend

Name	Kind	Description
`Transport`	class	Kombu transport powering `kubemq://` broker URLs. Registered as the `kubemq` and `kubemq+tls` aliases.
`KubeMQResultBackend`	class	Queue-peek result backend. Registered as the `kubemq` backend alias.
`__version__`	str	The installed package version (`"1.1.0"`).

The transport and backend aliases are registered as a side effect of the import. import kubemq_celery must run before Celery resolves the broker URL, or Celery raises ValueError: Unknown transport 'kubemq'. Keep the import at the top of your app module even if your editor flags it as unused.

Under the hood the import registers the following aliases:

src/kubemq_celery/__init__.py

from kombu.transport import TRANSPORT_ALIASES

TRANSPORT_ALIASES["kubemq"] = "kubemq_celery.transport:Transport"
TRANSPORT_ALIASES["kubemq+tls"] = "kubemq_celery.transport:Transport"

from celery.app.backends import BACKEND_ALIASES

BACKEND_ALIASES["kubemq"] = "kubemq_celery.backend:KubeMQResultBackend"

If you cannot guarantee import order, set the result backend by its fully-qualified path instead of the alias: result_backend = "kubemq_celery.backend:KubeMQResultBackend".

Broker URL schemes

The transport accepts four URL schemes. They differ in two dimensions: TLS on or off, and the synchronous gRPC client versus the native asyncio client.

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

Scheme	TLS	Client	Use with
`kubemq://`	No	Synchronous	Default worker pools (prefork, threads)
`kubemq+tls://`	Yes	Synchronous	Encrypted gRPC connections
`kubemq+async://`	No	Async (`AsyncQueuesClient`, `AsyncPubSubClient`)	`--pool=asyncio` workers
`kubemq+async+tls://`	Yes	Async	Async pools over TLS

The +async schemes use native async KubeMQ clients for non-blocking I/O and are intended for Celery's asyncio worker pool (celery -A myapp worker --pool=asyncio).

URL components

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

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

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

# With an authentication token
app.conf.broker_url = "kubemq://:my-secret-token@kubemq.default.svc:50000"

# TLS, with and without a token
app.conf.broker_url = "kubemq+tls://kubemq.default.svc:50000"
app.conf.broker_url = "kubemq+tls://:my-token@kubemq.default.svc:50000"

# Async transport, with and without TLS
app.conf.broker_url = "kubemq+async://localhost:50000"
app.conf.broker_url = "kubemq+async+tls://kubemq.default.svc:50000"

A broker is reachable on the gRPC port 50000. The quickest way to get one locally is Docker — the shared HTTP server (REST/health) is exposed 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

Celery settings compatibility

The following table records how kubemq-celery treats common Celery settings. Full means the setting behaves as it does on any other broker; Ignored means the transport disregards it; Stored means the value is preserved on the message but not enforced server-side.

Celery setting	Support	Notes
`broker_url`	Full	`kubemq://`, `kubemq+tls://`, `kubemq+async://`, `kubemq+async+tls://` schemes.
`broker_transport_options`	Full	All keys in Transport Options.
`broker_pool_limit`	Ignored	One client per Channel; Kombu's connection pool handles scaling.
`broker_connection_timeout`	Full	Passed to the SDK connection timeout.
`broker_connection_retry`	Full	Standard Celery reconnection behavior.
`broker_connection_max_retries`	Full	Standard Celery reconnection behavior.
`broker_failover_strategy`	Ignored	Not supported in this release; the KubeMQ cluster handles HA internally.
`task_acks_late`	Full	See the caveat below for long-running tasks.
`task_acks_on_failure_or_timeout`	Full	Standard Celery behavior.
`task_reject_on_worker_lost`	Full	Maps to KubeMQ `nack()`.
`task_default_queue`	Full	Queue name is sanitized for KubeMQ compatibility.
`task_routes`	Full	The virtual exchange layer handles routing.
`task_default_priority`	Stored	Priority is stored in message tags, not enforced server-side.
`task_queue_max_priority`	Stored	Priority is stored in message tags.
`worker_prefetch_multiplier`	Full	Managed by Kombu's virtual QoS layer.
`worker_concurrency`	Full	Standard Celery behavior.
`result_backend`	Full	`kubemq://` for the queue-peek result backend.
`result_expires`	Full	Controls result message expiration (max 24 hours).

task_acks_late caveat. With task_acks_late=True, a message is acknowledged only after the task completes. For tasks shorter than 60 seconds this works as expected. For tasks longer than 60 seconds, the KubeMQ server-side transaction timeout may expire before the task finishes, causing redelivery. For long-running tasks, either keep task_acks_late=False (the default, which uses auto_ack=True on receive) or accept at-least-once semantics and make the task idempotent.

Environment variables

The example applications read connection settings from the environment. These are conventions used by the example code — your own app can wire them up however you prefer.

Variable	Default	Description
`CELERY_BROKER_URL`	`kubemq://localhost:50000`	KubeMQ broker URL.
`CELERY_RESULT_BACKEND`	`kubemq://localhost:50000`	Result backend URL.
`CELERY_ENV`	`development`	Environment name; affects logging and config in some examples.

import os

import kubemq_celery  # noqa: F401
from celery import Celery

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

Monitoring and control

All of Celery's monitoring and control commands work over KubeMQ Events (pidbox fanout), so the standard tooling needs no changes beyond the broker URL.

celery inspect

# Check whether workers are alive (uses the transport's verify_connection ping)
celery -A myapp inspect ping

# List running tasks
celery -A myapp inspect active

# Worker pool statistics (uptime, task counts, prefetch)
celery -A myapp inspect stats

# Queues each worker is currently consuming
celery -A myapp inspect active_queues

celery control

Runtime control commands are broadcast to all online workers via Events fanout. They are fire-and-forget and are not persisted, so workers must be online to receive them.

# Add or remove a consumer at runtime
celery -A myapp control add_consumer new-queue
celery -A myapp control cancel_consumer old-queue

# Grow or shrink the worker pool
celery -A myapp control pool_grow 2
celery -A myapp control pool_shrink 1

# Set a runtime rate limit
celery -A myapp control rate_limit myapp.tasks.add 10/m

The same commands are available programmatically through app.control:

app.control.ping(timeout=5)
app.control.inspect().active()
app.control.add_consumer("new-queue")
app.control.shutdown()

Flower

Flower works against a kubemq:// broker like any other:

celery -A myapp flower --broker=kubemq://localhost:50000

KubeMQ dashboard

The KubeMQ Management API dashboard on port 8080 shows broker-level metrics — queue depth, send/receive rates, and per-channel statistics. (The shared HTTP server on 9090 serves REST and the /health probe.) Inside a cluster, port-forward to reach the dashboard:

kubectl port-forward svc/kubemq 8080:8080
# then open http://localhost:8080

Known limitations

Limitation	Detail
Priority is metadata-only	KubeMQ does not enforce message priority at the server level; `task_default_priority` and `task_queue_max_priority` are stored in message tags but not honored for ordering.
Maximum delay	`countdown` / `eta` delays are capped at `86400` seconds (24 hours); larger values are clamped with a warning log.
Maximum result expiry	`result_expires` and `message_expiration` are capped at `86400` seconds (24 hours), a KubeMQ limitation.
Chord polling fallback	Chords use Celery's `chord_unlock` task to poll for group completion rather than a native server-side primitive.
Long `task_acks_late` tasks	Tasks running longer than the KubeMQ transaction timeout (~60s) under `task_acks_late=True` may be redelivered; design for idempotency or keep acks early.
`broker_failover_strategy`	Not supported; the KubeMQ cluster handles HA internally.

Exceptions

The transport defines an exception hierarchy in kubemq_celery.exceptions, all rooted at the KubeMQ SDK's KubeMQError. The base for this package is KubeMQCeleryError, with transport, backend, serialization, and config subclasses.

from kubemq_celery.exceptions import (
    KubeMQCeleryError,            # base for all kubemq-celery errors
    KubeMQCeleryTransportError,   # send / receive / subscribe failures
    KubeMQCeleryConnectionError,  # broker connection failed or lost
    KubeMQCeleryChannelError,     # queue not found, permission denied
    KubeMQCeleryTimeoutError,     # operation timed out
    KubeMQCeleryBackendError,     # result backend store / retrieve errors
    KubeMQCelerySerializationError,  # JSON (de)serialization errors
    KubeMQCeleryConfigError,      # invalid transport configuration
)

You will also see these surfaced from the underlying KubeMQ SDK during connection problems:

Exception	When it appears
`KubeMQCeleryConnectionError`	The worker cannot reach the broker on startup (for example, "Connection refused").
`KubeMQAuthenticationError`	The auth token is missing or does not match the broker's configuration.
`KubeMQStreamBrokenError` / `KubeMQConnectionError`	Intermittent loss of a live connection — network instability, a broker restart, or a load-balancer timeout. Configure gRPC keepalive (`grpc_keepalive_time`, `grpc_keepalive_timeout`, `grpc_permit_without_calls`) to detect and recover.

For diagnosis steps see the Troubleshooting guide.

Optional dependency extras

The example apps in the repository require packages beyond the three runtime dependencies. They are declared under the examples extra in pyproject.toml and can be installed individually as needed.

# Framework integrations
pip install fastapi uvicorn   # FastAPI examples
pip install flask             # Flask examples
pip install starlette         # Starlette examples
pip install litestar          # Litestar examples
pip install aiohttp           # aiohttp examples
pip install django            # Django examples

# Serialization
pip install msgpack           # MessagePack serializer
pip install orjson            # Custom orjson serializer

Package	Minimum version	Used by
`fastapi`	`>= 0.100`	FastAPI integration example
`uvicorn`	`>= 0.20`	ASGI server for FastAPI / Starlette / Litestar
`flask`	`>= 3.0`	Flask integration example
`starlette`	`>= 0.30`	Starlette integration example
`litestar`	`>= 2.0`	Litestar integration example
`aiohttp`	`>= 3.9`	aiohttp integration example
`django`	`>= 4.2`	Django integration example
`django-celery-beat`	`>= 2.5`	Django periodic-task scheduling
`django-celery-results`	`>= 2.5`	Django result storage
`msgpack`	`>= 1.0`	MessagePack serializer example

orjson is referenced by the custom-serializer example but is installed separately (pip install orjson); it is not part of the declared examples extra. Install only the packages an example actually needs rather than the whole extra.

Configuration Reference