Request controls and observations

The configured APIs on this page belong to the unreleased local SDK update. Start with a routed answer, then choose the controls your application needs.

Handle a failed request

Factories preserve the official provider exception. Read error_details(error) in Python or errorDetails(error) in TypeScript for the response status, category, request ID and retry information. Missing metadata stays absent. These examples print structured metadata and re-raise the original provider error.

import os
import sys

from openai import APIError
from tokentriage.client import TokenTriage, TokenTriageError, error_details

NAMES = ("TOKENTRIAGE_GATEWAY_BASE", "TOKENTRIAGE_RUNTIME_KEY",
         "TOKENTRIAGE_GATEWAY_ID", "TOKENTRIAGE_ENVIRONMENT_ID")
missing = [name for name in NAMES if not os.environ.get(name)]
if missing:
    sys.exit(f"Set {', '.join(missing)} before running this example.")
base, runtime_key, gateway_id, environment_id = (os.environ[name] for name in NAMES)

try:
    with TokenTriage(base_url=base, api_key=runtime_key) as connection:
        client = connection.openai(gateway_id=gateway_id, environment_id=environment_id)
        answer = client.chat.completions.create(
            model="gateway",
            messages=[{"role": "user", "content": "Say hello in one sentence."}],
        )
        print(answer.choices[0].message.content)
except (TokenTriageError, APIError) as error:
    # A configuration error is raised locally and never reaches the gateway.
    details = error_details(error)
    print(f"Request failed: category={details.category} code={details.code} "
          f"status={details.status} request_id={details.request_id}", file=sys.stderr)
    # Re-raise the original exception unchanged; anything else shows its own text.
    raise

import OpenAI from "openai";
import { TokenTriage, TokenTriageError, errorDetails, gatewayModel } from "tokentriage/client";

const names = ["TOKENTRIAGE_GATEWAY_BASE", "TOKENTRIAGE_RUNTIME_KEY",
  "TOKENTRIAGE_GATEWAY_ID", "TOKENTRIAGE_ENVIRONMENT_ID"];
const missing = names.filter((name) => !process.env[name]);
if (missing.length > 0) {
  console.error(`Set ${missing.join(", ")} before running this example.`);
  process.exit(1);
}
const [baseUrl, apiKey, gatewayId, environmentId] = names.map((name) => process.env[name] ?? "");

let connection: TokenTriage | undefined;
try {
  connection = new TokenTriage({ baseUrl, apiKey });
  const client = await connection.openai({ gatewayId, environmentId });
  const answer = await client.chat.completions.create({
    model: gatewayModel,
    messages: [{ role: "user", content: "Say hello in one sentence." }],
  });
  console.log(answer.choices[0]?.message.content);
} catch (error) {
  if (error instanceof TokenTriageError || error instanceof OpenAI.APIError) {
    // A configuration error is raised locally and never reaches the gateway.
    const details = errorDetails(error);
    console.error(`Request failed: category=${details.category} code=${details.code} `
      + `status=${details.status} request_id=${details.requestId}`);
  }
  // Rethrow the original exception unchanged; anything else shows its own text.
  throw error;
} finally {
  connection?.close();
}

Failure	Next step
Missing configuration	Set the assigned gateway URL and both required credentials.
Account authentication failure	Check the account key and assigned deployment; keep the provider key in its provider option.
Provider authentication failure	Check the original provider error and credential.
Rate limit	Respect Retry-After and check whether replay is safe before retrying.
Timeout or cancelled stream	Close the stream; a partially delivered answer has no automatic replay guarantee.
No response request ID	Check the destination and status; older responses may lack additive metadata.

Factory inference retries default to zero. Existing-client wrappers preserve the caller’s setting. SDK instrumentation adds no retry loop or direct-provider fallback. Timeouts use seconds in new Python controls and milliseconds in new TypeScript controls.

Observe completion locally

