# Configuration Reference (/integrations/faststream/reference/configuration)



This page is the configuration reference for `kubemq-faststream`: package facts, the
`KubeMQBroker` constructor surface, URL formats, validation rules, the
environment-variable overrides, and the FastStream standard broker options. For the
exported symbols, enums, subscriber fields, and the `broker.request()` signature, see the
[API reference](/integrations/faststream/reference/api). For runnable walkthroughs,
start with [Getting Started](/integrations/faststream/tutorials/getting-started).

## Package Facts [#package-facts]

| Fact                  | Value                                  |
| --------------------- | -------------------------------------- |
| Package name          | `kubemq-faststream`                    |
| Version               | `0.1.0` (Development Status: 4 - Beta) |
| License               | MIT                                    |
| Python                | 3.11+ (tested on 3.11, 3.12, 3.13)     |
| FastStream dependency | `faststream>=0.6.7,<0.7.0`             |
| KubeMQ SDK dependency | `kubemq>=4.1.5,<5`                     |

Install with `uv` or `pip`:

<Tabs groupId="py-installer" items="['uv', 'pip']">
  <Tab value="uv">
    ```bash
    uv add kubemq-faststream
    ```
  </Tab>

  <Tab value="pip">
    ```bash
    pip install kubemq-faststream
    ```
  </Tab>
</Tabs>

<Callout type="info">
  `kubemq-faststream` is a native gRPC SDK client that talks to KubeMQ over the gRPC port
  `50000`. It is **always on** — there is no server-side connector enable flag to set. Start
  a broker with `docker run -d --rm -p 50000:50000 -p 9090:9090 -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY europe-docker.pkg.dev/kubemq/images/kubemq:next`.
  Port `50000` is the gRPC transport the adapter uses; port `9090` is the shared HTTP server
  and is not required for FastStream.
</Callout>

## KubeMQBroker Constructor [#kubemqbroker-constructor]

`KubeMQBroker` extends FastStream's `BrokerUsecase`, so it accepts the standard FastStream
broker options alongside the KubeMQ-specific connection settings. The first positional
argument is the broker URL; everything else is keyword-only.

```python
from kubemq_faststream import KubeMQBroker

broker = KubeMQBroker(
    "kubemq://localhost:50000",
    client_id="my-service",
    auth_token="my-token",
    default_cq_timeout=30,
    graceful_timeout=15.0,
)
```

### KubeMQ Connection Parameters [#kubemq-connection-parameters]

<TypeTable
  type="{
  url: { type: 'str', default: '&#x22;kubemq://localhost:50000&#x22;', description: 'Broker URL. Accepts kubemq://, kubemq+tls://, or a bare host:port.' },
  client_id: { type: 'str | None', default: 'None', description: 'Client identifier. Falls back to the system hostname; non-alphanumeric characters are replaced with &#x22;-&#x22;.' },
  auth_token: { type: 'str | None', default: 'None', description: 'JWT authentication token. An empty (whitespace-only) value is rejected.' },
  tls_enabled: { type: 'bool', default: 'false', description: 'Enable TLS. Also enabled implicitly by the kubemq+tls:// URL scheme.' },
  tls_cert_file: { type: 'str | None', default: 'None', description: 'Client certificate for mTLS. Requires tls_key_file.' },
  tls_key_file: { type: 'str | None', default: 'None', description: 'Client key for mTLS. Requires tls_cert_file.' },
  tls_ca_file: { type: 'str | None', default: 'None', description: 'CA certificate used to verify the server.' },
  max_send_size: { type: 'int', default: '4194304', description: 'Maximum outbound message size in bytes (4 MB). Must be > 0.' },
  max_receive_size: { type: 'int', default: '4194304', description: 'Maximum inbound message size in bytes (4 MB). Must be > 0.' },
  default_cq_timeout: { type: 'int', default: '30', description: 'Default command/query timeout in seconds. Must be > 0.' },
  keepalive_time_ms: { type: 'int', default: '30000', description: 'gRPC keepalive ping interval in milliseconds.' },
  keepalive_timeout_ms: { type: 'int', default: '10000', description: 'gRPC keepalive ping timeout in milliseconds.' },
  graceful_timeout: { type: 'float | None', default: '15.0', description: 'Seconds to wait for in-flight handlers to finish on shutdown.' },
}"
/>

