outlaya

Outlaya's Python SDK for server-side AI cost attribution. Wrap a supported provider client, read usage from its response, and send one normalized event to Outlaya without putting the ingest call on your request path.

Supported providers:

The SDK requires Python 3.9+ and has no runtime dependencies. Provider packages are optional extras.

Install

pip install outlaya

Install the matching provider extra to pin a tested provider SDK:

pip install "outlaya[openai]"
pip install "outlaya[anthropic]"
pip install "outlaya[google]"
pip install "outlaya[all]"

Quickstart

import os

import outlaya
from openai import OpenAI

outlaya.init(api_key=os.environ["OL_API_KEY"])

client = outlaya.wrap_openai(
    OpenAI(),
    metadata={"environment": "production"},
)

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}],
    ol_metadata={"customer_id": customer.id, "feature": "chat"},
)

The model call returns the provider's response unchanged. The usage event is queued and sent in the background. ol_metadata is stripped before the request reaches the provider.

Environment variables:

You can also pass api_key and base_url to outlaya.init() or outlaya.OutlayaClient().

Attribution metadata

Precedence, lowest to highest:

1. Wrapper defaults: wrap_openai(client, metadata={...}).
2. Execution scope: with outlaya.metadata({...}):.
3. Per-call: ol_metadata= / extra_body={"ol_metadata": {...}}.

import outlaya

with outlaya.metadata({"team": "support", "session_id": run_id}):
    agent.run(input)

with_metadata(values, callback) is available when a callback reads more naturally. Coroutine callbacks are awaited inside the scope.

First-class keys: customer_id, feature, team, environment, prompt_key, prompt_version, session_id, turn_id, region, deployment_type, service_tier. Any other scalar becomes a custom tag. custom_tags merge across layers; scalar fields override.

Providers

OpenAI

from openai import OpenAI

import outlaya

client = outlaya.wrap_openai(OpenAI())

Async clients work too:

from openai import AsyncOpenAI

client = outlaya.wrap(AsyncOpenAI())
response = await client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}],
)

Sync and async streaming are supported. For OpenAI chat completions the SDK injects stream_options={"include_usage": True} so the final chunk carries usage. Events are submitted when the stream finishes.

for chunk in client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}],
    stream=True,
):
    ...

# Or as a context manager:
with client.chat.completions.create(..., stream=True) as stream:
    for chunk in stream:
        ...

Anthropic

import anthropic

import outlaya

client = outlaya.wrap_anthropic(anthropic.Anthropic())
client.messages.create(
    model="claude-sonnet-4",
    max_tokens=256,
    messages=[{"role": "user", "content": "Hello"}],
)

with client.messages.stream(
    model="claude-sonnet-4",
    max_tokens=256,
    messages=[{"role": "user", "content": "Hello"}],
) as stream:
    message = stream.get_final_message()

The adapter supports messages.create non-streaming, raw stream=True, and the messages.stream() manager on sync and async clients. Cache reads are folded into input_tokens and sent separately as cached_tokens; cache-write TTL legs are preserved.

Gemini

from google import genai

import outlaya

client = outlaya.wrap_google_genai(genai.Client(api_key="..."))
client.models.generate_content(
    model="gemini-2.0-flash",
    contents="Hello",
)

for chunk in client.models.generate_content_stream(
    model="gemini-2.0-flash",
    contents="Hello",
):
    ...

The aio client is supported. Gemini thoughts_token_count is sent as custom_tags.reasoning_tokens.

OpenRouter

OpenRouter is OpenAI-compatible.

from openai import OpenAI

import outlaya

router = outlaya.wrap_openrouter(
    OpenAI(
        api_key=os.environ["OPENROUTER_API_KEY"],
        base_url="https://openrouter.ai/api/v1",
    )
)

The adapter adds extra_body={"usage": {"include": True}} and forwards the provider-reported cost as reported_cost_usd_micros, plus upstream provider, upstream inference cost, and BYOK flags for reconciliation.

Generic dispatch

wrap(client) detects the client shape and dispatches to the right adapter. Use the explicit wrapper when the provider is already known.

client = outlaya.wrap(OpenAI())

Cursor

Cursor is an IDE/agent product, so it is a pull integration rather than a wrapped client. Run the sync from a server-side job:

import os

import outlaya

outlaya.init(api_key=os.environ["OL_API_KEY"])

result = outlaya.sync_cursor_usage(
    api_key=os.environ["CURSOR_ADMIN_API_KEY"],
    since=last_synced_at,
    until=now,
    resolve_customer_id=lambda record: member_to_customer.get(record.get("email")),
    flush=True,
)

print(result.fetched, result.submitted, result.skipped)

Cursor records are sent as provider: "cursor", source: "pull", operation: "coding_agent", with a stable cursor:<id> idempotency key. The default transport targets POST {base_url}/teams/usage on the Cursor Admin API; pass transport= to override it if your Cursor contract differs.

Delivery

• submit() only enqueues. It never waits for ingest and never raises.
• The queue is bounded by JSON bytes (16 MiB by default) and optionally by event count. When full, the newest event is dropped and counted.
• Requests carry up to 50 events by default, with a 4 MiB body cap, a 500-event hard cap, a 15 second timeout, and two retries with exponential backoff.
• HTTP 413 halves the batch until it fits.
• Failed retryable batches return to the front of the queue and retry in the background.
• Each event gets an idempotency_key if missing; resends reuse it.
• Call outlaya.flush() before a script, Lambda, or job returns when you need to confirm delivery. On CPython, pending sends also flush at interpreter exit.

import outlaya

outlaya.flush()

Inspect queue and drop counters:

client = outlaya.get_client()
print(client.stats())

Set throw_on_error=True to raise collected delivery failures from flush(), or pass on_error= to observe them without raising. On platforms that restrict background threads, construct the client with auto_start=False and call flush() synchronously before returning.

Manual reporting

For harnesses that cannot be wrapped:

import outlaya

outlaya.track_usage(
    provider="openai",
    model="gpt-4o-mini",
    usage={"input_tokens": 120, "output_tokens": 40, "cached_tokens": 20},
    latency_ms=830,
    metadata={"customer_id": customer.id, "feature": "summarize"},
)

Usage accepts both Outlaya names (input_tokens, cached_input_tokens) and provider/JavaScript aliases (prompt_tokens, completion_tokens, cachedInputTokens). Do not report calls that a wrapped client already tracks.

Async and threads

Provider adapters preserve the client's sync or async nature. Metadata is stored in contextvars, so it propagates to asyncio tasks and to work submitted with asyncio.to_thread. Plain threading.Thread instances start with an empty context. Background delivery uses daemon threads; call outlaya.flush() or client.shutdown() before a short-lived process exits. In async code, run the blocking flush off the event loop:

import asyncio

import outlaya

await asyncio.to_thread(outlaya.flush)

Development

python -m venv .venv
. .venv/bin/activate
pip install -e ".[dev]"
pytest
ruff check src tests
mypy src/outlaya

Build distribution artifacts with:

python -m build

Licensing

MIT. See LICENSE and NOTICE. The architecture and event contract are derived from the MIT-licensed Outlaya JavaScript SDK.