Start an AI company by building the meter first

What you will have running

A FastAPI service on your laptop: text in, structured JSON out, each call priced from the token counts the API hands back, one line appended to ledger.jsonl, and HTTP 402 once a customer's month-to-date spend crosses a cap you set. An hour, if Python 3.11 and a funded OpenAI account are already on the machine.

That is a company's spine, not a demo. A demo proves the model can do the task. The meter tells you whether the task costs less than the price, per customer, on the day that stops being true.

You need Python 3.11 or newer and curl, plus an API key from platform.openai.com/api-keys on an account with credit on it — an unfunded key fails at the model call with insufficient_quota, after your cap check has already passed. Step 7 also wants a Stripe test-mode secret key and a billing meter. Nothing else.

The build in seven ticks

1. 1
   Create the project and install: python3 -m venv .venv && source .venv/bin/activate then pip install "fastapi[standard]" openai. pip show fastapi prints a Version line.
2. 2
   Export the key: export OPENAI_API_KEY=sk-.... Check with python -c "import os;print(bool(os.environ.get('OPENAI_API_KEY')))" — expect True.
3. 3
   Save the service below as app.py. It is complete; nothing to fill in except the price constants and the cap.
4. 4
   Start it: uvicorn app:app --reload. Expect INFO: Uvicorn running on http://127.0.0.1:8000.
5. 5
   Make a billed call with curl and read the cost_usd field in the response. A new line appears in ledger.jsonl.
6. 6
   Force the cap: append a fake $99 row to the ledger, call again, and confirm HTTP 402 with no new ledger row.
7. 7
   Optional: run meter.py to replay the ledger into Stripe as meter events.

Step 1–3: the service, complete

Two decisions are baked in. Cost comes from usage.prompt_tokens and usage.completion_tokens on the completion object, never from a client-side token estimate — estimate it locally and your ledger and your invoice disagree, and the invoice is the number that gets charged. Second, the cap is enforced in Python before the model call. A system prompt that says "stay under budget" is not a control; it has no idea what the last 400 calls cost, and it is the thing you are trying to constrain.

Unbounded work is the most common way agentic systems burn money in production. The limit has to live outside the model doing the work. Same principle as the step cap in three tools, a step cap, a JSONL note.

# app.py — a metered AI endpoint with a hard monthly spend cap per customer.
# Python 3.11+.  pip install "fastapi[standard]" openai
import json
import time
from pathlib import Path
from threading import Lock

from fastapi import FastAPI, Header, HTTPException
from openai import OpenAI
from pydantic import BaseModel

MODEL = "gpt-4o-mini"
PRICE_IN_PER_1M = 0.15      # USD per 1M input tokens — check openai.com/api/pricing
PRICE_OUT_PER_1M = 0.60     # USD per 1M output tokens
MONTHLY_CAP_USD = 5.00      # per API key, per calendar month (UTC)
MAX_OUTPUT_TOKENS = 400
LEDGER = Path("ledger.jsonl")

# Your customer table. In production this is a database row, not a dict.
KEYS = {
    "sk_demo_alice": {"account": "alice@example.com", "stripe_customer_id": "cus_00000000alice"},
    "sk_demo_bob": {"account": "bob@example.test", "stripe_customer_id": "cus_00000000bob"},
}

client = OpenAI()  # reads OPENAI_API_KEY from the environment
app = FastAPI(title="extract-api")
write_lock = Lock()


class ExtractIn(BaseModel):
    text: str


def month_key() -> str:
    return time.strftime("%Y-%m", time.gmtime())


def spent_this_month(api_key: str) -> float:
    if not LEDGER.exists():
        return 0.0
    total, this_month = 0.0, month_key()
    with LEDGER.open() as fh:
        for line in fh:
            row = json.loads(line)
            if row["api_key"] == api_key and row["month"] == this_month:
                total += row["cost_usd"]
    return round(total, 6)


def call_cost(input_tokens: int, output_tokens: int) -> float:
    return (input_tokens / 1_000_000) * PRICE_IN_PER_1M + (
        output_tokens / 1_000_000
    ) * PRICE_OUT_PER_1M


