# Configuration (/integrations/celery/how-to/configuration) 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 [#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](#full-configuration-example) below) * `import kubemq_celery` present before Celery resolves the broker URL, so the `kubemq*://` schemes are registered with Kombu ## Broker URL [#broker-url] ### Format [#format] The broker URL selects the transport scheme and points at your KubeMQ broker. All four schemes share the same `[:token@]host[:port]` shape: ```text 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](#transport-options)). ### Examples [#examples] ```python title="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 [#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: ```bash title="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 [#transport-options] All connection tuning beyond the URL goes through `broker_transport_options`: ```python title="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-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: ```python title="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 [#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. ```python title="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. ```python title="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. ```python title="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 [#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`: ```python title="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: ```bash title="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 [#full-configuration-example] A complete, production-shaped configuration that combines the broker, the result backend, and task and worker settings: ```python title="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`: ## Related [#related]