trusted-router-py 0.4.0

Python SDK for TrustedRouter.

TrustedRouter Python SDK

OpenAI-compatible Python client for TrustedRouter — the hosted, attested LLM router that lets you point one OpenAI-shaped client at every provider (Anthropic, OpenAI, Google Vertex, Gemini, DeepSeek, Mistral, Cerebras) and prove the prompt path doesn't log.

• Inference API: https://api.trustedrouter.com/v1
• Control API: https://trustedrouter.com/v1
• Trust release: https://trust.trustedrouter.com
• Source: https://github.com/Lore-Hex/trusted-router-py
• License: Apache-2.0

pip install trusted-router-py                  # base client
pip install trusted-router-py[attestation]     # + GCP attestation verification

Quick start

from trustedrouter import TrustedRouter, AUTO_MODEL

with TrustedRouter(api_key="sk-tr-v1-...") as client:
    resp = client.chat_completions(
        model=AUTO_MODEL,                     # "trustedrouter/auto" — multi-provider failover
        messages=[{"role": "user", "content": "hello"}],
    )
    print(resp.choices[0].message.content)    # typed: ChatCompletion model

chat_completions(...) defaults to AUTO_MODEL when model= is omitted, so the simplest possible call is client.chat_completions(messages=[...]).

Fusion

Fan a request across a panel of models and let a judge model pick or synthesize one answer. fusion(...) (and AsyncTrustedRouter.fusion(...)) returns the same typed ChatCompletion as chat_completions. FUSION_FREEDOM_PANEL, FUSION_FREEDOM_FALLBACK_JUDGES, and FUSION_FREEDOM_FALLBACK_FINALS are the recommended most-permissive config.

from trustedrouter import (
    TrustedRouter,
    FUSION_FREEDOM_FALLBACK_FINALS,
    FUSION_FREEDOM_PANEL,
    FUSION_FREEDOM_FALLBACK_JUDGES,
)

with TrustedRouter(api_key="sk-tr-v1-...") as client:
    resp = client.fusion(
        messages=[{"role": "user", "content": "explain how mRNA vaccines work"}],
        analysis_models=FUSION_FREEDOM_PANEL,       # the panel
        # omit selection_strategy to use synthesize_non_refusals
        fallback_judges=FUSION_FREEDOM_FALLBACK_JUDGES,  # tried in order if a judge refuses/fails
        fallback_final_models=FUSION_FREEDOM_FALLBACK_FINALS,  # tried in order for synthesis
        max_completion_tokens=800,
        timeout=600,
    )
    print(resp.choices[0].message.content)

Or build the spec with fusion_tool(...) and pass it to any chat call. preset="quality" or "budget" selects a built-in panel.

Every method returns a typed pydantic model — IDE autocomplete + runtime validation. Need a dict? Call .model_dump():

resp.model_dump()["choices"][0]["message"]["content"]

Streaming

for token in client.chat_completions_stream(
    model=AUTO_MODEL,
    messages=[{"role": "user", "content": "Write a haiku"}],
):
    print(token, end="", flush=True)

chat_completions_chunk_stream(...) yields the raw OpenAI chat.completion.chunk dicts (with finish_reason, model, id) when you need more than just the text delta.

Async

Every method on TrustedRouter is mirrored on AsyncTrustedRouter as a coroutine; streaming methods return AsyncIterators. Use it from FastAPI, asyncio, or any event-loop-driven app:

import asyncio
from trustedrouter import AsyncTrustedRouter

async def main():
    async with AsyncTrustedRouter(api_key="sk-tr-v1-...") as client:
        async for token in client.chat_completions_stream(
            model="trustedrouter/auto",
            messages=[{"role": "user", "content": "hi"}],
        ):
            print(token, end="", flush=True)

asyncio.run(main())

Bases and failover

Inference defaults to https://api.trustedrouter.com/v1, which is the global load-balancer apex. Regional failover re-requests that apex with the existing backoff; failover is handled server-side, and per-region hostnames are not used.

