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=[...]).

Routing, privacy, and orchestration

The SDK exports stable aliases for normal routing (AUTO_MODEL, FAST_MODEL), privacy (ZDR_MODEL, E2E_MODEL, CONFIDENTIAL_MODEL, EU_MODEL, US_MODEL), and named orchestration models such as SOCRATES_MODEL, PROMETHEUS_MODEL, ZEUS_MODEL, and ATHENA_MODEL.

Use ProviderPreferences when privacy or US provider jurisdiction must be a hard routing requirement even with an explicit model. Use EU_MODEL for the EU-focused routing pool:

from trustedrouter import ProviderPreferences

response = client.chat_completions(
    model="z-ai/glm-5.2",
    messages=[{"role": "user", "content": "Review this contract."}],
    provider=ProviderPreferences.confidential(),
)

TrustedRouter's five atomic orchestration primitives have typed builders with the same wire format in every official SDK: fusion_tool (Synth), advisor_tool, selector_tool, map_reduce_tool, and subagent_tool. Named models are the easiest defaults; builders are for custom compositions.

Cost allocation tags

Attach up to 50 AWS-style string tags to an inference request. Tags stay out of prompts and provider payloads and appear on TrustedRouter generation and activity metadata.

response = client.chat_completions(
    model="trustedrouter/zdr",
    messages=[{"role": "user", "content": "Summarize this contract."}],
    tags={
        "environment": "production",
        "team": "legal",
        "cost-center": "legal-01",
    },
    user="user_123",
    session_id="matter_456",
)

The same tags, user, session_id, and trace fields work with Responses, Messages, and Embeddings. Request tags override API key default tags with the same key.

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. The default SDK client probes the published US Central, US East, and Europe gateway health endpoints in parallel on its first inference call, pins the lowest-latency healthy region for the life of that client, and keeps the remaining regions plus the global apex as idempotent failover targets. Reuse one client to preserve region affinity, connection pooling, DNS caching, and improve prompt-cache locality.

Set regional_affinity=False to keep using only the global endpoint. A custom base_url= is never probed or rewritten. An injected httpx client also leaves affinity off by default; opt in explicitly with regional_affinity=True when the injected transport can reach the public regional hosts.

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.

Alias domains

The regions above all live under one name on one DNS provider, and the domain sits above every cloud behind it. A zone that stops answering, a registrar lock, or a resolver handing out a stale record takes the API down no matter how many regions are healthy.

api.allyrouter.com and api.uptimerouter.com are exact aliases of api.trustedrouter.com, on separate domains served by separate DNS providers, resolving to the same attested enclaves. They sit at the end of the candidate list, after the regional gateways, so a healthy deployment never touches them. Nothing to configure — it is on by default.

Failover changes host only on connection failures and on 502, 503, or 504. A 500 means a server received and processed the request. You are not charged twice for it — authorization is idempotent per Idempotency-Key and settlement happens once — but the work would run a second time, so the answer could differ and TrustedRouter pays the provider again. A 500 is retried on the same host.

Aliases are used only for the default base_url. A custom one — a private deployment, a test server, a regional pin — is never rewritten. Pass regional_failover=False to keep every attempt on a single host.

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.

Every error also exposes layer, source, provider, and request_id when the server supplies them. These fields distinguish routing failures from provider errors without requiring callers to parse the message string.

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

Typed inference and control-plane mutation helpers mint one idempotency key at the logical call boundary and reuse it unchanged for every retry. The generic client.request(...) escape hatch deliberately does not guess whether an arbitrary write is idempotent: pass idempotency_key= explicitly if an unsafe method such as POST should be replayed after an ambiguous transport failure or an ordinary retryable status. A failure known to happen before any bytes were sent remains safe to retry without a key.

Telemetry

TrustedRouter reports content-free client reliability telemetry by default when both inference and control use TrustedRouter hosts, so failures in DNS, TLS, connection setup, and broken streams can be measured even when the server never sees the request. Exact minute counters describe outcomes and bounded latency histograms, while sampled diagnostics describe closed-enum attempt, retry, failover, timeout, SDK, runtime, OS, and architecture fields. Delivery runs on a separate daemon thread and HTTP client, retains counters for up to 24 hours and 512 KiB through outages, sends one bounded batch at a time, and never delays or retries an inference request.

Disable both telemetry delivery and the x-tr-client header with TrustedRouter(..., telemetry=False), TRUSTEDROUTER_TELEMETRY=0, or DO_NOT_TRACK=1; custom inference or control hosts default to disabled. Set TRUSTEDROUTER_TELEMETRY_DEBUG=1 to echo the exact outbound batch JSON to stderr, and use telemetry_sample_rate= to lower or raise random sampling of otherwise healthy first-attempt calls. See the complete disclosure at trustedrouter.com/docs/telemetry.

Telemetry never sends prompts, completions, message text, workspace/key/user/ session IDs, IP addresses, or hostnames of custom endpoints.

Per-call extras

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

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

The package includes the official trustedrouter CLI for developers, coding agents, CI jobs, and gateway diagnostics. Install it into an isolated tool environment, or use the console script from your existing SDK environment:

pipx install trusted-router-py
# or: uv tool install trusted-router-py
# or: pip install trusted-router-py

Authentication comes from TRUSTEDROUTER_API_KEY, with TR_API_KEY retained as a backwards-compatible fallback. Agents can also set TRUSTEDROUTER_BASE_URL (or legacy TR_BASE_URL), TRUSTEDROUTER_CONTROL_BASE_URL, and TRUSTEDROUTER_WORKSPACE_ID without putting configuration on the process command line:

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

trustedrouter chat "hello"                 # one-shot completion
echo "summarize this" | trustedrouter chat # prompt from stdin
trustedrouter chat - < prompt.txt           # explicit stdin marker
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 attestation JWT bytes
trustedrouter attest --verify              # verify signature + workload identity
trustedrouter attest --session             # prove live same-socket TLS binding
trustedrouter --version

The raw attest command is available in the base install. Install trusted-router-py[attestation] to use attest --verify or attest --session. --verify validates the signed document against the published workload identity. Only --session proves the live TLS exporter and performs a same-socket follow-up challenge. --connect-ip is accepted only with --session and must not be empty.

Agent and JSON contract

Pass global --json before or after a subcommand. Non-streaming successes emit one compact JSON object on stdout:

{"command":"chat","data":{"id":"...","choices":[...]},"ok":true}

Errors emit one compact object on stderr and never mix in usage text:

{"error":{"message":"invalid","status_code":401,"type":"authentication_error"},"ok":false}

Typed SDK errors use their cross-runtime snake-case class names, such as rate_limit_error and internal_error; non-SDK failures use runtime_error. Attestation successes use attest, attest.verify, or attest.session as the command value so agents can distinguish the three result schemas.

chat --stream --json emits JSON Lines: one chat.delta record per nonempty text delta, followed by one terminal chat.done record. Plain mode keeps the original human-readable output. JSON records are UTF-8, key-sorted, and contain no ANSI formatting, so agents can parse them without checking whether stdout is a terminal.

trustedrouter --json models | jq '.data.data[].id'
printf 'hello' | trustedrouter chat --json
trustedrouter chat --stream --json "count to three"

Stable process exit codes:

For stdin prompts, use - by itself or omit the positional prompt when stdin is piped. Combining - with positional prompt text is rejected as ambiguous, and stdin leading/trailing whitespace is preserved. Input must be valid UTF-8 and is capped at 8 MiB before any network request is made. --retries must be at least 0 and --max-tokens must be at least 1; invalid values exit 2 before constructing a client. An explicit --model must contain a non-whitespace model id.

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.