Start an AI startup by shipping one metered agent endpoint
Build a POST /jobs endpoint that runs a step-capped agent loop, logs tokens and dollar cost per job, and returns HTTP 402 when a caller passes their spend cap.
The thing you will have running
A POST /jobs endpoint on localhost:8000 that takes a URL and a question, runs a Claude tool-use loop with a hard ceiling of 6 model calls, fetches only hosts on an allowlist, and returns the answer alongside the input tokens, output tokens and dollar cost of that one job. Cross the month-to-date cap you set and it returns HTTP 402 and spends nothing. One sitting of typing. The test calls here are a few thousand tokens each, so at Anthropic's listed $3 per million input tokens, ten of them cost cents.
You need Python 3.10 or newer (python3 --version), an Anthropic API key from console.anthropic.com/settings/keys, curl, and four packages: fastapi, uvicorn, anthropic, httpx. No Docker, no cloud account, no database server. The ledger is a SQLite file on disk.
Why this and not a landing page. I took Metadata.io from $0 to $15M ARR and raised $50M doing it, and the thing I would build first if I were starting today is not the demo. It is the meter. An AI product has a variable cost per request that a SaaS product did not, and you cannot price it, cap it, or tell which customer is losing you money until every call writes a row with tokens and dollars in it. The endpoint below is the smallest complete version of that: one billable unit of work, metered, capped, refusable. I have written elsewhere about what actually changed for founders; this is the code version of it.
app.py, complete
Save this as app.py. Nothing is elided.
# app.py -- a metered, capped, single-job AI service.
# Run: uvicorn app:app --port 8000
import hashlib
import os
import re
import sqlite3
from datetime import datetime, timezone
import httpx
from anthropic import Anthropic
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
MODEL = os.environ.get("MODEL", "claude-sonnet-4-5")
IN_PER_MTOK = float(os.environ.get("IN_PER_MTOK", "3.00")) # check anthropic.com/pricing
OUT_PER_MTOK = float(os.environ.get("OUT_PER_MTOK", "15.00"))
STEP_CAP = int(os.environ.get("STEP_CAP", "6"))
MONTHLY_CAP_USD = float(os.environ.get("MONTHLY_CAP_USD", "5.00"))
ALLOWED_HOSTS = {"example.com", "www.example.com", "docs.python.org"}
DB = os.environ.get("LEDGER_DB", "ledger.db")
client = Anthropic() # reads ANTHROPIC_API_KEY from the environment
app = FastAPI()
def db():
con = sqlite3.connect(DB)
con.execute(
"""CREATE TABLE IF NOT EXISTS jobs(
id INTEGER PRIMARY KEY, key_id TEXT, month TEXT, created_at TEXT,
steps INT, input_tokens INT, output_tokens INT,
cost_usd REAL, status TEXT)"""
)
return con
def cost(inp, out):
return inp / 1e6 * IN_PER_MTOK + out / 1e6 * OUT_PER_MTOK
def month_spend(key_id, month):
with db() as con:
row = con.execute(
"SELECT COALESCE(SUM(cost_usd), 0) FROM jobs WHERE key_id=? AND month=?",
(key_id, month),
).fetchone()
return row[0]
TOOLS = [
{
"name": "fetch_url",
"description": "Fetch the visible text of one web page. Allowlisted hosts only.",
"input_schema": {
"type": "object",
"properties": {"url": {"type": "string"}},
"required": ["url"],
},
}
]
def fetch_url(url):
host = httpx.URL(url).host
if host not in ALLOWED_HOSTS:
return f"BLOCKED: {host} is not on the allowlist. Do not retry it."
r = httpx.get(url, timeout=15, follow_redirects=True)
text = re.sub(r"<(script|style)[\s\S]*?</\1>", " ", r.text, flags=re.I)
text = re.sub(r"<[^>]+>", " ", text)
return re.sub(r"\s+", " ", text)[:8000]
class Job(BaseModel):
url: str
question: str
@app.post("/jobs")
def create_job(job: Job, x_api_key: str = Header(...)):
key_id = hashlib.sha256(x_api_key.encode()).hexdigest()[:16]
month = datetime.now(timezone.utc).strftime("%Y-%m")
spent = month_spend(key_id, month)
if spent >= MONTHLY_CAP_USD:
raise HTTPException(
402, f"cap reached: ${spent:.6f} of ${MONTHLY_CAP_USD:.6f}"
)
messages = [
{
"role": "user",
"content": (
f"Question: {job.question}\nSource: {job.url}\n"
"Fetch the page with fetch_url, then answer in under 60 words. "
"If the fetch is blocked, say so and stop."
),
}
]
in_tok = out_tok = 0
steps = 0
answer = None
status = "step_cap_hit"
for steps in range(1, STEP_CAP + 1):
resp = client.messages.create(
model=MODEL, max_tokens=800, tools=TOOLS, messages=messages
)
in_tok += resp.usage.input_tokens
out_tok += resp.usage.output_tokens
messages.append({"role": "assistant", "content": resp.content})
if resp.stop_reason != "tool_use":
answer = "".join(b.text for b in resp.content if b.type == "text")
status = "ok"
break
results = []
for b in resp.content:
if b.type == "tool_use":
out = (
fetch_url(b.input.get("url", ""))
if b.name == "fetch_url"
else f"unknown tool: {b.name}"
)
results.append(
{"type": "tool_result", "tool_use_id": b.id, "content": out}
)
messages.append({"role": "user", "content": results})
c = cost(in_tok, out_tok)
with db() as con:
con.execute(
"""INSERT INTO jobs(key_id, month, created_at, steps,
input_tokens, output_tokens, cost_usd, status)
VALUES(?,?,?,?,?,?,?,?)""",
(
key_id,
month,
datetime.now(timezone.utc).isoformat(),
steps,
in_tok,
out_tok,
c,
status,
),
)
return {
"status": status,
"steps": steps,
"answer": answer,
"input_tokens": in_tok,
"output_tokens": out_tok,
"cost_usd": round(c, 6),
"month_spend_usd": round(spent + c, 6),
"month_cap_usd": MONTHLY_CAP_USD,
}
@app.get("/usage")
def usage(x_api_key: str = Header(...)):
key_id = hashlib.sha256(x_api_key.encode()).hexdigest()[:16]
month = datetime.now(timezone.utc).strftime("%Y-%m")
with db() as con:
n, total, avg = con.execute(
"""SELECT COUNT(*), COALESCE(SUM(cost_usd), 0), COALESCE(AVG(cost_usd), 0)
FROM jobs WHERE key_id=? AND month=?""",
(key_id, month),
).fetchone()
return {
"month": month,
"jobs": n,
"spend_usd": round(total, 6),
"avg_cost_per_job_usd": round(avg, 6),
}
Three details matter here. The loop reads resp.stop_reason — the Messages API returns tool_use when the model wants a tool and something else when it is done, so the loop exits on the model's own signal rather than on string matching. Token counts come from resp.usage on every response, which makes the meter the provider's number instead of an estimate. And the raw API key is never stored: key_id is the first 16 hex characters of its SHA-256, enough to group usage per customer and useless to anyone who steals the database.
Build it
- Make the project and a virtualenv:
mkdir jobs-api && cd jobs-api && python3 -m venv .venv && source .venv/bin/activate. Your prompt now starts with(.venv). - Install:
pip install fastapi uvicorn anthropic httpx. The last line readsSuccessfully installed anthropic-… fastapi-… httpx-… uvicorn-…with version numbers. - Create a key at console.anthropic.com/settings/keys, then
export ANTHROPIC_API_KEY=sk-ant-…. Verify it is in the same shell you will run the server from:python3 -c "import os; print(os.environ['ANTHROPIC_API_KEY'][:7])"printssk-ant-. - Save
app.pyfrom the block above into the project directory. - Confirm the model ID exists on your account:
python3 -c "from anthropic import Anthropic; print([m.id for m in Anthropic().models.list().data][:6])". You get a Python list of model ID strings. Ifclaude-sonnet-4-5is not among them, pick one that is andexport MODEL=…. - Start the server:
uvicorn app:app --port 8000. The last startup line isINFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit). - In a second terminal, run one job:
curl -s localhost:8000/jobs -H "x-api-key: tenant_alpha" -H "content-type: application/json" -d '{"url":"https://example.com","question":"What is this domain reserved for?"}'. You get JSON with"status":"ok"and"steps":2— one model call to request the tool, one to answer — plus non-zeroinput_tokens,output_tokens, and acost_usdin the thousandths of a dollar. Theanswerfield paraphrases example.com's text about being reserved for use in documentation. - Check that the row landed:
curl -s localhost:8000/usage -H "x-api-key: tenant_alpha". You get{"month":"YYYY-MM","jobs":1,"spend_usd":…,"avg_cost_per_job_usd":…}with the same figure as the job'scost_usd. - Test the allowlist with a host you have not allowed: repeat the job call with
"url":"https://intranet.example.test". The tool returnsBLOCKED: intranet.example.test is not on the allowlist.and the answer says the page could not be fetched.cost_usdis still non-zero — a refused tool call is not a free call. - Test the spend cap. Stop uvicorn, then
export MONTHLY_CAP_USD=0.000001and start it again. The next job call returns{"detail":"cap reached: $0.00… of $0.000001"}with HTTP 402, and no row is written because no model call was made. Add-ito the curl to seeHTTP/1.1 402 Payment Required. - Prove per-customer isolation: run a job with
-H "x-api-key: tenant_beta"while the cap is low. Also 402 for beta? No — the cap is perkey_id, so setMONTHLY_CAP_USDback to5.00, restart, call astenant_beta, then compare/usagefor both keys. Each shows only its own jobs and spend.
Where the two caps sit
Both caps are enforced in Python, before and around the model call. Neither is asked for in the prompt, and that is the whole point. Tell a model to stop after five tool calls and it will do six; a for loop bounded by range(1, STEP_CAP + 1) cannot. Unbounded loops are how an agent turns into a bill: the tool returns an error the model does not understand, it retries with a small variation, and the retry costs the same as the first attempt. The Do not retry it. sentence in the blocked-host message is a nudge. The loop bound is the control.
Note the asymmetry. The 402 branch writes nothing, because nothing was spent. Every other path — success, blocked host, step cap hit — writes a row, because every other path cost money. A meter that only records successes reports an average cost per job that is wrong in the direction that flatters you. The longer version of this loop pattern, including dry-run mode, is in build an AI agent loop with a step cap, allowlist and dry run.
Reading the ledger
Run ten or twenty jobs against different questions, then read /usage. Four numbers there decide things a dashboard of signups cannot.
| Field | What it decides |
|---|---|
avg_cost_per_job_usd | Your price floor. Whatever you charge per job has to clear this plus support, infrastructure and the jobs you refund. |
steps distribution | Whether the loop wanders. Jobs landing on steps = 6 with status = "step_cap_hit" are prompt or tool failures billed at full price. |
input_tokens vs output_tokens | Where optimisation pays. Input-heavy jobs mean you are stuffing context; output-heavy jobs mean you are generating more than the customer reads. Anthropic lists output tokens at 5× the input rate for Sonnet, so the two are not interchangeable. |
spend_usd per key_id | Which customer is unprofitable. One heavy account can eat the margin from twenty light ones, and a flat monthly price hides it until you look. |
The arithmetic is the boring part and it is the part people skip. Take your measured avg_cost_per_job_usd, multiply by the number of jobs a customer runs in a month, compare with what you planned to charge. A customer on a $99/month plan running 400 jobs needs your cost per job to stay under about $0.25 before you have spent a cent on servers — and that is 0% margin. Now you can check instead of guess.
Two things I would add next, in this order. First, a per-key cap the customer sets themselves, so the 402 is a product feature and not an outage. Second, real metered billing — Stripe's billing docs cover usage-based subscriptions — reading from the same jobs table, so the number you invoice and the number you measured are the same number. A related walk-through of the pricing side is in ship one metered, capped endpoint.
When it doesn't work
Uvicorn exits during import. The traceback ends with Could not resolve authentication method. Expected either api_key or auth_token to be set. The Anthropic() constructor runs at import time and found no ANTHROPIC_API_KEY. This is almost always two terminals: you exported the key in one and ran uvicorn in the other. Export it in the shell that runs the server, or load it from a .env file you never commit.
The API returns 401. You get anthropic.AuthenticationError: Error code: 401 with a body whose error.type is authentication_error. The key exists but is wrong — usually a copy that picked up a trailing newline, or a key from a different organisation than the one with credit on it. Print len() of the environment variable before you blame the code.
The API returns 404 with not_found_error and your model string. Model IDs and aliases change; Anthropic's models overview page is the source of truth and the API will not guess for you. Run the models.list() one-liner from step 5, pick an ID that comes back, and export MODEL= that. This is why MODEL is an environment variable in the code rather than a literal in the call.
422 instead of a job. The response is {"detail":[{"type":"missing","loc":["header","x-api-key"],"msg":"Field required","input":null}]}. FastAPI turns the parameter name x_api_key into the header x-api-key, and the header is required. You either dropped the -H flag or sent x_api_key with an underscore.
status is step_cap_hit and answer is null. The loop spent all 6 model calls and never produced a final text block. The usual cause is a tool whose output the model cannot act on — a blocked host, an empty page, a 403 from the target site — so it tries again with a small variation. Log the tool results, fix what the tool returns, leave the cap alone. Raising STEP_CAP to make the symptom go away just multiplies the cost of every failing job.
What I would not wire up yet
I would not give this agent spend authority. Reading a page and answering a question is a job you can cap at a few cents and refund without a phone call. Buying ads, sending email to a list, or writing to a production database is not, and the control surface for those is approval, not a step cap. I hold 6 patents in AI-driven marketing and I am building a zero-human company in public, and I still keep the irreversible actions behind a human until the reversible ones have run clean for a long time.
I would also not add a second tool before the first one's cost per job is stable. Every tool you add widens the loop, and a wider loop is more paths to step_cap_hit at full price. One tool, one job type, one meter, one price — then the next tool.
And I would not treat the model as the product. Your competitor can call the same claude-sonnet-4-5 endpoint tomorrow at the same $3 per million input tokens. What they cannot call is your ledger: the per-customer record of what the work actually cost you and what people paid for it. That asymmetry is the thing worth compounding, and it is the argument in an AI moat is what survives when your rival uses your model.
Sources
- Anthropic Messages API reference — tools parameter, stop_reason values including tool_use, and usage.input_tokens / usage.output_tokens on every response
- Anthropic models overview — model IDs and aliases, and that they change over time — reason for checking models.list() rather than trusting a hardcoded string
- Anthropic pricing — per-million-token input and output prices used in the cost() function
- Anthropic API errors — 401 authentication_error and 404 not_found_error response shapes quoted in the failure section
- Anthropic Console — API keys — where the reader creates the ANTHROPIC_API_KEY used in this tutorial
- FastAPI — Header parameters — underscore-to-hyphen header conversion for x_api_key and the 422 validation error when a required header is missing
- Python sqlite3 documentation — Connection used as a context manager commits the transaction on exit
- RFC 2606 — Reserved Top Level DNS Names — example.com and .test are reserved for documentation, which is why the sample data uses them
- Stripe Billing documentation — metered billing as the step after you have a per-job cost ledger