def append_ledger(row: dict) -> None:
    with write_lock, LEDGER.open("a") as fh:
        fh.write(json.dumps(row) + "\n")


@app.post("/v1/extract")
def extract(body: ExtractIn, x_api_key: str = Header()):
    account = KEYS.get(x_api_key)
    if account is None:
        raise HTTPException(status_code=401, detail="unknown api key")

    spent = spent_this_month(x_api_key)
    if spent >= MONTHLY_CAP_USD:
        raise HTTPException(
            status_code=402,
            detail=f"monthly cap reached: spent ${spent:.4f} of ${MONTHLY_CAP_USD:.2f}",
        )

    completion = client.chat.completions.create(
        model=MODEL,
        max_completion_tokens=MAX_OUTPUT_TOKENS,
        response_format={"type": "json_object"},
        messages=[
            {
                "role": "system",
                "content": (
                    "Extract fields from the user's text. Reply with only a JSON object "
                    "with the keys company, person, intent. Use null when a field is absent."
                ),
            },
            {"role": "user", "content": body.text[:4000]},
        ],
    )

    usage = completion.usage
    cost = call_cost(usage.prompt_tokens, usage.completion_tokens)
    raw = completion.choices[0].message.content
    try:
        fields = json.loads(raw)
    except json.JSONDecodeError:
        fields = {"error": "model did not return json", "raw": raw}

    append_ledger(
        {
            "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "month": month_key(),
            "api_key": x_api_key,
            "account": account["account"],
            "model": MODEL,
            "input_tokens": usage.prompt_tokens,
            "output_tokens": usage.completion_tokens,
            "cost_usd": round(cost, 6),
        }
    )

    return {
        "fields": fields,
        "usage": {
            "input_tokens": usage.prompt_tokens,
            "output_tokens": usage.completion_tokens,
        },
        "cost_usd": round(cost, 6),
        "month_spend_usd": round(spent + cost, 6),
        "cap_usd": MONTHLY_CAP_USD,
    }


@app.get("/v1/usage")
def usage(x_api_key: str = Header()):
    if x_api_key not in KEYS:
        raise HTTPException(status_code=401, detail="unknown api key")
    spent = spent_this_month(x_api_key)
    return {
        "month": month_key(),
        "spend_usd": spent,
        "cap_usd": MONTHLY_CAP_USD,
        "remaining_usd": round(max(0.0, MONTHLY_CAP_USD - spent), 6),
    }

FastAPI maps the parameter name x_api_key to the HTTP header x-api-key, so callers send x-api-key. The two demo keys are hard-coded on purpose. A dict you can read beats an auth service you cannot debug on day one.

Step 4–5: run it and make one billed call

uvicorn app:app --reload

You should see, on the last line:

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

In a second terminal:

curl -s localhost:8000/v1/extract \
  -H "content-type: application/json" \
  -H "x-api-key: sk_demo_alice" \
  -d '{"text":"Hi, Dana Okoye here at Northwind Traders (dana@northwind.example). We need contact fields pulled out of about 40k inbound emails a month."}'

The response has this shape. Token counts move with the model and the prompt, so yours will differ in the last digits:

{"fields":{"company":"Northwind Traders","person":"Dana Okoye","intent":"bulk contact field extraction"},
 "usage":{"input_tokens":98,"output_tokens":29},
 "cost_usd":2.2e-05,"month_spend_usd":2.2e-05,"cap_usd":5.0}

Then check the ledger:

cat ledger.jsonl

One JSON object per line, ending in a cost_usd value. That file is the only thing in this build that is yours. The model is rented.

curl -s localhost:8000/v1/usage -H "x-api-key: sk_demo_alice" returns spend_usd, cap_usd and remaining_usd for the current UTC month.

Step 6: prove the cap refuses work

Caps that have never fired are not caps. Push the ledger over the line with a synthetic $99 row, then call again:

python - <<'PY'
import json, time
now = time.gmtime()
row = {"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", now), "month": time.strftime("%Y-%m", now),
       "api_key": "sk_demo_alice", "account": "alice@example.com", "model": "gpt-4o-mini",
       "input_tokens": 0, "output_tokens": 0, "cost_usd": 99.0}
open("ledger.jsonl", "a").write(json.dumps(row) + "\n")
PY

curl -s -o /dev/null -w "%{http_code}\n" localhost:8000/v1/extract \
  -H "content-type: application/json" -H "x-api-key: sk_demo_alice" \
  -d '{"text":"second call, should be refused"}'

You should see 402. Drop the -o /dev/null -w flags and the body reads {"detail":"monthly cap reached: spent $99.0000 of $5.00"}, with cents reflecting whatever your first call cost. 402 Payment Required is defined but left unassigned in RFC 9110 §15.5.3, which is exactly why it is a clean signal here: no intermediary retries it the way a 429 or a 503 invites, and no client library treats it as transient.

Two things to verify. wc -l ledger.jsonl did not grow, and the uvicorn terminal shows no new outbound request. The model was never called, so the refusal is free — which is the whole point of checking before the call rather than after. Delete ledger.jsonl to reset.

Three limits to fix before this carries real customers

The check and the write are not atomic. The handler reads the month total, makes the call, then appends the row. Two requests in flight can both pass a check at $4.99 and both bill. The overshoot is bounded by concurrency multiplied by the cost of a single call. For this handler the worst case per in-flight request is known: the body.text[:4000] truncation caps input at roughly 1,000 tokens (English averages about four characters per token), which is $0.00015, plus MAX_OUTPUT_TOKENS of 400 at the $0.60 output rate, which is $0.00024 — $0.00039 in total. Ten concurrent requests can therefore overshoot the cap by well under half a cent, but that ceiling scales with whatever per-request limits you set. The fix is to reserve before you spend: write a row for that worst-case figure, then correct it downward once usage comes back. A counter incremented inside a database transaction does the same job with less bookkeeping.

spent_this_month rescans the whole file every request. At a few thousand rows that is invisible. Each row here serialises to roughly 250 bytes, so a month with a million billed calls means parsing about 250 MB of JSON on every single request, and it gets slower every day of the month. Hold a running total per key per month in memory, rebuild it from the file at startup, and leave the file as the audit trail rather than the query path.

Your ledger is a derived number, not the invoice. It is arithmetic over the usage fields the provider returned per call; the bill is computed on the provider's side, and reflects calls this endpoint never saw. Prompt caching is one concrete source of drift: a repeated prefix that hits the cache is reported separately under usage.prompt_tokens_details.cached_tokens and billed at a discount — for gpt-4o-mini, $0.075 per 1M rather than $0.150 — while call_cost above charges the full input rate for those tokens. Compare your month-to-date total against the provider's usage dashboard weekly. In practice the gaps come from traffic made outside the metered path — one-off scripts, retries in a neighbouring service — far more often than from a mistake in the rate constants.

What a call actually costs, and what that means for a price

Cost per call is arithmetic on published rates. Take 1,200 input tokens and 300 output tokens at the gpt-4o-mini rates in the code:

At $0.00036 a call, the $5.00 cap in app.py clears at about 13,900 calls. The customer in the curl example above, with 40,000 emails a month, would cost $14.40 and trip that cap around day 11 of a 30-day month — which is the conversation the meter exists to start.

Check the rates on OpenAI's pricing page before you put a price in front of a customer. They move, and your margin moves with them.

Note what the table does not contain: retries, a second pass when the JSON comes back malformed, embeddings, a vector store, your own server. Model tokens are the smallest line on that bill. More on the wider cost shape in what AI can actually do for a business, and what it costs.

The reason to meter on day one is that a price per 1,000 extractions is an argument you can have with a buyer. "AI-powered enrichment" is not. At Metadata.io the pricing conversations that closed were the ones where the unit was legible.

Step 7: push usage into Stripe

