Nobody asks what a certificate costs to draft until the first monthly bill arrives. Then the principal asks, and the honest answer is usually "somewhere in the invoice". That is a bad answer for a workflow a licensed person signs off on, and it is a worse answer when the agency owns the AWS account and the bill.
Token spend is not the interesting number anyway. The number an agency can act on is cost per approved draft: what it costs to put one certificate, one remarket packet, or one service-request activity in front of a person, including the retries and the abstentions. This tutorial builds that number, then puts a cap around it so a runaway loop at 2am cannot spend a month of budget before anyone is awake.
You need the agent already running, a place to write rows (DynamoDB here), and access to the usage fields your model provider returns. Everything below is provider-agnostic in shape; the field names are where it differs.
Meter at the task, not the call
The instinct is to log tokens per model call. Do that and you get a number nobody can use, because one certificate request might be eleven calls: a classification, three document extractions, two endorsement lookups, a draft, and a repair pass after validation failed.
Attach every call to a task id instead. One task is one unit of work a human will eventually approve or reject. Everything the agent spends chasing that unit belongs to it, including the parts that failed.
import time, uuid, contextvars
current_task = contextvars.ContextVar("current_task")
class TaskMeter:
def __init__(self, task_id, kind, account_id=None):
self.task_id = task_id
self.kind = kind # coi_draft | remarket_packet | service_request
self.account_id = account_id
self.calls = []
self.started = time.time()
def record(self, model, usage, purpose):
self.calls.append({
"model": model,
"purpose": purpose, # classify | extract | endorsement_lookup | draft | repair
"input_tokens": usage.get("input_tokens", 0),
"output_tokens": usage.get("output_tokens", 0),
"cache_read_tokens": usage.get("cache_read_input_tokens", 0),
"cache_write_tokens": usage.get("cache_creation_input_tokens", 0),
"at": time.time(),
})
Wrap the model client once so no call site has to remember to meter.
def invoke(model, purpose, **kwargs):
resp = client.messages.create(model=model, **kwargs)
meter = current_task.get(None)
if meter is not None:
meter.record(model, dict(resp.usage), purpose)
return resp
If a call escapes the wrapper it is invisible, and invisible spend is the whole problem. We grep for direct client use in CI and fail the build.
Price it from a table you control
Do not hard-code rates in the agent. Put them in a config row per model, with an effective date, so last month's report still reconstructs last month's bill after a price change.
# USD per 1M tokens. Fill from your provider's current published rates.
PRICES = {
"model-a": {"input": 3.00, "output": 15.00, "cache_read": 0.30, "cache_write": 3.75},
"model-b": {"input": 0.80, "output": 4.00, "cache_read": 0.08, "cache_write": 1.00},
}
def cost_usd(call):
p = PRICES[call["model"]]
return (
call["input_tokens"] / 1e6 * p["input"]
+ call["output_tokens"] / 1e6 * p["output"]
+ call["cache_read_tokens"] / 1e6 * p["cache_read"]
+ call["cache_write_tokens"]/ 1e6 * p["cache_write"]
)
Two mistakes to avoid. Cached input reads are billed at a different rate from fresh input, so a meter that adds them into one bucket will under-report or over-report depending on the day. And OCR, storage, and portal fetches are real line items; if Textract runs on every submission page, add a per-page cost to the same task row or the report will quietly understate the work.
Write one row per task
At the end of the task, flatten the calls into a single row. Keep the per-purpose breakdown; it is what tells you which stage to fix.
def close(meter, outcome):
per_purpose = {}
total = 0.0
for c in meter.calls:
usd = cost_usd(c)
total += usd
per_purpose[c["purpose"]] = round(per_purpose.get(c["purpose"], 0.0) + usd, 6)
ddb.put_item(TableName=COST_TABLE, Item=marshal({
"task_id": meter.task_id,
"kind": meter.kind,
"account_id": meter.account_id,
"outcome": outcome, # drafted | abstained | failed
"calls": len(meter.calls),
"usd_total": round(total, 6),
"usd_by_purpose": per_purpose,
"seconds": round(time.time() - meter.started, 2),
"day": time.strftime("%Y-%m-%d"),
}))
return total
Record the outcome, not just the spend. An agent that abstains on 30% of certificate requests is doing the right thing, and those abstentions still cost money. Cost per approved draft is total spend divided by drafts a person actually approved, so abstentions and failures push it up. That is correct: they are part of what the workflow costs.
Cap it in three places
One cap is not enough, because the three failure modes are different.
Per task. A document that sends the agent into a repair loop should stop, not spend. Check the running total inside the loop and abstain when it crosses the ceiling.
class BudgetExceeded(Exception): pass
def guard(meter, ceiling_usd=0.75):
spent = sum(cost_usd(c) for c in meter.calls)
if spent > ceiling_usd:
raise BudgetExceeded(f"{meter.task_id} spent {spent:.2f}")
Catch BudgetExceeded where you catch a low-confidence match, and route it the same way: to a person, with the partial work and the reason attached. The item does not disappear, it queues.
Per run. A nightly batch has a known size. If the 120-day renewal window holds 380 policies and the per-task ceiling is $0.75, the run should never exceed roughly $285. Compute that before the first call and refuse to start if the queue is ten times the usual size, which is what a bad AMS query looks like.
Per day, per agency. A circuit breaker on a daily counter, checked at task entry. When it trips, the agent stops drafting and pages an engineer. It does not degrade to a cheaper model silently. Nobody wants to discover in a deposition that the certificate check ran on the budget tier.
def daily_gate(agency_id, limit_usd):
day = time.strftime("%Y-%m-%d")
r = ddb.update_item(
TableName=BUDGET_TABLE,
Key=marshal({"agency_id": agency_id, "day": day}),
UpdateExpression="ADD spent_usd :z SET updated = :t",
ExpressionAttributeValues=marshal({":z": 0, ":t": day}),
ReturnValues="ALL_NEW",
)["Attributes"]
if float(r["spent_usd"]["N"]) > limit_usd:
raise BudgetExceeded(f"{agency_id} daily cap {limit_usd}")
Then make it cheaper, in this order
Metering first, optimisation second. Once you have per-purpose numbers, the fixes are usually boring and large.
- Cache the static context. An endorsement-form library, the ACORD field map, and a 900-line system prompt are identical on every call. Prompt caching bills those tokens at a fraction of the input rate on reads. Order the prompt so everything stable sits above the first document byte, or nothing matches. Anthropic's prompt caching documentation and the equivalent Amazon Bedrock feature both key on an exact prefix, so a timestamp at the top of the system prompt is an expensive mistake.
- Route by judgment, not by vibes. Classifying an inbound message as a certificate request is not the same problem as reading additional insured wording against a policy. Send the classification to the small model, keep the coverage judgment on the strong one, and log which model made which call so the audit trail says so.
- Stop re-reading documents. Extraction output keyed by document hash is the single biggest win on a book with repeat holders. Same PDF, same fields, no second call.
- Trim retrieval. Sending twelve endorsement pages when three are relevant costs input tokens on every call in the chain.
- Cut the output, not the input. Structured output with a tight schema is cheaper than prose the next step has to parse, and output tokens are the expensive side.
Do not skip to step 2. Routing decisions made without a meter are guesses, and the usual guess sends the wrong stage to the cheap model.
What this does not tell you
- Cost per approved draft is not ROI. Staff hours saved is the number the principal cares about, and that comes out of the AMS activity log and timings, not the token meter.
- A cheaper run is not a better run. Track accuracy on the same case set whenever you change a model or a prompt, or you have traded quality for a number you can defend. The silent-run harness and the weekly canary are what keep that honest.
- Provider rates change and caching rules differ by provider and model. The table above is a shape, not a quote. Read your provider's current pricing page before you commit to a per-task ceiling.
- Budget caps are an operational control, not a safety control. They stop spend. They do not stop a wrong draft. The approval gate does that.
Where this fits
We build metering into every engagement from the first week, because an agency that owns its AWS account and its source code should be able to answer what a certificate costs without calling us. On a typical COI Agent or Renewal Radar build it is a day of work and it settles the pricing conversation for good.
If you are running an agent against Applied Epic, HawkSoft, or EZLynx and cannot say what one drafted item costs, tell us which workflow it is and a senior engineer will reply within one business day.