KubeMQBroker connects to KubeMQ over native gRPC. Connection settings are supplied three ways, in increasing order of precedence: constructor arguments, the broker URL scheme, and environment variables. A single set of settings is shared across the three internal SDK clients the broker creates on connect.

This page covers every connection option: URL formats, the full constructor signature, validation rules, the environment-variable overrides, and the TLS, mTLS, and authentication setups.

Prerequisites

kubemq-faststream installed (pip install kubemq-faststream) alongside faststream
A running KubeMQ broker reachable from the app (see Starting a Broker below)

URL Formats

The first positional argument to KubeMQBroker is the broker URL. Three formats are accepted; the scheme decides whether TLS is enabled on the gRPC channel.

Format	Description
`kubemq://host:port`	Plain gRPC connection
`kubemq+tls://host:port`	gRPC with TLS
`host:port`	Plain gRPC (bare format)

The default URL is kubemq://localhost:50000 — the native gRPC port. Use kubemq+tls:// to turn on transport encryption without setting tls_enabled explicitly.

broker_url_formats.py

from kubemq_faststream import KubeMQBroker

# Plain gRPC (these are equivalent — the bare host:port form is also accepted)
broker = KubeMQBroker("kubemq://localhost:50000")
broker = KubeMQBroker("kubemq://127.0.0.1:50000")

# gRPC with TLS — the kubemq+tls:// scheme enables TLS on the channel
broker = KubeMQBroker("kubemq+tls://my-broker.example.com:50000")

The kubemq+tls:// scheme enables TLS regardless of the tls_enabled argument. The broker treats TLS as on if the scheme requests it or tls_enabled=True is passed.

Constructor Options

Every connection setting is a keyword argument on the constructor. Defaults shown below match the source; the only positional argument is the URL.

all_options.py

from kubemq_faststream import KubeMQBroker

broker = KubeMQBroker(
    # --- Connection ---
    "kubemq://localhost:50000",  # Broker URL (kubemq:// or kubemq+tls://)
    client_id="all-options-demo",  # Client identifier (default: hostname)
    auth_token=None,  # JWT auth token (default: None)
    # --- TLS ---
    tls_enabled=False,  # Enable TLS (default: False)
    tls_cert_file=None,  # Client cert for mTLS (default: None)
    tls_key_file=None,  # Client key for mTLS (default: None)
    tls_ca_file=None,  # CA cert to verify server (default: None)
    # --- Message limits ---
    max_send_size=4_194_304,  # Max outgoing message size in bytes (4 MB)
    max_receive_size=4_194_304,  # Max incoming message size in bytes (4 MB)
    # --- Timeouts ---
    default_cq_timeout=30,  # Default command/query timeout in seconds
    graceful_timeout=15.0,  # Seconds to wait for handlers on shutdown
    # --- Keepalive ---
    keepalive_time_ms=30_000,  # gRPC keepalive ping interval (ms)
    keepalive_timeout_ms=10_000,  # gRPC keepalive ping timeout (ms)
)

What each option controls:

Option	Default	Purpose
`client_id`	system hostname	Client identifier reported to the broker. Non-alphanumeric characters are normalized to `-`.
`auth_token`	`None`	JWT token attached to every gRPC call. An empty string is rejected.
`tls_enabled`	`False`	Enables TLS on the channel. Implied by a `kubemq+tls://` URL.
`tls_cert_file`	`None`	Path to the client certificate (mTLS). Requires `tls_key_file`.
`tls_key_file`	`None`	Path to the client private key (mTLS). Requires `tls_cert_file`.
`tls_ca_file`	`None`	Path to the CA bundle used to verify the server certificate.
`max_send_size`	`4194304` (4 MB)	Maximum outbound message size in bytes.
`max_receive_size`	`4194304` (4 MB)	Maximum inbound message size in bytes.
`default_cq_timeout`	`30`	Default deadline (seconds) for `broker.request()` commands and queries.
`keepalive_time_ms`	`30000`	gRPC keepalive ping interval in milliseconds.
`keepalive_timeout_ms`	`10000`	gRPC keepalive ping timeout in milliseconds.
`graceful_timeout`	`15.0`	Seconds to wait for in-flight handlers to finish on shutdown.

client_id defaults to the system hostname when omitted. The broker sanitizes it by replacing any character outside [a-zA-Z0-9_-] with -, so a hostname like host.local becomes host-local.

Validation Rules

The broker config validates these settings as it is constructed. Violations raise ValueError immediately — before any connection is attempted — so misconfiguration fails fast.

max_send_size must be greater than 0.
max_receive_size must be greater than 0.
default_cq_timeout must be greater than 0.
For mutual TLS, tls_cert_file and tls_key_file are paired: setting one without the other raises ValueError. A client certificate and its key are useless apart, so the config refuses a half-configured pair.

