Start an AI company by shipping one capped, metered endpoint

What you'll have running

A POST /run endpoint that will not spend more than $0.25 or make more than 6 model calls on a single request. It takes an inbound sales email, extracts a structured lead with Claude, and writes both the lead and the token cost of that run into SQLite. Budget an hour of typing if Python 3.10+ is already on the machine.

That endpoint is the smallest honest version of an AI company: one unit of work, priced, capped, logged. Everything else is downstream of it — landing page, billing, a second workflow.

Prerequisites: Python 3.10 or newer (python3 --version), curl, and an Anthropic API key from console.anthropic.com/settings/keys with credit on the account. A dollar of credit covers dozens of test runs at the prices below. Nothing to install for storage — SQLite ships with Python.

The whole service, one file

The complete main.py, which runs as written once ANTHROPIC_API_KEY is set. The two price constants are Claude Sonnet 4.5 list price at the time of writing — $3 per million input tokens, $15 per million output (Anthropic pricing). Re-check them before you charge anybody.

# main.py — metered agent endpoint with hard caps
import json
import sqlite3
import time

from anthropic import Anthropic
from fastapi import FastAPI
from pydantic import BaseModel

MODEL = "claude-sonnet-4-5"

# USD per million tokens. Verify at https://www.anthropic.com/pricing
PRICE_IN = 3.00
PRICE_OUT = 15.00

MAX_STEPS = 6        # model calls per run, enforced in Python, not in the prompt
MAX_RUN_USD = 0.25   # one run may not cost more than this
DB_PATH = "runs.db"

client = Anthropic()  # reads ANTHROPIC_API_KEY from the environment
app = FastAPI()

SYSTEM = (
    "You triage inbound sales email. Extract exactly one lead and call "
    "record_lead once. Never invent an email address that is not in the text. "
    "If the message is not a sales enquiry, reply with the single word SKIP."
)

TOOLS = [
    {
        "name": "record_lead",
        "description": "Save one lead to the database. Call at most once per email.",
        "input_schema": {
            "type": "object",
            "properties": {
                "company": {"type": "string"},
                "contact_email": {"type": "string"},
                "ask": {"type": "string", "description": "What they want, under 15 words"},
                "urgency": {"type": "string", "enum": ["now", "this_quarter", "unknown"]},
            },
            "required": ["company", "contact_email", "ask", "urgency"],
        },
    }
]


def conn():
    c = sqlite3.connect(DB_PATH, timeout=10)
    c.execute("PRAGMA journal_mode=WAL")
    c.execute(
        """CREATE TABLE IF NOT EXISTS runs (
             idem TEXT PRIMARY KEY, created REAL, status TEXT, steps INTEGER,
             in_tok INTEGER, out_tok INTEGER, usd REAL, output TEXT)"""
    )
    c.execute(
        """CREATE TABLE IF NOT EXISTS leads (
             id INTEGER PRIMARY KEY AUTOINCREMENT, idem TEXT, company TEXT,
             contact_email TEXT, ask TEXT, urgency TEXT)"""
    )
    return c


def cost(in_tok: int, out_tok: int) -> float:
    return round(in_tok / 1e6 * PRICE_IN + out_tok / 1e6 * PRICE_OUT, 6)


def record_lead(idem: str, **lead) -> dict:
    c = conn()
    with c:
        c.execute(
            "INSERT INTO leads (idem, company, contact_email, ask, urgency)"
            " VALUES (?,?,?,?,?)",
            (idem, lead["company"], lead["contact_email"],
             lead["ask"], lead["urgency"]),
        )
    c.close()
    return {"saved": True, **lead}


