Pin a Pricing Snapshot Before an AI Agent Run

Long-running AI agents do not fail like normal API calls.

A chat completion either returns or it does not. An agent run may plan, search, call tools, retry a failing route, summarize intermediate state, and then ask another model to write the final answer. By the time the user sees a result, the original request may have turned into ten model calls across two or three routes.

That is useful when the work is complex. It is also a billing problem if the system does not know which prices and limits were in force when the run started.

The fix is to treat pricing as an admission input, not a page someone checks after the invoice looks strange. Before an agent run begins, the gateway should pin a pricing snapshot, calculate a budget envelope, record the selected billing group, and attach those facts to every model call that belongs to the run.

This article shows one practical pattern for doing that in an OpenAI-compatible gateway.

The problem with late pricing lookup

Most teams start with a simple model wrapper:

response = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=messages,
)

That wrapper is enough for a demo. It is not enough for a production agent.

An agent run has several cost states:

State	What can change
Planning	The model may read a large reusable prefix and decide which tools to call.
Tool use	Search, retrieval, code execution, or browser steps can add context.
Retry	A failed call may be retried on the same model or a fallback route.
Compression	The agent may summarize history before the next step.
Final answer	A stronger route may write the final response.

If pricing is fetched only when the invoice is rendered, the system can accidentally explain yesterday's run with today's catalog. That becomes more likely when providers expose separate buckets for uncached input, cached input, output, context creation, tool use, and group-specific multipliers.

The request needs a durable answer to a plain question: which pricing facts admitted this run?

Use a dated gateway snapshot

Before writing this article, I checked AIWave's public pricing endpoint on 2026-08-31 at 21:11 Asia/Shanghai time. The endpoint returned success=true, pricing_version=a42d372ccf0b5dd13ecf71203521f9d2, 63 model records, auto_groups=["default"], and group ratios of default=3 and vip=1.

The same snapshot exposed route fields for models such as:

Route	model_ratio	completion_ratio	cache_ratio	Enabled groups
deepseek-v4-flash	0.319	3	0.0318	default, vip, svip
deepseek-v4-pro	0.957	3	0.0333	default, vip, svip
glm-5.1	1.05	3.142857	0.32381	default, vip, svip
kimi-k3	2.25	5	0.2	default, vip, svip

Those fields are useful because they are machine-readable. A gateway can use them before routing, and support can use the same version later when reviewing a run.

This does not mean every article, dashboard, or invoice should invent a new public price table. For public claims, keep a dated source link beside the number. For internal admission control, store the snapshot version and the fields used by the policy.

Admission control should happen before planning

Agent systems often run a planner first, then decide what work to do. That planner still costs money. It can also load a large prompt prefix.

Admission control should happen before the planner call:

1. Fetch or load a fresh enough pricing snapshot.
2. Resolve the user's effective billing group.
3. Estimate the starting context size.
4. Choose allowed routes for the run.
5. Calculate a budget envelope.
6. Create an agent_run record with the snapshot version.
7. Reject, downscope, or ask for confirmation if the run cannot fit.

The point is not to predict every token exactly. The point is to define the cost boundary before the agent starts making decisions.

A compact run record

Use one run record and many request records. The run record pins policy. The request records describe what happened.

create table agent_run (
  run_id text primary key,
  account_id_hash text not null,
  effective_group text not null,
  pricing_version text not null,
  pricing_source text not null,
  pricing_checked_at text not null,
  budget_usd_limit real not null,
  budget_token_limit integer not null,
  status text not null,
  started_at text not null,
  finished_at text
);

create table agent_model_call (
  call_id text primary key,
  run_id text not null references agent_run(run_id),
  parent_call_id text,
  attempt integer not null,
  purpose text not null,
  requested_model text not null,
  resolved_model text not null,
  input_tokens integer,
  cache_hit_tokens integer,
  output_tokens integer,
  status text not null,
  provider_error_code text,
  created_at text not null
);

The purpose field should use boring values: plan, tool_summary, retrieval_answer, retry, compression, or final_answer. Avoid a blended "agent cost" row. It hides the behavior you will need to debug.

Fetch the snapshot once

The gateway does not need to call pricing on every model call inside a run. It needs a snapshot that is current under your policy.

import datetime as dt
import json
import urllib.request

