Lago Agent SDK

https://github.com/getlago/lago-agent-sdk-python
Admin
Lago Agent SDK is a Python library that wraps LLM clients (Bedrock, Anthropic, OpenAI, Mistral,...
Tokens:40,896
Snippets:258
Trust Score:9.5
License:MIT
Update:1 day ago
Show doc for...
Tokens:
Raw
### Development setup commands

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Clone the repository, create a virtual environment, install the package in editable mode with dev dependencies, and run tests. This is the standard workflow for contributing to the SDK.

```bash
git clone https://github.com/getlago/lago-agent-sdk-python
cd lago-agent-sdk-python
python -m venv venv && source venv/bin/activate
pip install -e '.[dev]'
pytest
```

--------------------------------

### Set up with plain pip

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/CONTRIBUTING.md

Alternative setup using a standard Python virtual environment and pip. Requires Python 3.11 and installs the package with development extras.

```bash
python3.11 -m venv venv && source venv/bin/activate
pip install -e '.[dev]'
```

--------------------------------

### Gemini Quickstart with LagoSDK wrap

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Shows the minimal setup for using the Lago SDK with Google's Gemini client. Requires a Lago API key and a default subscription ID. The `sdk.flush()` call ensures billing data is sent.

```python
from google import genai
from lago_agent_sdk import LagoSDK

sdk = LagoSDK(api_key="...", default_subscription_id="sub_acme")
client = sdk.wrap(genai.Client(api_key="..."))

resp = client.models.generate_content(
    model="gemini-2.5-flash",
    contents="Hello",
)
sdk.flush()
```

--------------------------------

### Clone and sync with uv

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/CONTRIBUTING.md

Sets up the development environment using uv, which provides faster installs and a reproducible lockfile-driven environment. Run these commands from the project root after cloning the repository.

```bash
git clone https://github.com/getlago/lago-agent-sdk-python
cd lago-agent-sdk-python
uv sync --all-extras       # creates .venv, installs from uv.lock
```

--------------------------------

### Run linting and type checks directly

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/CONTRIBUTING.md

Runs each linting and type-checking tool individually with uv.

```bash
# or directly:
uv run ruff check src tests
uv run ruff format --check src tests
uv run mypy src
```

--------------------------------

### Install lago-agent-sdk with pip

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Install the base package. Optional extras add provider dependencies: [bedrock] for boto3, [mistral] for mistralai, [anthropic] for anthropic, [openai] for openai, and [gemini] for google-genai.

```bash
pip install lago-agent-sdk
```

--------------------------------

### Quickstart with OpenAI using LagoSDK

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Wrap an OpenAI client with LagoSDK to track usage. The wrapped client behaves identically to the original, with the SDK adding an in-memory queue that batches events to Lago in the background. Call sdk.flush() to send queued events.

```python
from openai import OpenAI
from lago_agent_sdk import LagoSDK

sdk = LagoSDK(api_key="...", default_subscription_id="sub_acme")
client = sdk.wrap(OpenAI(api_key="..."))

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}],
    max_completion_tokens=200,
)
sdk.flush()
```

--------------------------------

### Quickstart with Mistral using LagoSDK

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Wrap a Mistral client with LagoSDK to track usage. The wrapped client behaves identically to the original, with the SDK adding an in-memory queue that batches events to Lago in the background. Call sdk.flush() to send queued events.

```python
from mistralai.client import Mistral
from lago_agent_sdk import LagoSDK

sdk = LagoSDK(api_key="...", default_subscription_id="sub_acme")
client = sdk.wrap(Mistral(api_key="..."))

resp = client.chat.complete(
    model="mistral-small-latest",
    messages=[{"role": "user", "content": "Hello"}],
)
sdk.flush()
```

--------------------------------

### Quickstart with Anthropic using LagoSDK

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Wrap an Anthropic client with LagoSDK to track usage. The wrapped client behaves identically to the original, with the SDK adding an in-memory queue that batches events to Lago in the background. Call sdk.flush() to send queued events.

```python
from anthropic import Anthropic
from lago_agent_sdk import LagoSDK

sdk = LagoSDK(api_key="...", default_subscription_id="sub_acme")
client = sdk.wrap(Anthropic(api_key="..."))

resp = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=200,
    messages=[{"role": "user", "content": "Hello"}],
)
sdk.flush()
```

--------------------------------

### Install the released package with pip

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/RELEASING.md

After the workflow publishes the package, customers can install the exact version with pip. Replace 0.2.0 with the released version.

```bash
pip install lago-agent-sdk==0.2.0
```

--------------------------------

### Run unit tests with coverage

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/CONTRIBUTING.md

Runs unit tests with a coverage report using pytest and the coverage plugin.

```bash
# Unit tests with coverage report
uv run pytest tests/unit --cov=lago_agent_sdk --cov-report=term-missing
```

--------------------------------

### Run unit tests

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/CONTRIBUTING.md

Runs the fast unit tests that do not require network access.

```bash
# Unit tests (fast, no network)
make test
```

--------------------------------

### Common Makefile workflows

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/CONTRIBUTING.md

Lists the primary Makefile targets for syncing dependencies, running tests, linting, formatting, and running the full CI check.

```bash
make sync     # install/sync deps from uv.lock
make test     # unit tests
make lint     # ruff check + ruff format --check + mypy
make format   # auto-fix lint and format
make check    # lint + test (what CI runs)
```

--------------------------------

### Quickstart with Amazon Bedrock using LagoSDK

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Wrap a boto3 Bedrock Runtime client with LagoSDK to track usage. The wrapped client behaves identically to the original, with the SDK adding an in-memory queue that batches events to Lago in the background. Call sdk.flush() to send queued events.

```python
import boto3
from lago_agent_sdk import LagoSDK

sdk = LagoSDK(
    api_key="<YOUR_LAGO_API_KEY>",
    api_url="https://api.getlago.com/api/v1/",
    default_subscription_id="sub_acme",
)
client = sdk.wrap(boto3.client("bedrock-runtime", region_name="eu-west-1"))

resp = client.converse(
    modelId="eu.amazon.nova-lite-v1:0",
    messages=[{"role": "user", "content": [{"text": "Hello"}]}],
)
sdk.flush()
```

--------------------------------

### Run all linting and type checks

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/CONTRIBUTING.md

Runs ruff check, ruff format check, and mypy together via the Makefile.

```bash
make lint        # all three at once
```

--------------------------------

### Bump a single package

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/CONTRIBUTING.md

Updates only the specified package in the lockfile. Replace X with the package name.

```bash
uv lock --upgrade-package X  # bump a single package
```

--------------------------------

### Refresh the lockfile

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/CONTRIBUTING.md

Updates the uv.lock file to the latest compatible versions of all dependencies. Commit the resulting diff.

```bash
uv lock --upgrade            # refresh the lockfile (commit the diff)
```

--------------------------------

### Cut a release: bump version, commit, tag, and push

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/RELEASING.md

Steps to trigger the publish workflow. Update the version in pyproject.toml and add release notes to CHANGELOG.md, then commit, push, tag with a v*.*.* tag, and push the tag. The workflow runs CI, verifies the tag version, builds sdist+wheel, and publishes to PyPI via OIDC trusted publishing.

```bash
# 1. Update the version
$EDITOR pyproject.toml              # bump version, e.g. 0.1.0 -> 0.2.0
$EDITOR CHANGELOG.md                # add release notes under a new heading

# 2. Commit + push
git commit -am "Release 0.2.0"
git push

# 3. Tag and push the tag — this triggers the publish workflow
git tag v0.2.0
git push --tags
```

--------------------------------

### sdk.warm_pricing()

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/cloudflare.md

Pre-fetches pricing for partner models so that even the first call bills dollars instead of tokens. This avoids the initial token billing and on_error report.

```APIDOC
## sdk.warm_pricing()

### Description
Pre-fetches pricing for partner models so that even the first call bills dollars instead of tokens. This avoids the initial token billing and on_error report.

### Parameters
- **models** (list) - Required - List of model categories, e.g., ["workers-ai"]
- **workers_ai_models** (list) - Optional - List of partner model IDs to warm pricing for, e.g., ["typesafe/jev"]

### Example
```python
sdk.warm_pricing(["workers-ai"], workers_ai_models=["typesafe/jev"])
```
```

--------------------------------

### Databricks AI Gateway - Wrap OpenAI Client

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Configures an OpenAI client to point at the Databricks AI Gateway. BYOK calls bill dollar cost at the vendor's rates; Databricks-hosted models (`system.ai.*`) bill token counts. Use `sdk.backfill_databricks` to fill in what `wrap()` didn't see.

```python
client = sdk.wrap(OpenAI(
    api_key=DATABRICKS_TOKEN,
    base_url=f"{DATABRICKS_HOST}/ai-gateway/mlflow/v1",
    default_headers={"Databricks-Ai-Gateway-Request-Tags": json.dumps({"lago_subscription": "sub_acme"})},
))
```

--------------------------------

### Backfill Databricks usage with DatabricksSource.from_env()

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/databricks.md

Backfills Databricks usage for a window (e.g., "7 days") using credentials from DATABRICKS_HOST, DATABRICKS_TOKEN, and DATABRICKS_WAREHOUSE_ID environment variables. The window is floored to whole closed hours and excludes the current hour; pass a datetime for an exact lower bound or unified=True to bill the whole window to default_subscription. The printed output shows a sample result dict.

```python
from lago_agent_sdk.gateway.databricks import DatabricksSource

source = DatabricksSource.from_env()   # DATABRICKS_HOST / _TOKEN / _WAREHOUSE_ID
print(sdk.backfill_databricks(source, "7 days", default_subscription="sub_default"))
sdk.flush()
# {'cost': 60, 'tokens': 47, 'skipped': 0}

```

--------------------------------

### Cloudflare AI Gateway - Workers AI Models via SDK Client

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Uses the SDK's own one-method client for Workers AI models that have no client to wrap, including partner models like `typesafe/jev`. Requires the partner's key stored under the gateway's Provider Keys (BYOK) and the gateway name. Cloudflare bills nothing; the partner bills you.

```python
ai = sdk.workers_ai(account_id, cf_api_token, gateway_id=gateway_id, gateway_auth=gateway_auth)
out = ai.run("typesafe/jev", {"state": ticket_text, "questions": {...}}, extra_lago={"subscription": "sub_acme"})
out = ai.run("@cf/meta/llama-3.2-3b-instruct", {"messages": [{"role": "user", "content": "Hello"}]})
```

--------------------------------

### Architecture diagram of lago-agent-sdk

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

ASCII diagram showing how your code calls a wrapped client, which extracts usage and sends Lago events to api.getlago.com.

```text
                                ┌────────────────┐
            your code ────────► │ wrapped client │ ───► provider (Bedrock / Mistral / …)
                                └───────┬────────┘
                                        │ (extract usage)
                                        ▼
                                ┌────────────────┐
                                │  Lago events   │ ───► api.getlago.com
                                └────────────────┘
```

--------------------------------

### Snowflake Cortex - Wrap OpenAI Client and Backfill

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Wraps the Snowflake Cortex REST surface as an OpenAI-compatible client and backfills AI SQL functions from Snowflake's usage views. Everything bills as token counts. Requires Snowflake account and PAT environment variables.

```python
client = sdk.wrap(OpenAI(
    base_url=f"https://{os.environ['SNOWFLAKE_ACCOUNT']}.snowflakecomputing.com/api/v2/cortex/v1",
    api_key=os.environ["SNOWFLAKE_PAT"],
))

sdk.backfill_snowflake(SnowflakeSource.from_env(), "7 days", default_subscription="sub_default")
```

--------------------------------

### Ramp Router - Wrap OpenAI Client

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Points an OpenAI client at the Ramp Router gateway. The model that answered is the one billed, even if Router serves a different model than requested. Requires the Ramp Router API key and model environment variables.

```python
client = sdk.wrap(
    OpenAI(api_key=os.environ["RAMP_ROUTER_API_KEY"], base_url="https://api.router.com/v1"),
    subscription="sub_acme",
)
client.responses.create(model=os.environ["RAMP_ROUTER_MODEL"], input="Summarize this invoice.")
```

--------------------------------

### Initialize and run Workers AI partner and catalog models with sdk.workers_ai

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/cloudflare.md

Use this client for Cloudflare Workers AI models that are not chat-shaped or not addressable by the official cloudflare package. It requires a Cloudflare API token with Workers AI access; gateway_id and gateway_auth are optional. run() returns Cloudflare's full response envelope unchanged and raises WorkersAIError on error status (402 for missing partner key, 403 for model not on plan) before billing. Streaming is not supported; use the /compat endpoint through a wrapped OpenAI client for streamed chat.

```python
ai = sdk.workers_ai(
    account_id,
    cf_api_token,                       # a Cloudflare API token with Workers AI access
    gateway_id=gateway_id,              # optional — see what it buys below
    gateway_auth=gateway_auth,          # the gateway's cf-aig-authorization token, if authentication is on
    subscription="sub_acme",            # default for every call; extra_lago={"subscription": ...} per call
)

# a partner model: the partner's key must be stored under the gateway's Provider Keys (BYOK)
out = ai.run(
    "typesafe/jev",
    {
        "state": "I was charged twice and need the duplicate refunded before Friday.",
        "questions": {
            "department": {"type": "choice", "instructions": "Which team should handle this?",
                           "criteria": {"billing": "Payments, refunds", "technical": "Bugs, outages"}},
        },
    },
    extra_lago={"dimensions": {"ticket": "T-4821"}},
)
out["result"]["result"]["answers"]["department"]["choice"]   # "billing"

# a catalog model: same client, same billing
ai.run("@cf/meta/llama-3.2-3b-instruct", {"messages": [{"role": "user", "content": "Hello"}], "max_tokens": 50})
sdk.flush()
```

--------------------------------

### Backfill Snowflake source from environment

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/snowflake.md

Initializes a SnowflakeSource from SNOWFLAKE_ACCOUNT/_PAT environment variables plus a warehouse, then backfills 7 days of data with a default subscription. The returned dict reports billed tokens and skipped rows; both skip causes are also reported through on_error with where="backfill".

```python
from lago_agent_sdk.gateway.snowflake import SnowflakeSource

source = SnowflakeSource.from_env()  # SNOWFLAKE_ACCOUNT / _PAT, plus a warehouse
print(sdk.backfill_snowflake(source, "7 days", default_subscription="sub_default"))
sdk.flush()
# {'tokens': 47, 'skipped': 0}
```

--------------------------------

### Wrap OpenAI client with LagoSDK for Ramp Router

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/ramp-router.md

Point an OpenAI client at Ramp Router's base_url and wrap() detects it automatically. The SDK bills the model the response reports, and streaming bills once from the terminal usage event. The typed Python client rejects the non-standard 'models' kwarg, so a fallback list must go through extra_body.

```python
from openai import OpenAI
from lago_agent_sdk import LagoSDK

sdk = LagoSDK(api_key=os.environ["LAGO_API_KEY"])
client = sdk.wrap(
    OpenAI(
        api_key=os.environ["RAMP_ROUTER_API_KEY"],
        base_url="https://api.router.com/v1",
    ),
    subscription="sub_acme",
)

# A model id from GET /v1/models — Router's ids are account-specific.
client.responses.create(model=os.environ["RAMP_ROUTER_MODEL"], input="Summarize this invoice.")
sdk.flush()
```

--------------------------------

### Anthropic client with Ramp Router for Messages surface

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/ramp-router.md

Use an Anthropic client to reach Router's /v1/messages surface, the only surface where cold cache writes are priced exactly. Requires RAMP_ROUTER_API_KEY and RAMP_ROUTER_MODEL environment variables. The system prompt includes cache_control for ephemeral caching.

```python
from anthropic import Anthropic

client = sdk.wrap(
    Anthropic(api_key=os.environ["RAMP_ROUTER_API_KEY"], base_url="https://api.router.com"),
    subscription="sub_acme",
)
client.messages.create(
    model=os.environ["RAMP_ROUTER_MODEL"],
    max_tokens=256,
    system=[{"type": "text", "text": LONG_SYSTEM_PROMPT, "cache_control": {"type": "ephemeral"}}],
    messages=[{"role": "user", "content": "Summarize this invoice."}],
)
```

--------------------------------

### Configure pricing_mode="price" with markup

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

By default the SDK emits token counts (pricing_mode="tokens"). Set pricing_mode="price" to emit dollar cost of each call: Σ(unit_price_per_token × tokens) × markup. Price mode emits one llm_cost event per priced field, each carrying precise_total_amount_cents for Lago's dynamic charge model plus a token_type property. Prices are fetched and cached in the background; if unavailable, falls back to token-count events and reports via on_error.

```python
from lago_agent_sdk import LagoSDK, LagoConfig

sdk = LagoSDK(api_key="...", config=LagoConfig(
    api_key="...",
    default_subscription_id="sub_123",
    pricing_mode="price",     # "tokens" (default) | "price"
    markup=1.2,               # optional cost multiplier (1.2 = +20%)
))
```

--------------------------------

### Cloudflare AI Gateway - Wrap Anthropic Client

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Configures an Anthropic client to point at a Cloudflare AI Gateway. Cache hits (`cf-aig-cache-status: HIT`) are not billed, and Workers AI models are priced automatically. Requires your Cloudflare account ID, gateway ID, and gateway auth token.

```python
client = sdk.wrap(Anthropic(
    api_key="...",
    base_url=f"https://gateway.ai.cloudflare.com/v1/{account_id}/{gateway_id}/anthropic",
    default_headers={"cf-aig-authorization": f"Bearer {gateway_auth}"},
))
```

--------------------------------

### Databricks-hosted foundation models with OpenAI client

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/databricks.md

Use the OpenAI client to call Databricks-hosted foundation models (system.ai.*) via the /ai-gateway/mlflow/v1 endpoint. Billed by Databricks in DBUs. Requires DATABRICKS_TOKEN and DATABRICKS_HOST.

```python
from openai import OpenAI
from lago_agent_sdk import LagoSDK

sdk = LagoSDK(api_key="...", default_subscription_id="sub_acme")
client = sdk.wrap(OpenAI(
    api_key=DATABRICKS_TOKEN,
    base_url=f"{DATABRICKS_HOST}/ai-gateway/mlflow/v1",
    default_headers={"Databricks-Ai-Gateway-Request-Tags": json.dumps({"lago_subscription": "sub_acme"})},
))
client.chat.completions.create(model="system.ai.llama-4-maverick", messages=[{"role": "user", "content": "Hi"}])
```

--------------------------------

### Set QUERY_TAG for subscription attribution

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/snowflake.md

Sets the QUERY_TAG session variable to the lago_subscription key that Snowflake backfill reads for attribution. Untagged rows fall back to default_subscription, or are skipped if none is set. role_names and user_id are opt-in via subscription_order=("query_tag", "role_names").

```sql
ALTER SESSION SET QUERY_TAG = '{"lago_subscription": "sub_123"}';
SELECT AI_COMPLETE('claude-sonnet-4-5', 'summarize this');
```

--------------------------------

### Read Databricks usage rows directly with source.read_usage()

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/databricks.md

Reads usage rows for a window (e.g., "7 days") without billing, each row already shaped for emit(). Access row.usage.model, row.subscription, and row.usd_cost (usd_cost is None for hosted models). Requires a PAT with the sql scope and a SQL warehouse; the live backfill calls do not need these.

```python
for row in source.read_usage("7 days"):
    print(row.usage.model, row.subscription, row.usd_cost)  # usd_cost is None for hosted

```

--------------------------------

### Retag and push after fixing a CI failure

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/RELEASING.md

If CI fails before the build, delete the tag locally and remotely, fix the issue, recommit, then retag and push. This re-triggers the publish workflow.

```bash
git tag -d v0.2.0
git push --delete origin v0.2.0
# fix the issue, recommit
git tag v0.2.0
git push --tags
```

--------------------------------

### sdk.flush()

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/cloudflare.md

Flushes any pending billing events to Lago. Should be called after all AI calls to ensure billing data is sent.

```APIDOC
## sdk.flush()

### Description
Flushes any pending billing events to Lago. Should be called after all AI calls to ensure billing data is sent.

### Example
```python
sdk.flush()
```
```

--------------------------------

### ai.run()

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/cloudflare.md

Runs a Workers AI model. Returns Cloudflare's full response envelope unchanged and raises WorkersAIError on an error status before anything is billed.

```APIDOC
## ai.run()

### Description
Runs a Workers AI model. Returns Cloudflare's full response envelope unchanged and raises WorkersAIError on an error status before anything is billed.

### Parameters
- **model** (string) - Required - Model identifier (e.g., "typesafe/jev" or "@cf/meta/llama-3.2-3b-instruct")
- **input** (object) - Required - Model input; for partner models, an object with `state` and `questions`; for catalog models, an object with `messages` and optional `max_tokens`
- **extra_lago** (object) - Optional - Additional Lago billing parameters (e.g., `dimensions`, `subscription`)
- **extra_headers** (object) - Optional - Additional HTTP headers (e.g., `cf-aig-skip-cache`)

### Returns
Cloudflare's full response envelope.

### Errors
- **WorkersAIError** - Raised on error status (402 for partner model without stored key, 403 for model not on Workers plan)

### Example
```python
out = ai.run(
    "typesafe/jev",
    {
        "state": "I was charged twice and need the duplicate refunded before Friday.",
        "questions": {
            "department": {"type": "choice", "instructions": "Which team should handle this?",
                           "criteria": {"billing": "Payments, refunds", "technical": "Bugs, outages"}},
        },
    },
    extra_lago={"dimensions": {"ticket": "T-4821"}},
)
out["result"]["result"]["answers"]["department"]["choice"]   # "billing"
```

```python
ai.run("@cf/meta/llama-3.2-3b-instruct", {"messages": [{"role": "user", "content": "Hello"}], "max_tokens": 50})
```
```

--------------------------------

### Wrap Snowflake Cortex REST client with sdk.wrap()

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/snowflake.md

Use this snippet to wrap an OpenAI-compatible client for Snowflake Cortex REST calls. The base_url identifies the calls as Snowflake; use max_completion_tokens because Cortex rejects max_tokens. Cached tokens are additive and caching requires an explicit cache_control part.

```python
import os
from openai import OpenAI

client = sdk.wrap(
    OpenAI(
        base_url=f"https://{os.environ['SNOWFLAKE_ACCOUNT']}.snowflakecomputing.com/api/v2/cortex/v1",
        api_key=os.environ["SNOWFLAKE_PAT"],
    )
)
```

--------------------------------

### BYOK Anthropic passthrough with native client

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/databricks.md

Use the Anthropic client for BYOK (bring your own key) via the /ai-gateway/anthropic endpoint. The real credential goes in Authorization, and the Unity Catalog connection holding your Anthropic key is named in Databricks-Model-Provider-Service.

```python
from anthropic import Anthropic
client = sdk.wrap(Anthropic(
    api_key="unused",
    base_url=f"{DATABRICKS_HOST}/ai-gateway/anthropic",
    default_headers={
        "Authorization": f"Bearer {DATABRICKS_TOKEN}",
        "Databricks-Model-Provider-Service": "workspace.default.anthropickey",
    },
))
```

--------------------------------

### Add Router metadata for attribution

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/ramp-router.md

Pass the same subscription id in Router's own metadata field, which Router stores with its usage record and shows in the request detail. The SDK does not send this metadata for you; it is useful for future backfills.

```python
client.responses.create(
    model=os.environ["RAMP_ROUTER_MODEL"],
    input="Summarize this invoice.",
    # Router stores this with its usage record. The SDK does not send it for you.
    metadata={"lago_subscription": "sub_acme"},
)
```

--------------------------------

### Backfill usage from Cloudflare Logs API

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/cloudflare.md

Extract usage from gateway log entries using `extract_cloudflare_log` and `resolve_subscription`. Bills Cloudflare's metered `cost` directly; re-running over the same window never double-bills. For BYOK entries, bill tokens only because `cost` carries Cloudflare's list price even though the partner bills directly.

```python
from lago_agent_sdk.gateway.adapters import extract_cloudflare_log, resolve_subscription

for entry in fetch_gateway_logs():  # GET .../ai-gateway/gateways/{id}/logs
    usage = extract_cloudflare_log(entry)
    sub = resolve_subscription(entry) or "sub_default"  # from the call's cf-aig-metadata, if set
    if usage.extras.get("byok"):
        # Served with the customer's own provider key (BYOK): Cloudflare charged nothing and the
        # partner bills them directly, yet `cost` still carries Cloudflare's list price. Bill the
        # tokens, never that number.
        sdk.emit(usage, subscription=sub, mode="tokens", event_id=f"cf_{entry['id']}")
        continue
    sdk.emit(usage, subscription=sub, mode="price", usd_cost=entry.get("cost") or 0, event_id=f"cf_{entry['id']}")
sdk.flush()
```

--------------------------------

### Configure on_error callback in LagoConfig

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Use this snippet to wire custom error handling for SDK instrumentation failures. The SDK swallows and logs errors by default, but you can provide an on_error callback to capture exceptions with a phase tag. The callback receives the exception and a string indicating where the error occurred.

```python
from lago_agent_sdk import LagoConfig, LagoSDK

def on_error(exc: Exception, where: str) -> None:
    sentry.capture_exception(exc, tags={"sdk_phase": where})

sdk = LagoSDK(
    api_key="...",
    config=LagoConfig(api_key="...", on_error=on_error),
)
```

--------------------------------

### sdk.workers_ai()

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/cloudflare.md

Creates a client for Cloudflare Workers AI models that are not chat-shaped and cannot be wrapped by the OpenAI-compatible client. It handles both partner models and catalog models, with optional gateway integration for caching and billing.

```APIDOC
## sdk.workers_ai()

### Description
Creates a client for Cloudflare Workers AI models that are not chat-shaped and cannot be wrapped by the OpenAI-compatible client. It handles both partner models and catalog models, with optional gateway integration for caching and billing.

### Parameters
- **account_id** (string) - Required - Cloudflare account ID
- **cf_api_token** (string) - Required - Cloudflare API token with Workers AI access
- **gateway_id** (string) - Optional - Gateway ID for caching and billing features
- **gateway_auth** (string) - Optional - The gateway's cf-aig-authorization token, if authentication is on
- **subscription** (string) - Optional - Default subscription for every call; can be overridden per call with extra_lago

### Returns
A client object with a `run()` method.

### Example
```python
ai = sdk.workers_ai(
    account_id,
    cf_api_token,
    gateway_id=gateway_id,
    gateway_auth=gateway_auth,
    subscription="sub_acme",
)
```
```

--------------------------------

### Inspect Snowflake usage rows with read_usage()

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/snowflake.md

Use this generator to read usage rows for a window before billing. Each row is already shaped for emit(), exposing kind, usage.model, subscription, and occurred_at. Useful for inspecting deferred rows listed on source.deferred_rows.

```python
for row in source.read_usage("7 days"):
    print(row.kind, row.usage.model, row.subscription, row.occurred_at)
```

--------------------------------

### Backfill only REST view traffic

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/docs/snowflake.md

Pass views=("rest",) only when no wrapped client is billing REST traffic, because the REST view reports calls wrap() already billed and the idempotency protection (derived from REQUEST_ID) only holds for traffic wrap() never saw.

```python
sdk.backfill_snowflake(
    source,
    "7 days",
    default_subscription="sub_default",
    views=("rest",),  # ONLY if no wrapped client is billing this traffic
)
```

--------------------------------

### Set external_subscription_id per call, context, or init

Source: https://github.com/getlago/lago-agent-sdk-python/blob/main/README.md

Three ways to set external_subscription_id, in priority order: per-call override (highest precedence), context-bound via set_subscription, and default at init. Backed by contextvars for safe propagation across asyncio tasks.

```python
# 1. Per-call override (highest precedence)
client.converse(..., extra_lago={"subscription": "sub_acme", "dimensions": {"feature": "summarize"}})

# 2. Context-bound (use in middleware to set once per request)
sdk.set_subscription("sub_acme")
# all calls in this thread/asyncio task → sub_acme

# 3. Default at init (fallback)
sdk = LagoSDK(api_key="...", default_subscription_id="sub_default")
```
Tokens:
Raw