def run_agent(idem: str, text: str) -> dict:
    messages = [{"role": "user", "content": text}]
    in_tok = out_tok = 0
    saved = None

    for step in range(1, MAX_STEPS + 1):
        r = client.messages.create(
            model=MODEL,
            max_tokens=1024,
            system=SYSTEM,
            tools=TOOLS,
            messages=messages,
        )
        in_tok += r.usage.input_tokens
        out_tok += r.usage.output_tokens
        spent = cost(in_tok, out_tok)

        if spent > MAX_RUN_USD:
            return {"status": "aborted_cost", "steps": step, "in_tok": in_tok,
                    "out_tok": out_tok, "usd": spent, "output": saved}

        if r.stop_reason != "tool_use":
            said = "".join(b.text for b in r.content if b.type == "text").strip()
            return {"status": "done", "steps": step, "in_tok": in_tok,
                    "out_tok": out_tok, "usd": spent, "output": saved or said}

        messages.append({"role": "assistant", "content": r.content})
        results = []
        for block in r.content:
            if block.type == "tool_use" and block.name == "record_lead":
                saved = record_lead(idem, **block.input)
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": json.dumps(saved)})
            elif block.type == "tool_use":
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": "unknown tool", "is_error": True})
        messages.append({"role": "user", "content": results})

    return {"status": "aborted_steps", "steps": MAX_STEPS, "in_tok": in_tok,
            "out_tok": out_tok, "usd": cost(in_tok, out_tok), "output": saved}


class RunIn(BaseModel):
    text: str
    idempotency_key: str


@app.post("/run")
def run(body: RunIn):
    c = conn()
    row = c.execute(
        "SELECT status, steps, in_tok, out_tok, usd, output FROM runs WHERE idem=?",
        (body.idempotency_key,),
    ).fetchone()
    if row:
        c.close()
        return {"replayed": True, "status": row[0], "steps": row[1],
                "in_tok": row[2], "out_tok": row[3], "usd": row[4],
                "output": json.loads(row[5]) if row[5] else None}

    result = run_agent(body.idempotency_key, body.text)
    with c:
        c.execute(
            "INSERT INTO runs (idem, created, status, steps, in_tok, out_tok, usd,"
            " output) VALUES (?,?,?,?,?,?,?,?)",
            (body.idempotency_key, time.time(), result["status"], result["steps"],
             result["in_tok"], result["out_tok"], result["usd"],
             json.dumps(result["output"])),
        )
    c.close()
    return {"replayed": False, **result}


@app.get("/ledger")
def ledger():
    c = conn()
    rows = c.execute(
        "SELECT status, COUNT(*), SUM(usd), AVG(usd), MAX(usd)"
        " FROM runs GROUP BY status"
    ).fetchall()
    c.close()
    return [
        {"status": s, "runs": n, "usd_total": round(t or 0, 4),
         "usd_mean": round(a or 0, 4), "usd_max": round(m or 0, 4)}
        for s, n, t, a, m in rows
    ]

Two things in there are the actual product. MAX_STEPS is a Python for loop, not a line in the system prompt — a limit you write into the prompt is a request, and a model can negotiate with a request. It cannot negotiate with a range. And cost() runs on usage.input_tokens and usage.output_tokens, which the Messages API returns on every response (API reference), so every run leaves behind a number you can bill against.

Why the gate runs after the call, not before

Token usage is only known once a response comes back, so the cap is checked against cumulative spend after each call rather than ahead of it. The worst-case overshoot is one call, and max_tokens=1024 is what bounds that call: 1,024 output tokens at $15 per million is about $0.0154, so six steps can add at most roughly $0.092 of output cost no matter how verbose the model gets. Input is the side that grows without a ceiling, because each step appends the previous assistant turn and its tool results to messages — step six re-sends everything from steps one through five. That growth, not long replies, is what pushes a run toward the cap, so the input a caller controls (a 400-line forwarded thread with quoted history) is the variable to watch.

Build it

1. 1
   Make a project and a virtualenv: mkdir lead-agent && cd lead-agent && python3 -m venv .venv && source .venv/bin/activate. Your prompt now starts with (.venv).
2. 2
   Install three packages: pip install anthropic fastapi "uvicorn[standard]". The last line of output starts with Successfully installed and includes anthropic- and fastapi- with version numbers.