<Callout type="warn">
  mTLS requires both `tls_cert_file` and `tls_key_file`. Supplying one without the other raises `ValueError` at construction time. Likewise, `max_send_size`, `max_receive_size`, and `default_cq_timeout` must all be greater than zero.
</Callout>

For TLS, mTLS, and authentication setups with full examples, see
[Configuration & Security](/integrations/faststream/how-to/configuration).

### FastStream Standard Parameters [#faststream-standard-parameters]

These keyword-only options mirror the other FastStream brokers (the `KafkaBroker`
pattern). They are optional and default to FastStream's standard values.

| Parameter           | Type                           | Purpose                                                          |
| ------------------- | ------------------------------ | ---------------------------------------------------------------- |
| `decoder`           | `CustomCallable \| None`       | Custom message decoder applied broker-wide                       |
| `parser`            | `CustomCallable \| None`       | Custom message parser applied broker-wide                        |
| `dependencies`      | `Iterable[Dependant]`          | FastDepends dependencies injected into every handler             |
| `middlewares`       | `Sequence[BrokerMiddleware]`   | Broker-level middleware chain                                    |
| `routers`           | `Iterable[KubeMQRegistrator]`  | Routers to include at construction time                          |
| `security`          | `BaseSecurity \| None`         | AsyncAPI security scheme                                         |
| `specification_url` | `str \| Iterable[str] \| None` | AsyncAPI server URL(s); defaults to the broker URL               |
| `protocol`          | `str \| None`                  | AsyncAPI protocol label; defaults to `kubemq` or `kubemq+tls`    |
| `protocol_version`  | `str \| None`                  | AsyncAPI protocol version; defaults to `"1.0"`                   |
| `description`       | `str \| None`                  | AsyncAPI broker description                                      |
| `tags`              | `Iterable[Tag \| TagDict]`     | AsyncAPI tags                                                    |
| `logger`            | `LoggerProto \| None`          | Custom logger; FastStream default when unset                     |
| `log_level`         | `int`                          | Logging level (default `logging.INFO`)                           |
| `apply_types`       | `bool`                         | Enable FastDepends type casting (default `True`)                 |
| `serializer`        | `SerializerProto \| None`      | Custom FastDepends serializer                                    |
| `provider`          | `Provider \| None`             | FastDepends dependency provider                                  |
| `context`           | `ContextRepo \| None`          | FastStream context repository                                    |
| `include_in_schema` | `bool`                         | Include broker in the generated AsyncAPI schema (default `True`) |
| `prefix`            | `str`                          | Channel prefix prepended to every handler on this broker         |

## URL Formats [#url-formats]

The `url` argument (or the `KUBEMQ_ADDRESS` environment variable) accepts three forms:

| Format                   | Description                                   |
| ------------------------ | --------------------------------------------- |
| `kubemq://host:port`     | Plain gRPC connection                         |
| `kubemq+tls://host:port` | gRPC with TLS (sets `tls_enabled` implicitly) |
| `host:port`              | Plain gRPC, bare form (no scheme)             |

```python
KubeMQBroker("kubemq://localhost:50000")           # plain
KubeMQBroker("kubemq+tls://broker.example:50000")  # TLS via scheme
KubeMQBroker("localhost:50000")                    # bare host:port
```

## Environment Variables [#environment-variables]

Every connection parameter can be supplied through an environment variable. **Environment
variables take precedence over constructor arguments when set**, so you can ship code with
sensible defaults and override the target broker at deploy time without changing source.

| 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`                       |

A broker constructed with no arguments reads its entire configuration from the
environment:

```python title="env_var_config.py"
import os
from kubemq_faststream import KubeMQBroker

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

broker = KubeMQBroker()  # picks up KUBEMQ_ADDRESS and KUBEMQ_CLIENT_ID
```

## 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`.
* An `auth_token` that is set but blank (empty or whitespace-only) is rejected. To run
  without authentication, leave the token unset rather than passing an empty string.

## See also [#see-also]

<Cards>
  <Card title="API Reference" href="/integrations/faststream/reference/api" description="Exported symbols, the KubeMQPattern / AckPolicy / StartPosition enums, subscriber configuration, and the broker.request() signature." />

  <Card title="Configuration & Security" href="/integrations/faststream/how-to/configuration" description="TLS, mTLS, auth tokens, message-size tuning, and timeouts with full examples." />
</Cards>