validation.py

from kubemq_faststream import KubeMQBroker

# Raises ValueError: max_send_size must be > 0
KubeMQBroker("kubemq://localhost:50000", max_send_size=0)

# Raises ValueError: tls_cert_file requires tls_key_file for mTLS
KubeMQBroker("kubemq://localhost:50000", tls_cert_file="/path/client.pem")

In addition, an auth_token that is set but blank (an empty or whitespace-only string) is rejected with a ValueError. To run without authentication, leave the token unset rather than passing an empty string.

Environment Variables

Every connection setting can also be supplied through an environment variable. Environment variables take precedence over constructor arguments when set — the broker resolves the effective configuration by overlaying the environment on top of the values you passed in code.

Environment Variable	Constructor Parameter	Default
`KUBEMQ_ADDRESS`	`url`	`kubemq://localhost:50000`
`KUBEMQ_CLIENT_ID`	`client_id`	System hostname
`KUBEMQ_AUTH_TOKEN`	`auth_token`	None (no auth)
`KUBEMQ_TLS_ENABLED`	`tls_enabled`	`false`
`KUBEMQ_TLS_CERT_FILE`	`tls_cert_file`	None
`KUBEMQ_TLS_KEY_FILE`	`tls_key_file`	None
`KUBEMQ_TLS_CA_FILE`	`tls_ca_file`	None
`KUBEMQ_MAX_SEND_SIZE`	`max_send_size`	`4194304`
`KUBEMQ_MAX_RECEIVE_SIZE`	`max_receive_size`	`4194304`
`KUBEMQ_DEFAULT_CQ_TIMEOUT`	`default_cq_timeout`	`30`

This lets you ship code with sensible defaults and point it at a different broker — or layer in TLS and auth — entirely from the deployment environment, with no source changes. Construct the broker with no arguments and let the environment fill everything in:

env_var_config.py

import os
from faststream import FastStream
from kubemq_faststream import KubeMQBroker

os.environ.setdefault("KUBEMQ_ADDRESS", "localhost:50000")
os.environ.setdefault("KUBEMQ_CLIENT_ID", "env-var-demo")

# No constructor args — KUBEMQ_* env vars are resolved automatically.
broker = KubeMQBroker()
app = FastStream(broker)

run with overrides

KUBEMQ_ADDRESS=kubemq://my-broker:50000 \
KUBEMQ_CLIENT_ID=order-service \
python app.py

TLS with a Server Certificate

To encrypt the transport and verify the broker's identity, use the kubemq+tls:// scheme and point tls_ca_file at the CA bundle that signed the server certificate. The scheme turns on TLS for the channel; the CA file lets the client validate the server.

tls_server_cert.py

import os
from faststream import FastStream
from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq+tls://localhost:50000")

# CA certificate path — override via KUBEMQ_TLS_CA_FILE env var.
TLS_CA_FILE = os.environ.get("KUBEMQ_TLS_CA_FILE", "/path/to/ca.pem")

broker = KubeMQBroker(
    KUBEMQ_ADDRESS,
    tls_ca_file=TLS_CA_FILE,
)
app = FastStream(broker)

The same setup can be configured entirely from the environment, keeping certificate paths out of code:

TLS via env vars

export KUBEMQ_ADDRESS="kubemq+tls://your-broker:50000"
export KUBEMQ_TLS_CA_FILE="/path/to/ca.pem"
python app.py

Mutual TLS

Mutual TLS adds client authentication on top of server verification: both sides present certificates. Supply the client certificate, the client private key, and the CA bundle. Set tls_enabled=True (or use the kubemq+tls:// scheme) to turn TLS on.

mtls_mutual.py

import os
from faststream import FastStream
from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")

TLS_CERT_FILE = os.environ.get("KUBEMQ_TLS_CERT_FILE", "/path/to/client.pem")
TLS_KEY_FILE = os.environ.get("KUBEMQ_TLS_KEY_FILE", "/path/to/client-key.pem")
TLS_CA_FILE = os.environ.get("KUBEMQ_TLS_CA_FILE", "/path/to/ca.pem")

broker = KubeMQBroker(
    KUBEMQ_ADDRESS,
    tls_enabled=True,  # explicitly enable TLS
    tls_cert_file=TLS_CERT_FILE,  # client certificate
    tls_key_file=TLS_KEY_FILE,  # client private key
    tls_ca_file=TLS_CA_FILE,  # CA cert to verify server
)
app = FastStream(broker)

For mTLS, tls_cert_file and tls_key_file must both be set. Passing one without the other raises ValueError at construction time. The same three paths can be supplied via KUBEMQ_TLS_CERT_FILE, KUBEMQ_TLS_KEY_FILE, and KUBEMQ_TLS_CA_FILE, which take precedence over the constructor arguments.

Authentication Token

When the broker enforces authentication, attach a JWT via the auth_token argument. The token is included in the metadata of every gRPC call; if the broker rejects it, the connection fails with an authentication error.

auth_token_constructor.py

import os
from faststream import FastStream
from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")

# Replace with a real JWT token in production.
AUTH_TOKEN = "my-jwt-token-here"

broker = KubeMQBroker(
    KUBEMQ_ADDRESS,
    auth_token=AUTH_TOKEN,
)
app = FastStream(broker)

auth_token_env_var.py

import os
from faststream import FastStream
from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")

# No auth_token= constructor arg — KUBEMQ_AUTH_TOKEN is resolved automatically.
broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)