3. 3
   Get a key at console.anthropic.com/settings/keys and export it in the shell you will run the server from: export ANTHROPIC_API_KEY=sk-ant-.... Check it took: python3 -c "import os;print(os.environ['ANTHROPIC_API_KEY'][:7])" prints sk-ant-.
4. 4
   Save the code block above as main.py, then confirm it parses: python3 -c "import ast;ast.parse(open('main.py').read())". Success prints nothing at all.
5. 5
   Start the server: uvicorn main:app --port 8000. You should see INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit).
6. 6
   In a second terminal, send one email through it: curl -s localhost:8000/run -H 'content-type: application/json' -d '{"idempotency_key":"email-0001","text":"From: dana@northwind.example\nSubject: pricing\n\nHi - we are 40 people at Northwind Labs and our RevOps lead saw your demo. We need lead routing live before the end of the quarter. Can you send pricing? - Dana"}' | python3 -m json.tool
7. 7
   Read the response. status is done, steps is 2, and output contains "company": "Northwind Labs" and "contact_email": "dana@northwind.example". in_tok, out_tok and usd are filled in from the API's usage numbers — yours will differ from mine, because token counts vary run to run.
8. 8
   Fire the exact same command again, same idempotency_key. Now "replayed": true and usd is unchanged: the retry cost you nothing and did not write a second lead.
9. 9
   Prove the money cap fires. Edit MAX_RUN_USD = 0.000001, restart uvicorn, and re-run step 6 with "idempotency_key":"email-0002". You get "status": "aborted_cost", "steps": 1, "output": null — the gate tripped after the first model call, before any tool ran. Set it back to 0.25.
10. 10
    Read your ledger: curl -s localhost:8000/ledger | python3 -m json.tool. You get one row per status, e.g. done with runs: 1 and aborted_cost with runs: 1, each with usd_total, usd_mean and usd_max.

A done run on that sample email looks like this:

{
    "replayed": false,
    "status": "done",
    "steps": 2,
    "in_tok": 1042,
    "out_tok": 118,
    "usd": 0.004896,
    "output": {
        "saved": true,
        "company": "Northwind Labs",
        "contact_email": "dana@northwind.example",
        "ask": "pricing and lead routing before quarter end",
        "urgency": "this_quarter"
    }
}

steps: 2 is the loop working as designed: call one returns a tool_use block, the tool writes the row, call two returns text and the loop exits on stop_reason not being tool_use.

A second cap: spend per day

The per-run cap bounds one request. It does nothing about a caller who sends two thousand of them overnight. The ledger already has what you need to close that, because created is a timestamp on every row:

DAY_USD = 5.00


def spent_today(c) -> float:
    row = c.execute(
        "SELECT SUM(usd) FROM runs WHERE created > ?", (time.time() - 86400,)
    ).fetchone()
    return row[0] or 0.0

Then, in /run, after the replay lookup and before run_agent — so retries of already-paid-for work still succeed when the budget is gone:

    if spent_today(c) > DAY_USD:
        c.close()
        raise HTTPException(429, "daily budget exhausted")

That needs from fastapi import FastAPI, HTTPException at the top. Two caps in sequence — one per unit of work, one per day — give you the two numbers you can actually promise a customer: what a single run can cost you, and what a bad day can cost you.

When it doesn't work

The key. Start the server without exporting the variable and the SDK refuses at client construction with anthropic.AnthropicError: The api_key client option must be set either by passing api_key to the client or by setting the ANTHROPIC_API_KEY environment variable. A key with a stray character or newline gets you anthropic.AuthenticationError: Error code: 401 - {'type': 'error', 'error': {'type': 'authentication_error', 'message': 'invalid x-api-key'}} instead. An account with no funds returns a 400 whose message says the credit balance is too low. Fix in order: export the key in the same shell that runs uvicorn, re-copy it from the console, add credit.

