Python SDK

The Python SDK is a zero-dependency, never-throws helper for the TokenTriage proxy. It emits x-tt-* headers and reads back the receipt the server stamped — it forwards and reads; it never computes a cost of its own.

Install

pip install tokentriage
# adapters are opt-in extras (each pulls only its own client):
pip install "tokentriage[httpx]"      # or [requests], [langchain], [crewai], [otel]

import tokentriage as tt

Configure

tt.init(...) is the one entry that can raise (TokenTriageConfigError); resolution is arg → env → default:

tt.init(
    base_url="http://127.0.0.1:8787",     # env: TOKENTRIAGE_BASE_URL
    admin_url="http://127.0.0.1:9090",    # env: TOKENTRIAGE_ADMIN_URL (for the ledger pull)
    log="warn",                            # "debug" | "warn" | "off"
    agent="my-service",                    # default dims applied to every request
)

The five recognized env vars are TOKENTRIAGE_BASE_URL, TOKENTRIAGE_ADMIN_URL, TOKENTRIAGE_DISABLED, TOKENTRIAGE_LOG, and TOKENTRIAGE_ADMIN_TOKEN (the admin bearer, read only at ledger-call time and never stored).

The public API

• Context: trace(agent=None, user=None, tenant=None, feature=None, tags=None, trace_id=None) and step(agent=None, step_class=None, user=None, tenant=None, feature=None, tags=None, parent=None, step_id=None) — each is both a context manager and a decorator. observe(...) is decorator sugar over step. current_step() returns the active Step | None.
• Wire: headers(**overrides) returns the x-tt-* dict (override keys: agent, tier, cache, user, tenant, feature, tags); receipt(source, status=None) parses a response (or a header mapping) into a typed Receipt; last_receipt() returns the most recent one recorded on the open scope.
• Adapters: httpx_client(*, origins=None, base_url=None, **kwargs), async_httpx_client(...), instrument_httpx(client, ...), instrument_requests(session, ...); from tokentriage.adapters.langchain import TTCallbackHandler; from tokentriage.adapters.crewai import instrument_crew; from tokentriage.adapters import otel (otel.bridge()).
• Ledger & doctor: from tokentriage import ledger → ledger.fetch(receipt_or_req_id, admin_url=None, wait_s=0.0), ledger.fetch_trace(...), ledger.feedback(receipt_or_req_id, ok, note=None, admin_url=None); tt.check(admin_url=None, *, probe=True).

Examples

Library-agnostic manual path (stdlib only — the SDK is optional sugar over the wire):

import json, urllib.request, tokentriage as tt

tt.init(base_url="http://127.0.0.1:8787")
with tt.trace(agent="triage", user="u-42", tenant="acme", feature="chat", tags={"env": "prod"}):
    with tt.step(agent="planner", step_class="user_turn"):
        body = json.dumps({"model": "gpt-4o-mini",
                           "messages": [{"role": "user", "content": "hi"}]}).encode()
        req = urllib.request.Request("http://127.0.0.1:8787/v1/chat/completions",
                                     data=body,
                                     headers={"Content-Type": "application/json", **tt.headers()})
        with urllib.request.urlopen(req) as resp:
            rcpt = tt.receipt(resp)           # typed, read-only; never raises
    print(rcpt.request_id, rcpt.mode, rcpt.decision, rcpt.cost_display)
    print(rcpt.dashboard_url())               # deep link to the ledger record

httpx adapter behind the OpenAI SDK, with the honest cost from the ledger:

import tokentriage as tt
from openai import OpenAI

tt.init(base_url="http://127.0.0.1:8787", admin_url="http://127.0.0.1:9090")
client = OpenAI(base_url="http://127.0.0.1:8787/v1", http_client=tt.httpx_client())

with tt.trace(agent="quickstart"):
    client.chat.completions.create(model="gpt-4o-mini", messages=[{"role": "user", "content": "hi"}])
    row = tt.ledger.fetch(tt.last_receipt(), wait_s=2)   # server-typed dollar
    if row.cost and row.cost.state == "priced":
        print(row.cost.usd)                              # Decimal — never fabricated

First-run doctor + feedback write:

import tokentriage as tt
from tokentriage import ledger

tt.init(base_url="http://127.0.0.1:8787", admin_url="http://127.0.0.1:9090")
report = tt.check()                                       # prints + returns CheckReport
if report.ok:
    ack = ledger.feedback(tt.last_receipt(), ok=False, note="bad answer")  # raises on failure
    print(ack.accepted, ack.verdict, ack.source)