# Connection & Security (/integrations/rayserve/how-to/connection-security)



Every Ray Serve deployment that uses the KubeMQ adapter is configured through a single `KubeMQAdapterConfig` object. It controls where the adapter connects, how it identifies itself, and how the gRPC channel is secured. This guide walks through each connection and security option — from the zero-config default to mutual TLS — using the runnable examples that ship in the repository's `examples/connection/` directory.

All examples assume a broker reachable at `localhost:50000`. Start one with Docker:

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

Port `50000` is the gRPC endpoint the adapter's SDK clients connect to. Port `9090` exposes the shared HTTP server used by connector endpoints such as KEDA-based queue scaling.

## Prerequisites [#prerequisites]

* `kubemq-rayserve` installed (see [Getting Started with Ray Serve](/integrations/rayserve/tutorials/getting-started))
* The broker above running and reachable
* A JWT token or TLS certificates on hand if your broker enforces authentication or TLS

## Default Connection [#default-connection]

A bare `KubeMQAdapterConfig()` targets `localhost:50000` — the `address` field defaults to that value, so no arguments are required for local development. The example below builds the default config, initializes the adapter, and verifies the broker is reachable with a `health_check()` call.

```python title="examples/connection/basic_connection.py"
import os
import uuid

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

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


def main():
    channel = f"example-connection-{uuid.uuid4().hex[:8]}"

    # Default config connects to localhost:50000
    config = KubeMQAdapterConfig(address=BROKER)
    print(f"Config address: {config.address}")
    print(f"Config client_id: {config.client_id!r} (auto-generated if empty)")

    adapter = KubeMQTaskProcessorAdapter(config)

    class _Cfg:
        queue_name = channel
        max_retries = 0
        failed_task_queue_name = ""
        unprocessable_task_queue_name = ""

    adapter.initialize(consumer_concurrency=1, task_processor_config=_Cfg())

    try:
        # Verify connectivity
        healthy = adapter.health_check()
        print(f"Health check: {healthy}")
    finally:
        adapter.stop_consumer()
```

The small `_Cfg` shim stands in for the `TaskProcessorConfig` that Ray Serve supplies in a real deployment. For standalone scripts like this one, the adapter only needs the four queue fields — `queue_name`, `max_retries`, `failed_task_queue_name`, and `unprocessable_task_queue_name` — to initialize.

## Custom Broker Address [#custom-broker-address]

Set `address='host:port'` to point the adapter at a remote or in-cluster broker. The most common in-cluster value is `kubemq:50000`, the Kubernetes service name resolved inside the same namespace.

```python title="examples/connection/custom_address.py"
import os

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

# Override the default with a custom broker address
config = KubeMQAdapterConfig(address="kubemq:50000")
print(f"Connecting to custom address: {config.address}")

adapter = KubeMQTaskProcessorAdapter(config)
```

<Callout type="info">
  Every bundled example honors a `KUBEMQ_ADDRESS` environment variable as a fallback, so you can run them against any broker without editing code: `KUBEMQ_ADDRESS=my-broker:50000 python examples/connection/custom_address.py`.
</Callout>

## Client Identification [#client-identification]

Set `client_id` to give the connection a stable, recognizable name. The client ID appears in the broker logs and the web UI at port `9090`, which makes it easy to tell workers apart when you run several replicas.

```python title="examples/connection/custom_client_id.py"
config = KubeMQAdapterConfig(
    address="localhost:50000",
    client_id="my-worker-01",
)
print(f"Config client_id: {config.client_id}")
```

When `client_id` is left empty (the default), the adapter generates a unique one in the form `rayserve-{8hex}` — a `rayserve-` prefix followed by the first eight characters of a UUID. This guarantees a distinct identity per connection even if you do not set one explicitly, but a fixed name is preferable for monitoring.

## JWT Authentication [#jwt-authentication]

For brokers that require authentication, set `auth_token` to a JWT issued by your KubeMQ license. The adapter maps this value to the SDK's `ClientConfig.auth_token`, normalizing an empty string to `None` so an unset token is never sent as a blank credential.

```python title="examples/connection/auth_token.py"
import os

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

auth_token = os.environ.get("KUBEMQ_AUTH_TOKEN", "<your-jwt-token>")

config = KubeMQAdapterConfig(
    address="localhost:50000",
    auth_token=auth_token,
)

# auth_token is hidden in repr for security
print(f"Config: {config}")
print(f"Auth token set: {bool(config.auth_token)}")
```

The `auth_token` field is declared with `repr=False`, so it is omitted when the config object is printed or logged. Printing `config` shows every other field but never the token itself — important when configs end up in log lines or crash reports. Use the `bool(config.auth_token)` check shown above when you need to confirm a token is present without revealing it.

<Callout type="warn">
  Never hard-code a JWT in source. Load it from an environment variable (as above) or, in Kubernetes, from a Secret. The production wiring is covered in the [Kubernetes production deployment](/integrations/rayserve/scenarios/kubernetes-production-deployment) scenario.
</Callout>

## TLS [#tls]

To encrypt the gRPC channel, set `tls=True` and point `tls_ca_file` at the CA certificate that signed the broker's certificate. The adapter assembles these fields into the SDK's `TLSConfig`.

```python title="examples/connection/tls_setup.py"
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

# TLS with the CA certificate that signed the broker's cert
config = KubeMQAdapterConfig(
    address="localhost:50000",
    tls=True,
    tls_ca_file="ca.pem",
)

print(f"tls={config.tls}")
print(f"tls_ca_file={config.tls_ca_file!r}")

adapter = KubeMQTaskProcessorAdapter(config)
```