Stripe's usage-based billing takes meter events, each carrying a customer ID and a value. Create the billing meter in the dashboard first, event name extract_calls — events for an unknown meter name are rejected outright. The identifier field makes the send idempotent, so re-running this script does not double-bill.

# meter.py — replay the ledger into Stripe as meter events, one per call.
# pip install stripe ; export STRIPE_SECRET_KEY=sk_test_...
import json
import os
from pathlib import Path

import stripe

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

CUSTOMERS = {
    "sk_demo_alice": "cus_00000000alice",
    "sk_demo_bob": "cus_00000000bob",
}

for line in Path("ledger.jsonl").read_text().splitlines():
    row = json.loads(line)
    customer = CUSTOMERS.get(row["api_key"])
    if not customer:
        continue
    event = stripe.billing.MeterEvent.create(
        event_name="extract_calls",
        payload={"stripe_customer_id": customer, "value": "1"},
        identifier=f'{row["ts"]}-{row["api_key"]}',
    )
    print(event.identifier, "accepted")

Run python meter.py. One accepted line per ledger row whose API key is in CUSTOMERS, and the event count on that meter rises in the dashboard. Note that the identifier here is timestamp plus key, and the ledger's ts has one-second resolution — two calls from the same key inside the same second collapse into one event, so use the ledger line number or a UUID per row if your traffic is above one call per key per second. Swap cus_00000000alice for real test-mode customer IDs first; the placeholders return No such customer.

When it doesn't work

No key in the environment. Starting uvicorn raises on import of the client:

openai.OpenAIError: The api_key client option must be set either by passing api_key
to the client or by setting the OPENAI_API_KEY environment variable

Fix: export OPENAI_API_KEY=sk-... in the same shell that runs uvicorn. A key exported in the other terminal does not count.

Header name wrong. Sending x_api_key, or omitting the header entirely, gets a FastAPI validation error, not a 401:

{"detail":[{"type":"missing","loc":["header","x-api-key"],"msg":"Field required"}]}

Fix: send -H "x-api-key: sk_demo_alice". Hyphen, not underscore.

Quota or rate limit. A 429 surfaces as openai.RateLimitError, either Rate limit reached for ... or You exceeded your current quota, please check your plan and billing details with "type": "insufficient_quota". Those are different problems. The first needs backoff; the second needs money on the account. Read the x-ratelimit-remaining-requests and x-ratelimit-remaining-tokens response headers to tell them apart, and x-ratelimit-reset-requests for how long to wait, per the rate limits guide. A quota failure sets none of them. This code does not retry, deliberately: adding retries without a budget check per attempt is how a cap gets bypassed. Count every attempt against the ledger, not every success.

Malformed JSON from the model. response_format={"type": "json_object"} makes this rare, not impossible, especially if a low max_completion_tokens truncates the output mid-object — check completion.choices[0].finish_reason for length to distinguish truncation from a genuinely bad response. The handler catches json.JSONDecodeError and returns {"error": "model did not return json", "raw": ...} while still billing the call, because the tokens were spent either way. Decide deliberately whether you charge the customer for it. I would not, and I would log it as a defect rather than an edge case.

What I would not hand to an agent in this loop yet

An agent can write this service. An agent can call it, read ledger.jsonl, and tell you which customer is unprofitable this month. Raising a cap, issuing a refund and changing a price are a different class of permission, and in production deployments they are routinely kept on the human side of the line even where agents otherwise run unattended.

Those are the three actions where a wrong decision is not a bug you patch. It is a customer conversation. The cap stays a constant a human edits.

What the agent should own is the boring half: reading the ledger daily, flagging any key above 80% of cap — $4.00 of the $5.00 set here — opening a PR when PRICE_IN_PER_1M drifts from the vendor's published rate. Agents on the read path, humans on the money path. That is the part of the zero-human company that has held so far.

And if you are still choosing what to build rather than how, the finding in the Harvard experiment on AI and entrepreneurship is that the gains showed up in execution speed, not in picking the idea. This endpoint is execution. Point it at a task someone has already paid a human to do badly.