set the token

export KUBEMQ_AUTH_TOKEN="my-jwt-token-here"
python app.py

Supplying the token through KUBEMQ_AUTH_TOKEN keeps secrets out of source code and works well with container orchestrators that inject env vars from a secrets store.

An auth token that is set but empty ("" or whitespace) is rejected with ValueError. To connect without authentication, leave the token unset entirely rather than passing a blank string.

For production, combine TLS encryption with token authentication — encrypted transport plus identity verification on every call:

tls_with_auth.py

import os
from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq+tls://localhost:50000")
AUTH_TOKEN = os.environ.get("KUBEMQ_AUTH_TOKEN", "my-jwt-token-here")
TLS_CA_FILE = os.environ.get("KUBEMQ_TLS_CA_FILE", "/path/to/ca.pem")

broker = KubeMQBroker(
    KUBEMQ_ADDRESS,
    auth_token=AUTH_TOKEN,
    tls_ca_file=TLS_CA_FILE,
)

FastStream BaseSecurity Integration

The constructor also accepts a security argument of any FastStream BaseSecurity instance — the same security API used by the Kafka and RabbitMQ brokers. Pair it with a standard-library ssl.SSLContext when you need fine-grained control over cipher suites, hostname checking, or protocol versions beyond what the tls_* parameters expose.

faststream_security.py

import os
import ssl
from faststream import FastStream
from faststream.security import BaseSecurity
from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq+tls://localhost:50000")

CA_FILE = os.environ.get("KUBEMQ_TLS_CA_FILE", "/path/to/ca.pem")
CERT_FILE = os.environ.get("KUBEMQ_TLS_CERT_FILE", "/path/to/client.pem")
KEY_FILE = os.environ.get("KUBEMQ_TLS_KEY_FILE", "/path/to/client-key.pem")

# Build an SSLContext with fine-grained settings.
ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_ctx.minimum_version = ssl.TLSVersion.TLSv1_2

# Load CA cert for server verification.
ssl_ctx.load_verify_locations(CA_FILE)

# Load client cert + key for mTLS (optional).
ssl_ctx.load_cert_chain(certfile=CERT_FILE, keyfile=KEY_FILE)

# Wrap in FastStream's BaseSecurity so the broker can consume it.
security = BaseSecurity(ssl_context=ssl_ctx)

broker = KubeMQBroker(
    KUBEMQ_ADDRESS,
    security=security,
)
app = FastStream(broker)

Tuning Message Size

The default message-size limit is 4 MB (4_194_304 bytes) in each direction. Raise it for large payloads or lower it to cap memory usage. Both max_send_size and max_receive_size must be greater than 0.

message_size.py

from kubemq_faststream import KubeMQBroker

MAX_SIZE = 1_048_576  # 1 MB

broker = KubeMQBroker(
    "kubemq://localhost:50000",
    max_send_size=MAX_SIZE,
    max_receive_size=MAX_SIZE,
)

Tuning Timeouts

default_cq_timeout sets the default deadline (in seconds) for broker.request() commands and queries when no explicit timeout= is passed. graceful_timeout controls how long the app waits for in-flight handlers to finish when stopping.

timeouts.py

from kubemq_faststream import KubeMQBroker

broker = KubeMQBroker(
    "kubemq://localhost:50000",
    default_cq_timeout=10,  # 10-second default for commands/queries
    graceful_timeout=5.0,  # 5-second shutdown window
)

A per-request timeout= argument on broker.request(...) overrides default_cq_timeout for that single call.

Starting a Broker

The examples above expect a running KubeMQ broker. Start one locally with Docker — port 50000 is the native gRPC port kubemq-faststream connects to:

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

Port 9090 is the shared HTTP server (REST and connector endpoints) and is not required for FastStream, which connects directly over gRPC. No server-side connector enable flag is needed.

Getting Started with FastStream — install the package and run your first app.
Queues — point-to-point messaging with ack policies and batch publishing.
FastStream BaseSecurity — the security API shared across FastStream brokers.

Configuration & Security