arbr-client (Python)

Official Python client for the Arbr AI control plane — one function to route, observe, and govern every LLM call your app makes.

Your app calls the gateway instead of provider SDKs. The gateway holds the provider keys, honors the model you pin (or picks one when you say "auto"), applies human-approved routing rules and cost policies, and logs every call with full cost attribution — visible in the dashboard.

• Zero dependencies — Python ≥ 3.11, stdlib only. Sync and async (achat/astream).
• One function for the 90% case — chat().
• Robust by default — per-attempt timeouts, retries with exponential backoff + jitter on network errors / 429 / 5xx, typed errors.
• Optional LangChain integration — a real BaseChatModel via arbr-client[langchain].

Install

pip install arbr-client                # core (zero deps)
pip install "arbr-client[langchain]"   # + the LangChain BaseChatModel adapter
# (pre-release: pip install /path/to/arbr_client-0.1.0-py3-none-any.whl)

60-second quickstart

from arbr_client import create_client

arbr = create_client(
    "http://localhost:4100",      # or set ARBR_GATEWAY_URL
    application="my-app",         # attribution — shows up in the dashboard
)

res = arbr.chat("Summarise this support ticket: ...", model="auto", max_tokens=300)
print(res.text)
print(res.model, res.routing_decision)   # e.g. "gpt-4o-mini", "ai"

Async (FastAPI, LangGraph, etc.):

res = await arbr.achat("Summarise this ticket: ...", model="auto")

That's a complete integration. No provider keys in your app, and every call is logged, costed, and governable from the dashboard.

How model choice works

res.model_requested shows what you asked for, res.model what served it, res.routing_decision why (explicit / rule / auto / ai / cache / fallback / passthrough), and res.classified_by how the task type was determined (provided / keyword / ai).

API

create_client(base_url=None, *, application=None, workflow=None, department=None, user_id=None, api_key=None, timeout_s=60, retries=2) → Client

base_url falls back to $ARBR_GATEWAY_URL; api_key to $ARBR_API_KEY. A gateway API key (ab_…, dashboard → Settings → API keys) is sent as Authorization: Bearer and binds attribution server-side — required once the gateway has Require API keys on. The metadata kwargs are defaults merged into every call (per-call kwargs override them).

Client.chat(messages, *, model=None, provider=None, task_type=None, temperature=None, max_tokens=None, ...) → ChatResponse

messages accepts a bare string, {"role", "content"} dicts, or LangChain message objects. ChatResponse is a frozen dataclass: text, usage (input_tokens/output_tokens/total_tokens), model, model_requested, provider, routing_decision, classified_by, cache_hit, request_id, plus .raw (the unmodified gateway payload).

Client.achat(...) / Client.astream(...) / Client.astatus()

Async counterparts (the blocking call runs in a worker thread via asyncio.to_thread).

Streaming

The gateway supports two streaming modes:

Real SSE (token-by-token) — use the OpenAI-compatible endpoint at POST /v1/chat/completions with stream=True. Works with the OpenAI Python SDK, any chat UI, or a raw httpx/requests call:

from openai import OpenAI

client = OpenAI(api_key="ab_…", base_url="http://localhost:4100")
stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Tell me a joke"}],
    stream=True,
)
for chunk in stream:
    print(chunk.choices[0].delta.content or "", end="", flush=True)

Client.stream(messages, ...) → Iterator[str] — makes one buffered chat() call and yields the text in small chunks. Useful when you want full routing metadata (res.model, res.routing_decision, etc.) alongside a streaming-style emit:

for chunk in arbr.stream("Explain quantum entanglement simply"):
    print(chunk, end="", flush=True)

Use the OpenAI-compat endpoint when you need real token-by-token delivery or are integrating with chat UIs. Use stream() when you want the routing metadata the OpenAI endpoint doesn't expose.

Client.status() → dict

Healthcheck against GET /api/status — demoMode, liveProviders, defaultProvider, defaultModel, routingMode, breachedCaps. When the gateway has admin auth enabled (ARBR_ADMIN_KEY set server-side), this endpoint requires a credential — your gateway api_key is accepted, so set it and status() keeps working.

Client.models() → dict

List every model available on this Arbr instance — GET /v1/models. Uses the same gateway API key as chat calls (no admin key needed).

result = arbr.models()

# Filter and sort by tier / cost
cheap = sorted(
    [m for m in result["data"] if m["tier"] == "light"],
    key=lambda m: m["inputPer1M"],
)
print(cheap[0]["id"], cheap[0]["provider"])  # e.g. "us.amazon.nova-micro-v1:0", "bedrock-nova"

Response shape is OpenAI-compatible ({"object": "list", "data": [...]}) with Arbr extensions:

Async counterpart: await arbr.amodels().

Client.providers() → dict

List configured live providers — GET /v1/providers. Returns {"object": "list", "data": [{"id": ..., "models": [...]}]}. No credentials exposed.

result = arbr.providers()

for p in result["data"]:
    print(p["id"], "→", len(p["models"]), "models")

# openai       → 2 models
# bedrock-nova → 11 models
# anthropic    → 3 models

Async counterpart: await arbr.aproviders().

Error handling

All failures raise GatewayError with .status, .code, .retryable, .request_id:

LangChain integration

Two options, by how deep your LangChain usage goes:

1. Full BaseChatModel (recommended for LangChain/LangGraph apps) — requires the extra:

from arbr_client import create_client
from arbr_client.langchain import ArbrChatModel

client = create_client("http://localhost:4100", application="my-app")
llm = ArbrChatModel(client=client, model_name="auto", max_tokens=1024)

chain = my_prompt | llm           # full Runnable compatibility:
await chain.ainvoke({...})        # pipes, async, batching, callbacks

2. Zero-dep duck-typed adapter — when you don't want a langchain-core dependency:

from arbr_client import as_langchain_model
llm = as_langchain_model(client, workflow="answer-drafting")
msg = llm.invoke(messages)        # .invoke()/.ainvoke(); AIMessage-shaped result

Out of gateway scope either way: tool calling / with_structured_output, embeddings, and token-level streaming — keep those on direct provider SDKs.

Gradual rollout pattern

Gate the swap at your app's LLM factory so nothing else changes:

def get_llm():
    if os.environ.get("ARBR_GATEWAY_URL"):
        return ArbrChatModel(client=_arbr_client(), model_name=settings.llm_model)
    return build_direct_provider_model()   # unchanged path

Unset ARBR_GATEWAY_URL to revert instantly.

License

MIT