Pass base_url= for a custom inference endpoint, such as a self-hosted gateway. base_url controls only inference-plane calls: chat completions, messages, responses, response input-token counting, embeddings, and attestation. Catalog, account, billing, broadcast, and OAuth calls use control_base_url instead, defaulting to https://trustedrouter.com/v1:

client = TrustedRouter(
    api_key="sk-tr-v1-...",
    base_url="https://inference.internal/v1",
    control_base_url="https://control.internal/v1",
)

Overriding base_url no longer changes models(), providers(), regions(), credits(), activity(), billing checkout, auth, OAuth, or broadcast calls.

Typed errors

Every HTTP failure raises a typed subclass of TrustedRouterError so callers can discriminate without inspecting status codes:

from trustedrouter import (
    TrustedRouter, AuthenticationError, RateLimitError,
    BadRequestError, EndpointNotSupportedError, NotFoundError, InternalError,
)

try:
    client.chat_completions(messages=[...])
except RateLimitError as e:
    time.sleep(e.retry_after or 5)        # honors Retry-After header
except AuthenticationError:
    refresh_key()
except BadRequestError as e:
    log.warning("bad request: %s", e)
except EndpointNotSupportedError:
    disable_optional_feature()
except InternalError:
    pass                                   # auto-retried; still failing

All subclasses inherit from TrustedRouterError, so existing except TrustedRouterError blocks keep working.

Automatic retries

By default the client retries 429 and 5xx responses up to 2 times with exponential backoff + jitter (capped at 30s, honors Retry-After). Disable with max_retries=0:

client = TrustedRouter(api_key="...", max_retries=0)   # raise immediately on transient

Per-call extras

Every chat method (and request() for ad-hoc paths) accepts:

kwarg	use
api_key=	override the instance bearer for this call only (threadsafe — used by validate_bearer)
extra_headers=	dict of headers to merge in (trace IDs, custom routing)
workspace_id=	sets X-TrustedRouter-Workspace for workspace-scoped management calls
idempotency_key=	adds Idempotency-Key: so the gateway dedupes retries — strongly recommended for billing
timeout=	override the client-level timeout for this call

client.billing_checkout(
    amount=25,
    payment_method="stablecoin",
    idempotency_key=f"checkout-{user_id}-{order_id}",   # never double-charge
)

Sign in with TrustedRouter

Let users "bring their own TrustedRouter account" instead of pasting a key: the OpenRouter-style OAuth PKCE flow mints a user-scoped key so LLM calls are billed to that user's credits. create_oauth_authorization(...) builds the authorize URL and returns the code_verifier + state to keep across the redirect; exchange_oauth_key(...) swaps the returned code for the delegated key + verified identity. Async variants (exchange_oauth_key_async, fetch_userinfo_async) mirror these. OAuth helpers default to the control API origin (https://trustedrouter.com/v1).

from trustedrouter import create_oauth_authorization, exchange_oauth_key, fetch_userinfo

# 1. sign-in: keep auth.code_verifier + auth.state in the user's session
auth = create_oauth_authorization(
    callback_url="https://myapp.com/auth/callback",
    key_label="My App", limit="5", usage_limit_type="monthly",
)
redirect_to(auth.url)

# 2. in /auth/callback (verify state == saved state first)
token = exchange_oauth_key(code=request.args["code"], code_verifier=saved_verifier)
store_for_user(token.key, token.identity)        # sk-tr-v1-… + {sub, email, …}

# 3. anytime
who = fetch_userinfo(api_key=token.key)          # {sub, email, …}

Full flow, endpoints, and security notes: Sign in with TrustedRouter.

Attestation verification (the differentiator)

Every TrustedRouter response is generated inside a Google Confidential Space workload. The gateway's /attestation endpoint mints a Google-signed JWT that commits to the workload image digest, image reference, your nonce, and the TLS leaf cert SHA-256. Verifying it proves the prompt path you're about to use is the exact build the trust page advertises:

import secrets, ssl, socket
from trustedrouter import TrustedRouter
from trustedrouter.attestation import (
    verify_gateway_attestation, policy_from_trust_release,
)