PRICING_URL = "https://aiwave.live/api/pricing"


def utc_now() -> str:
    return dt.datetime.now(dt.timezone.utc).isoformat()


def fetch_pricing_snapshot() -> dict:
    req = urllib.request.Request(
        PRICING_URL,
        headers={"User-Agent": "agent-admission-example"},
    )
    with urllib.request.urlopen(req, timeout=20) as response:
        payload = json.loads(response.read().decode("utf-8"))

    if payload.get("success") is not True:
        raise RuntimeError("pricing endpoint did not return success=true")

    rows = payload.get("data") or []
    return {
        "source": PRICING_URL,
        "checked_at": utc_now(),
        "pricing_version": payload.get("pricing_version"),
        "group_ratio": payload.get("group_ratio") or {},
        "rows_by_model": {row["model_name"]: row for row in rows},
    }

You can cache this result for a short window. The important part is that the run stores the version it used, not just the model name.

Estimate with token buckets

Agent budgets should keep input, cached input, and output separate. A single blended estimate is usually where surprises enter the system.

def estimate_call_units(
    *,
    model_row: dict,
    group_ratio: float,
    uncached_input_tokens: int,
    cached_input_tokens: int,
    output_tokens: int,
) -> float:
    model_ratio = float(model_row["model_ratio"])
    completion_ratio = float(model_row["completion_ratio"])
    cache_ratio = float(model_row.get("cache_ratio") or 0)

    return (
        uncached_input_tokens * model_ratio * group_ratio
        + cached_input_tokens * cache_ratio * group_ratio
        + output_tokens * completion_ratio * group_ratio
    )

This example returns relative billable units because gateway catalogs often express model ratios before invoice conversion. Your production version can convert those units into USD, credits, quota, or invoice rows. Keep the source unit explicit in the field name.

Admit the run

The admission function should be small enough to test.

import hashlib
import os
import uuid


def digest_account(account_id: str) -> str:
    salt = os.environ["AUDIT_HASH_SALT"]
    return hashlib.sha256((salt + account_id).encode("utf-8")).hexdigest()[:20]


def admit_agent_run(
    *,
    account_id: str,
    effective_group: str,
    requested_models: list[str],
    estimated_uncached_tokens: int,
    estimated_cached_tokens: int,
    estimated_output_tokens: int,
    budget_token_limit: int,
    budget_usd_limit: float,
    snapshot: dict,
) -> dict:
    group_ratio = float(snapshot["group_ratio"][effective_group])
    rows_by_model = snapshot["rows_by_model"]

    missing = [model for model in requested_models if model not in rows_by_model]
    if missing:
        raise ValueError(f"unknown model route: {missing[0]}")

    estimates = {}
    for model in requested_models:
        estimates[model] = estimate_call_units(
            model_row=rows_by_model[model],
            group_ratio=group_ratio,
            uncached_input_tokens=estimated_uncached_tokens,
            cached_input_tokens=estimated_cached_tokens,
            output_tokens=estimated_output_tokens,
        )

    return {
        "run_id": str(uuid.uuid4()),
        "account_id_hash": digest_account(account_id),
        "effective_group": effective_group,
        "pricing_version": snapshot["pricing_version"],
        "pricing_source": snapshot["source"],
        "pricing_checked_at": snapshot["checked_at"],
        "requested_models": requested_models,
        "estimated_billable_units_by_model": estimates,
        "budget_token_limit": budget_token_limit,
        "budget_usd_limit": budget_usd_limit,
        "status": "admitted",
    }

The example uses a hash for the account id and keeps credentials out of the record. Do the same for token ids, workspace ids, and support links.

Route every call through the run

Once the run exists, every model call should inherit its pricing policy.

from openai import OpenAI

client = OpenAI(
    base_url="https://aiwave.live/v1",
    api_key=os.environ["AIWAVE_API_KEY"],
)


def create_model_call(run: dict, *, purpose: str, model: str, messages: list[dict]):
    if model not in run["requested_models"]:
        raise ValueError(f"model {model} was not admitted for this run")

    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0,
        max_tokens=800,
    )

    usage = response.usage
    return {
        "run_id": run["run_id"],
        "purpose": purpose,
        "resolved_model": model,
        "pricing_version": run["pricing_version"],
        "input_tokens": getattr(usage, "prompt_tokens", None),
        "output_tokens": getattr(usage, "completion_tokens", None),
        "status": "success",
    }