Internally, `_build_client_config()` constructs a `TLSConfig` from the four TLS-related fields and passes it to `ClientConfig`:

```python title="adapter.py (_build_client_config)"
from kubemq.core.config import TLSConfig

tls_config = TLSConfig(
    enabled=self._config.tls,
    cert_file=self._config.tls_cert_file or None,
    key_file=self._config.tls_key_file or None,
    ca_file=self._config.tls_ca_file or None,
)
```

<Callout type="info">
  On `KubeMQAdapterConfig`, `tls` is a simple boolean toggle. On the underlying SDK's `ClientConfig`, however, `tls` is a `TLSConfig` object — not a bool. The adapter performs this translation for you, so you only ever set the boolean and the certificate paths.
</Callout>

## Mutual TLS [#mutual-tls]

For mutual TLS — where the broker also verifies the client's certificate — supply all three certificate paths in addition to `tls=True`: `tls_cert_file` (the client certificate), `tls_key_file` (the client private key), and `tls_ca_file` (the shared CA).

```python title="examples/connection/mtls_setup.py"
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

# Full mTLS: client cert + key + CA
config = KubeMQAdapterConfig(
    address="localhost:50000",
    tls=True,
    tls_cert_file="client.pem",
    tls_key_file="client-key.pem",
    tls_ca_file="ca.pem",
)

print(f"tls_cert_file={config.tls_cert_file!r}")
print(f"tls_key_file={config.tls_key_file!r}")
print(f"tls_ca_file={config.tls_ca_file!r}")

adapter = KubeMQTaskProcessorAdapter(config)
```

The three files map directly onto the `TLSConfig` shown above: `tls_cert_file` becomes `cert_file`, `tls_key_file` becomes `key_file`, and `tls_ca_file` becomes `ca_file`. Each is normalized to `None` when left empty, so the same config shape covers plain TLS (CA only) and mutual TLS (CA plus client credentials).

The table below summarizes which fields each connection mode requires:

| Mode                | `tls`   | `tls_ca_file` | `tls_cert_file` | `tls_key_file` |
| ------------------- | ------- | ------------- | --------------- | -------------- |
| Plaintext (default) | `False` | —             | —               | —              |
| TLS (server auth)   | `True`  | required      | —               | —              |
| Mutual TLS          | `True`  | required      | required        | required       |

## Health Checks [#health-checks]

The adapter exposes two health-check methods, both backed by the SDK's `ping()` call against the broker:

* `health_check()` returns a single `bool` — `True` when the broker is reachable, `False` otherwise. Ideal for readiness and liveness probes.
* `health_check_sync()` returns a list with one status dict: `[{"healthy": True}]` on success, or `[{"healthy": False, "error": "..."}]` when the ping fails or the client is not initialized.

```python title="examples/connection/health_check.py"
adapter.initialize(consumer_concurrency=1, task_processor_config=_Cfg())

# Simple boolean health check
healthy = adapter.health_check()
print(f"health_check() -> {healthy}")

# Detailed health check returning a list of status dicts
detailed = adapter.health_check_sync()
print(f"health_check_sync() -> {detailed}")

# Use in a readiness probe pattern
if healthy:
    print("Broker is reachable — ready to process tasks.")
else:
    print("Broker is NOT reachable — check connection settings.")
```

`health_check()` is a thin convenience wrapper: it calls `health_check_sync()` and returns the `healthy` flag from the first entry. Use the boolean form for probes and the detailed form when you want the error string for logging.

## Environment-Variable Configuration [#environment-variable-configuration]

For twelve-factor deployments, build the whole config from environment variables so the same image runs unchanged across environments. Every field has a sensible default, so unset variables fall back gracefully.

```python title="examples/connection/env_var_config.py"
import os

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

# Build config entirely from environment variables with defaults
config = KubeMQAdapterConfig(
    address=os.environ.get("KUBEMQ_ADDRESS", "localhost:50000"),
    client_id=os.environ.get("KUBEMQ_CLIENT_ID", ""),
    auth_token=os.environ.get("KUBEMQ_AUTH_TOKEN", ""),
    tls=os.environ.get("KUBEMQ_TLS", "false").lower() == "true",
    tls_ca_file=os.environ.get("KUBEMQ_TLS_CA_FILE", ""),
    tls_cert_file=os.environ.get("KUBEMQ_TLS_CERT_FILE", ""),
    tls_key_file=os.environ.get("KUBEMQ_TLS_KEY_FILE", ""),
)

print(f"address={config.address}")
print(f"client_id={config.client_id!r}")
print(f"auth_token={'[set]' if config.auth_token else '[not set]'}")
print(f"tls={config.tls}")
```

Note the `tls` line: the environment carries strings, so the example compares `KUBEMQ_TLS` against `"true"` to produce the boolean the config expects. In Kubernetes, these same variables are populated from a `ConfigMap` (address, client ID) and a `Secret` (auth token, certificate paths) — see the Kubernetes scenario below for the full Secret and ConfigMap wiring.

## Related [#related]

<Cards>
  <Card title="Configuration" href="/integrations/rayserve/how-to/configuration" description="Every KubeMQAdapterConfig field — result backend, poll timeout, message sizes, and timeouts." />

  <Card title="Kubernetes Production Deployment" href="/integrations/rayserve/scenarios/kubernetes-production-deployment" description="Wire auth tokens and TLS certificates from Kubernetes Secrets in a production deployment." />
</Cards>