# Pull the published image digest/reference from the trust page
policy = policy_from_trust_release()                 # or pin one explicitly

with TrustedRouter(api_key="sk-tr-v1-...") as client:
    nonce = secrets.token_hex(16)
    jwt = client.attestation()                       # raw JWT bytes

    # Bind the JWT to the live TLS connection's cert
    with ssl.create_default_context().wrap_socket(
        socket.create_connection(("api.trustedrouter.com", 443)),
        server_hostname="api.trustedrouter.com",
    ) as s:
        cert_der = s.getpeercert(binary_form=True)

    attestation = verify_gateway_attestation(
        jwt, policy=policy, nonce_hex=nonce, tls_cert_der=cert_der
    )
    print("verified gateway:", attestation.image_digest)

verify_gateway_attestation() raises AttestationVerificationError on any of: bad signature, expired JWT, wrong issuer, audience mismatch, image_digest mismatch, image_reference mismatch, missing nonce echo, or TLS cert mismatch. Never returns falsey for a failed verification.

This codepath needs cryptography; install with pip install trusted-router-py[attestation].

Bring your own httpx client

Pass client= if you need a custom transport (cert pinning, retries you manage, observability hooks). The SDK won't close it on aclose():

import httpx
from trustedrouter import AsyncTrustedRouter

my_client = httpx.AsyncClient(
    timeout=30.0,
    event_hooks={"response": [my_cert_pin_hook]},
)
sdk = AsyncTrustedRouter(api_key="...", client=my_client)
# ...use sdk...
await sdk.aclose()        # no-op for caller-owned clients
await my_client.aclose()  # caller still owns lifecycle

This is exactly how the Quill device wraps the SDK so it can pin Quill Cloud's self-signed leaf cert via an httpx event hook, while delegating chat streaming to the SDK.

CLI

pip install exposes a trustedrouter console script for sniff tests:

export TRUSTEDROUTER_API_KEY=sk-tr-v1-...

trustedrouter chat "hello"                 # one-shot completion
trustedrouter chat --stream "long answer"  # token-by-token
trustedrouter regions                      # list deployed regions
trustedrouter providers                    # list provider catalog
trustedrouter models                       # list model catalog
trustedrouter trust                        # show trust release
trustedrouter attest                       # raw JWT bytes (pipe to `jq`-able tools)

Other endpoints

client.models()             # OpenAI-shape catalog
client.providers()          # provider list
client.regions()            # deployed regions
client.credits(workspace_id="ws_...")  # current prepaid balance for a workspace
client.activity(since="2026-01-01", limit=50)
client.messages(            # Anthropic-shape, preserves system + content blocks
    model="anthropic/claude-3-5-sonnet",
    messages=[{"role": "user", "content": "hi"}],
    max_tokens=512,
)
client.billing_checkout(amount=25, payment_method="stablecoin", idempotency_key=...)

These catalog, account, billing, and broadcast helpers are control-plane calls and use control_base_url, not the inference base_url.

client.embeddings(...) is present for API compatibility, but the hosted TrustedRouter route currently raises EndpointNotSupportedError instead of returning fake vectors. Use client.models() / /embeddings/models to inspect the future embedding catalog.

For routes the SDK doesn't wrap, drop down to client.request(...):

client.request("GET", "/some/new/route", headers={"x-trace": "abc"})

Roadmap

• v0.3 (shipped): typed pydantic response models — every method returns a typed model. Migration: replace resp["k"] with resp.k, or call resp.model_dump() to get the dict back. Models use extra="allow" so the gateway can add fields without an SDK release.
• v0.4 (shipped): default inference host is https://api.trustedrouter.com/v1; catalog/account/billing/OAuth/broadcast calls use the control plane at https://trustedrouter.com/v1 with a new control_base_url= override.
• v0.x: Regional failover improvements for the GCP Confidential Space path.

Contributing

uv sync --group dev
uv run ruff check .
uv run pytest                              # ~110 tests, ≥85% coverage gate

CI runs lint + tests on every push to main and PR. Coverage gate is enforced — PRs that drop coverage below 85% fail. Add tests with new public surface.