The wrapper blocks a quiet route change. If the planner wants to use a model that was not admitted, it has to ask the admission layer for a new run policy or fail closed.

Retries need a separate rule

Retries should not consume an unlimited share of the budget just because the agent framework is trying to be helpful.

Define retry rules against the run budget:

Retry case	Policy
Provider timeout before output	Retry once on the same route if the remaining budget allows it.
Provider error after partial output	Record the partial call and require an explicit continuation policy.
Context overflow	Do not retry blindly. Compress or reject with a clear reason.
Route unavailable	Use only fallback routes admitted at run start.
Tool failure	Do not spend model tokens until the tool error is classified.

Store retries as model calls with parent_call_id and attempt. That keeps retry spend visible.

select
  run_id,
  resolved_model,
  count(*) as calls,
  sum(case when attempt > 1 then 1 else 0 end) as retry_calls,
  sum(coalesce(input_tokens, 0)) as input_tokens,
  sum(coalesce(output_tokens, 0)) as output_tokens
from agent_model_call
where created_at >= datetime('now', '-1 day')
group by run_id, resolved_model;

This query gives engineering and finance the same view of the run. If retry calls are high, the issue may be timeout policy, SDK behavior, or provider stability, not the rate card.

What the user should see

Most users do not need the full event log. They do need enough evidence to trust the result.

A usage page for an agent run should show:

Field	Why it helps
Run status	Completed, interrupted, failed, or stopped by budget
Models used	Shows actual routes, not only the requested family
Pricing snapshot	Gives a dated source for the calculation
Token buckets	Separates input, cached input, and output where available
Retry count	Explains extra calls without hiding them
Budget result	Shows whether the run stopped inside the approved envelope

This is especially important for teams evaluating a multi-model gateway from a Tier 1 or Tier 2 market. They may be fine with a flexible route. They are usually not fine with an invoice that cannot explain which route, group, and token bucket produced the charge.

What not to store

An admission record is not a transcript archive.

Do not store raw API keys, full prompts, full completions, payment details, or customer-identifying data in the billing trail. Store ids, hashes, route names, pricing versions, token counts, statuses, and source timestamps. If support needs prompt-level inspection for a specific case, put that behind a separate access-controlled workflow with a short retention window.

This boundary matters during procurement. A stronger answer is not "we keep everything." A stronger answer is: the gateway keeps enough billing evidence to explain the run while keeping content and credentials out of the accounting log.

Operational checks

Run these checks daily:

Check	Failure signal
Missing pricing version	A run cannot be recalculated later.
Stale snapshot	Admission used a catalog older than policy allows.
Unknown group	The effective group was not present in the snapshot.
Unadmitted route	A call used a model outside the run policy.
Retry overspend	Retry calls consumed more than the allowed budget share.
Missing usage	A successful call returned no usage fields.
Secret pattern	A key-like string entered logs or article drafts.

The last check belongs in content pipelines too. A technical article can show environment-variable usage, but it should never publish a real key or a customer-specific trace.

Source links to keep with the runbook

Keep dated source links close to the admission code:

Source	Use
AIWave pricing	Public gateway pricing and route context
https://aiwave.live/api/pricing	Machine-readable snapshot for admission
AIWave model docs	Route names and model catalog context
DeepSeek pricing	Provider bucket calibration for DeepSeek routes
QwenCloud pricing	Context and tool billing behavior

Recheck sources before changing policy. Do not assume a provider's rate card, cache bucket, or tool billing rule stayed the same because yesterday's tests passed.

Implementation checklist

Before your agent planner sends its first model call, the gateway should know:

1. Which pricing snapshot admitted the run.
2. Which effective billing group applies.
3. Which model routes are allowed.
4. Which budget envelope the run must stay inside.
5. How retries consume that envelope.
6. Where token buckets will be recorded.
7. Which evidence can be shown to support without exposing secrets.

This is a small amount of structure, but it changes the failure mode. Instead of explaining an agent bill after the fact, the gateway can say what it allowed, what actually ran, and which dated pricing source tied those two facts together.

For high-volume agent workloads, that is the difference between a flexible gateway and an auditable one.