The model ID. Aliases change as models ship. A stale one produces anthropic.NotFoundError: Error code: 404 - {'type': 'error', 'error': {'type': 'not_found_error', 'message': 'model: claude-sonnet-4-5'}}. Don't guess, and don't copy an ID out of a blog post. Ask the API: python3 -c "from anthropic import Anthropic; print([m.id for m in Anthropic().models.list().data])", then cross-check the models overview.

Every run comes back aborted_steps. The cap is not the bug here. It's the alarm. Unbounded tool loops are the most common failure mode in agent code, and they come from a broken result contract: an assistant turn that was never appended to messages, or a tool_use block that got no tool_result with a matching tool_use_id. The model, seeing no answer, asks again. And again. The Messages API requires every tool_use in an assistant turn to be answered by a tool_result carrying that tool_use_id in the next user turn. Print r.stop_reason and the block IDs for one run and you will see which side is missing.

A row that saved cleanly with the wrong email. The input_schema guarantees shape, not truth: required and the urgency enum mean the arguments parse, and say nothing about whether that address appeared in the message. The cheap guard is to pass text into record_lead and refuse the insert when lead["contact_email"] is not a substring of it, returning {"saved": False, "error": "email not present in source"} as the tool_result with "is_error": True. The model gets the rejection inside the same run and can correct itself on the next step, still under the step budget. Validating tool arguments in the tool, rather than trusting the schema, is the general pattern — the schema constrains the caller's syntax, your code is the only thing that constrains the caller's facts.

One more you will meet as soon as two callers hit the endpoint at once: sqlite3.OperationalError: database is locked. WAL mode lets readers and one writer coexist, but only one writer — that's SQLite's design, not a misconfiguration. Fine for a demo and for your first customer. Move to Postgres before you take concurrent traffic, and note the related gap: two simultaneous requests with the same idempotency_key can both miss the SELECT, and one will fail the primary-key insert. Catch sqlite3.IntegrityError there and re-read the row.

The numbers this gives you on day one

Every row in runs is a priced unit of work. That table, not the model, is what makes it a business.

The arithmetic, at $3 and $15 per million tokens: a run using 4,000 input and 600 output tokens costs 0.004 × 3 + 0.0006 × 15 = $0.021. A thousand of those is $21 of model spend a month. That's the floor, not the cost — add the aborted runs you can't invoice, the retries, and the person who reads the exception queue.

Two of those columns earn their keep as regression tests rather than as accounting. usd_max from /ledger is the number to watch after any prompt or schema change: if the worst run in your sample doubles, you changed the economics, and you'll see it before a customer does. steps is the same signal one level down — a mean creeping from 2 toward 4 means the model is taking extra turns to satisfy a tool it now finds ambiguous, which is a schema problem you can fix for free.

If you're weighing where this thing should live once it has real traffic, I wrote up what you actually control on each option in self-hosted AI agent platforms.

What I would not hand to it yet

I took Metadata.io from $0 to $15M ARR, raised $50M along the way, and I'm now building a zero-human company where agents do the work and I pay the bills for them. Here is where I still stop this service short.

No spend authority. Extracting a lead is a cheap, reversible write; issuing a refund or buying media is not, and the gap between those two is the whole risk surface. Nothing goes to a customer unreviewed until I have a few hundred done rows I've read by hand. And the database user gets INSERT and SELECT only — an agent that can DELETE will eventually find a reason to.

The reversibility test is the one worth keeping as you add tools. A row in leads can be corrected by a human in a second, at no cost, with no third party involved; anything that leaves the building — an email sent, a card charged, a calendar invite accepted — cannot. Caps and step limits are the right control for the first category. The second needs a different control entirely: a queue a person clears, or a tool that drafts rather than sends.

The cap in that file is about twelve lines of Python. It is also the only part of the system a model cannot argue with, which is why I'd write it before the landing page. If you want the same caps applied to a broader first agent, that's your first agent, with hard caps; if you're still deciding whether you're selling software or selling work, the two senses of "AI SaaS" is the argument to settle first.