Callbacks can inspect request IDs, observed timing, usage and outcome without capturing prompts or answers. The SDK inspects a body as your code consumes it. Finish or close a stream to receive its terminal observation.

import asyncio
import json
import os
import sys

import tokentriage as tt
from tokentriage import observations
from tokentriage.client import TokenTriage

NAMES = ("TOKENTRIAGE_GATEWAY_BASE", "TOKENTRIAGE_RUNTIME_KEY",
         "TOKENTRIAGE_GATEWAY_ID", "TOKENTRIAGE_ENVIRONMENT_ID")
missing = [name for name in NAMES if not os.environ.get(name)]
if missing:
    sys.exit(f"Set {', '.join(missing)} before running this example.")
base, runtime_key, gateway_id, environment_id = (os.environ[name] for name in NAMES)


def completed(value: observations.Observation) -> None:
    if value.phase == "headers":
        return
    print("Observation:", json.dumps({
        "version": value.version,
        "phase": value.phase,
        "request_id": value.gateway_request_id,
        "trace_id": value.trace_id,
        "total_ms": value.timings.get("total_ms"),
        "usage_source": value.usage_source,
    }))


async def main() -> None:
    subscription = observations.register_observer(completed, max_queue=64)
    try:
        async with TokenTriage(base_url=base, api_key=runtime_key) as connection:
            client = connection.async_openai(gateway_id=gateway_id, environment_id=environment_id)
            try:
                with tt.trace(feature="example"):
                    response = await client.chat.completions.with_raw_response.create(
                        model="gateway",
                        messages=[{"role": "user", "content": "Say hello in one sentence."}],
                    )
                    print("Request ID:", connection.metadata(response).request_id)
            except Exception as error:
                details = connection.error_details(error)
                print(f"Request failed: code={details.code} status={details.status} "
                      f"request_id={details.request_id}", file=sys.stderr)
                raise
    finally:
        result = await subscription.ashutdown(timeout_s=1)
        state = subscription.stats()
        print("Observer state:", json.dumps({
            "pending": result.pending,
            "timed_out": result.timed_out,
            "dropped": state.dropped,
            "failures": state.failures,
        }))


asyncio.run(main())

import * as tt from "tokentriage";
import { TokenTriage, gatewayModel } from "tokentriage/client";

const names = ["TOKENTRIAGE_GATEWAY_BASE", "TOKENTRIAGE_RUNTIME_KEY",
  "TOKENTRIAGE_GATEWAY_ID", "TOKENTRIAGE_ENVIRONMENT_ID"];
const missing = names.filter((name) => !process.env[name]);
if (missing.length > 0) {
  console.error(`Set ${missing.join(", ")} before running this example.`);
  process.exit(1);
}
const [baseUrl, apiKey, gatewayId, environmentId] = names.map((name) => process.env[name] ?? "");

const connection = new TokenTriage({ baseUrl, apiKey });
const detach = connection.onObservation((event) => {
  if (event.phase === "headers") return;
  console.log("Observation:", JSON.stringify({
    version: event.version,
    phase: event.phase,
    requestId: event.metadata?.requestId,
    traceId: event.traceId,
    elapsedMs: event.timings.completedMs,
    usageSource: event.usageSource,
  }));
});
try {
  const client = await connection.openai({ gatewayId, environmentId });
  await tt.trace({ feature: "example" }, async () => {
    const result = await client.chat.completions.create({
      model: gatewayModel,
      messages: [{ role: "user", content: "Say hello in one sentence." }],
    }).withResponse();
    console.log("Request ID:", connection.metadata(result.response).requestId);
  });
} catch (error) {
  const details = connection.errorDetails(error);
  console.error(`Request failed: code=${details.code} status=${details.status} request_id=${details.requestId}`);
  process.exitCode = 1;
} finally {
  detach();
  const result = await connection.shutdown(1000);
  console.log("Observer state:", JSON.stringify({ ...result, ...connection.observationStats() }));
}

