qbrix 0.2.0

Python SDK for the qbrix multi-armed bandit platform

Python SDK for the qbrix platform.

---

Typed sync and async clients for Qbrix — pool/experiment/gate management and the agent select/feedback loop.

Installation

The SDK speaks to the qbrix proxy over either HTTP/JSON or gRPC — pick a transport (or install both):

pip install qbrix[http]   # HTTP/JSON transport (httpx)
pip install qbrix[grpc]   # gRPC transport (grpcio)
pip install qbrix[all]    # both

pip install qbrix with no extra installs only the models and config layer — constructing a client then raises a clear ImportError naming the extra to add.

Quick Start

Set your credentials as environment variables and call resources directly — no client instantiation needed:

export QBRIX_API_KEY="optiq_xxx"
export QBRIX_BASE_URL="https://cloud.qbrix.io"

import qbrix

# 1. Create a pool of arms (variants)
pool = qbrix.pool.create(
    name="homepage-buttons",
    arms=[{"name": "blue"}, {"name": "green"}, {"name": "red"}],
)

# 2. Create an experiment with a bandit policy
exp = qbrix.experiment.create(
    name="button-color-test",
    pool_id=pool.id,
    policy="BetaTSPolicy",
)

# 3. Select an arm for a user
result = qbrix.agent.select(
    experiment_id=exp.id,
    context={"id": "user-123", "metadata": {"country": "US"}},
)
print(result.arm.name)       # "green"
print(result.is_default)     # False (bandit selected)

# 4. Send feedback (reward) after observing the outcome
qbrix.agent.feedback(request_id=result.request_id, reward=1.0)

The system learns from every reward and adjusts future selections automatically.

Transports

The same Qbrix client, resources, and Pydantic models work over either wire format — only the transport differs:

from qbrix import Qbrix

# HTTP (default)
client = Qbrix(transport="http", base_url="https://cloud.qbrix.io")

# gRPC
client = Qbrix(transport="grpc", base_url="grpcs://cloud.qbrix.io:443")

When transport is omitted it's resolved in this order: the transport= kwarg → the QBRIX_TRANSPORT env var → the base_url scheme (grpc:// / grpcs:// → gRPC) → HTTP.

The gRPC transport covers pool, experiment, gate, and agent operations. The auth, policy, and runtime resources are HTTP-only (the proxy doesn't expose them over gRPC) — calling them on a gRPC client raises NotImplementedError. Install qbrix[all] and use transport="http" if you need them.

Explicit Client

For full control over configuration or lifecycle (e.g. closing the transport connection, using a context manager), instantiate the client directly:

from qbrix import Qbrix

with Qbrix(api_key="optiq_xxx", base_url="https://cloud.qbrix.io") as client:
    pool = client.pool.create(
        name="homepage-buttons",
        arms=[{"name": "blue"}, {"name": "green"}, {"name": "red"}],
    )
    result = client.agent.select(experiment_id="exp-uuid", context={"id": "user-123"})
    client.agent.feedback(request_id=result.request_id, reward=1.0)

Async

from qbrix import AsyncQbrix

async with AsyncQbrix(api_key="optiq_xxx") as client:
    result = await client.agent.select(
        experiment_id="exp-uuid",
        context={"id": "user-456"},
    )
    await client.agent.feedback(request_id=result.request_id, reward=1.0)

Configuration

Constructor kwargs take priority over environment variables, which take priority over defaults.

export QBRIX_API_KEY="optiq_xxx"
export QBRIX_BASE_URL="https://cloud.qbrix.io"

from qbrix import Qbrix

client = Qbrix()  # picks up env vars automatically

Env Var	Default	Description
QBRIX_API_KEY	None	API key (optiq_xxx)
QBRIX_BASE_URL	http://localhost:8080	Proxy service URL
QBRIX_TRANSPORT	(auto)	http or grpc — overrides URL-scheme detection
QBRIX_TIMEOUT	30.0	Request timeout / gRPC deadline (seconds)
QBRIX_MAX_RETRIES	2	Retry count on transient failures (429/5xx, gRPC UNAVAILABLE)

gRPC-only knobs: QBRIX_GRPC_KEEPALIVE_TIME_MS (30000), QBRIX_GRPC_KEEPALIVE_TIMEOUT_MS (10000), QBRIX_GRPC_USE_TLS (false).

Feature Gates

Attach a feature gate to control rollout before the bandit kicks in:

import qbrix

qbrix.gate.create(
    experiment_id=exp.id,
    enabled=True,
    rollout_percentage=80.0,
    default_arm_id=pool.arms[0].id,
    rules=[
        {"key": "plan", "operator": "==", "value": "enterprise", "arm_id": pool.arms[1].id},
    ],
)

# Gate-matched selections return is_default=True
result = qbrix.agent.select(
    experiment_id=exp.id,
    context={"id": "user-789", "metadata": {"plan": "enterprise"}},
)
print(result.is_default)  # True

Error Handling

import qbrix
from qbrix import NotFoundError, RateLimitedError

try:
    exp = qbrix.experiment.get("nonexistent-id")
except NotFoundError as e:
    print(f"Not found: {e.detail}")
except RateLimitedError as e:
    print(f"Retry after {e.retry_after}s")

Supported Policies

Policy	Type	Best For
BetaTSPolicy	Stochastic	Binary rewards (clicks, conversions)
GaussianTSPolicy	Stochastic	Continuous rewards
UCB1TunedPolicy	Stochastic	Theoretical regret guarantees
KLUCBPolicy	Stochastic	Binary rewards with tight bounds
MOSSPolicy	Stochastic	Fixed horizon problems
LinUCBPolicy	Contextual	Linear reward models with features
LinTSPolicy	Contextual	Linear models with uncertainty
EXP3Policy	Adversarial	Non-stationary environments
FPLPolicy	Adversarial	Follow the perturbed leader

License

MIT