# Configuration & Security (/integrations/faststream/how-to/configuration)



`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 [#prerequisites]

* `kubemq-faststream` installed (`pip install kubemq-faststream`) alongside `faststream`
* A running KubeMQ broker reachable from the app (see [Starting a Broker](#starting-a-broker) below)

## URL Formats [#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.

```python title="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")
```

<Callout type="info">
  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.
</Callout>

## Constructor Options [#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.

```python title="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.                                |

<Callout type="info">
  `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`.
</Callout>

## Validation Rules [#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.

```python title="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 [#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:

```python title="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)
```

```bash title="run with overrides"
KUBEMQ_ADDRESS=kubemq://my-broker:50000 \
KUBEMQ_CLIENT_ID=order-service \
python app.py
```

## TLS with a Server Certificate [#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.

```python title="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:

```bash title="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]

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.

```python title="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)
```

<Callout type="warn">
  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.
</Callout>

## Authentication Token [#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.

<Tabs groupId="auth-source" items="['Constructor', 'Environment variable']">
  <Tab value="Constructor">
    ```python title="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)
    ```
  </Tab>

  <Tab value="Environment variable">
    ```python title="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)
    ```

    ```bash title="set the token"
    export KUBEMQ_AUTH_TOKEN="my-jwt-token-here"
    python app.py
    ```
  </Tab>
</Tabs>

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.

<Callout type="error">
  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.
</Callout>

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

```python title="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 [#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.

```python title="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 [#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`.

```python title="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 [#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.

```python title="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 [#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:

<RunKubeMQ ports="[50000, 9090]" />

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.

## Related [#related]

* [Getting Started with FastStream](/integrations/faststream/tutorials/getting-started) — install the package and run your first app.
* [Queues](/integrations/faststream/how-to/queues) — point-to-point messaging with ack policies and batch publishing.
* [FastStream BaseSecurity](https://faststream.ag2.ai/latest/kafka/security/) — the security API shared across FastStream brokers.