Managed requests already produce gateway records. Observing them locally does not export another independent request to Central. Managed gateway records are not currently visible in the customer dashboard.

Use native chat with metadata

The optional native facade composes the verified official Chat Completions client. It preserves tools, structured request options and provider stream types. native.provider exposes the guarded official client. Other provider resources still depend on your deployment’s endpoint support.

import asyncio
import os
import sys

from tokentriage.client import TokenTriage

NAMES = ("TOKENTRIAGE_GATEWAY_BASE", "TOKENTRIAGE_RUNTIME_KEY",
         "TOKENTRIAGE_GATEWAY_ID", "TOKENTRIAGE_ENVIRONMENT_ID")
missing = [name for name in NAMES if not os.environ.get(name)]
if missing:
    sys.exit(f"Set {', '.join(missing)} before running this example.")
base, runtime_key, gateway_id, environment_id = (os.environ[name] for name in NAMES)


async def main() -> None:
    async with TokenTriage(base_url=base, api_key=runtime_key) as connection:
        client = connection.async_native(gateway_id=gateway_id, environment_id=environment_id)
        try:
            result = await client.chat.completions.with_metadata(
                model="gateway",
                messages=[{"role": "user", "content": "Say hello in one sentence."}],
                stream=True,
            )
        except Exception as error:
            details = connection.error_details(error)
            sys.exit(f"Request failed: code={details.code} status={details.status} "
                     f"request_id={details.request_id}")
        print("Request ID:", result.metadata.request_id)
        try:
            async for chunk in result.data:
                for choice in chunk.choices:
                    if choice.delta.content:
                        print(choice.delta.content, end="", flush=True)
            print()
        finally:
            await result.data.close()


asyncio.run(main())

import { TokenTriage, gatewayModel } from "tokentriage/client";

const names = ["TOKENTRIAGE_GATEWAY_BASE", "TOKENTRIAGE_RUNTIME_KEY",
  "TOKENTRIAGE_GATEWAY_ID", "TOKENTRIAGE_ENVIRONMENT_ID"];
const missing = names.filter((name) => !process.env[name]);
if (missing.length > 0) {
  console.error(`Set ${missing.join(", ")} before running this example.`);
  process.exit(1);
}
const [baseUrl, apiKey, gatewayId, environmentId] = names.map((name) => process.env[name] ?? "");

const connection = new TokenTriage({ baseUrl, apiKey });
try {
  const client = await connection.native({ gatewayId, environmentId });
  const result = await client.chat.completions.withMetadata({
    model: gatewayModel,
    messages: [{ role: "user", content: "Say hello in one sentence." }],
    stream: true,
  });
  console.log("Request ID:", result.metadata.requestId);
  try {
    for await (const chunk of result.data) {
      process.stdout.write(chunk.choices[0]?.delta.content ?? "");
    }
    console.log();
  } finally {
    result.data.controller.abort();
  }
} catch (error) {
  const details = connection.errorDetails(error);
  console.error(`Request failed: code=${details.code} status=${details.status} request_id=${details.requestId}`);
  process.exitCode = 1;
} finally {
  await connection.shutdown(1000);
}

Choose an advanced integration

Task	Guide
Manage your own deployment, paginate results, or subscribe to ledger events	Management requests and streams
Read ledger records, submit feedback, or check admin configuration	Ledger and admin controls
Route locally, export events, or add OTel attempt spans in Python	Python observations and propagation
Export events from Node or workers, recover a spool, or add OTel in TypeScript	TypeScript observations and propagation
Resolve a stable diagnostic code	Diagnostic actions
Upgrade an existing application	Migration and support

Admin access is separate from managed inference. Use it only with your own deployment’s admin URL and credentials. A response request ID needs no admin access. No automatic initialize-once provider bootstrap is supported; use a factory or the options helper when constructing the provider client.