Run a company with AI: your first agent, with hard caps
A four-file Python agent that drafts overdue-invoice reminders, refuses to send anything without your yes, and halts on a step cap or a dollar cap. Full code.
What you'll have running
Four files, about 180 lines of Python, one sitting. It reads a CSV of open invoices, drafts one payment reminder for every account 30 or more days overdue, and cannot send a single one until you type yes. Around that loop sit a step cap, a dollar cap, and an append-only SQLite ledger with one row per model turn and one per tool call.
Prerequisites:
| Need | Detail |
|---|---|
| Python 3.11 or newer, plus the SDK | pip install openai — the official one, repo here |
| An OpenAI API key | platform.openai.com/api-keys. Any key on a paid account works |
sqlite3 on the command line | Only if you want to read the ledger by hand. The Python side uses the stdlib sqlite3 module, so nothing to install |
Pick a task you'd hand a competent contractor with written instructions and no phone call. Accounts receivable is the example here because the inputs are structured and the output is a draft, not a wire transfer.
Where the caps sit in the loop
An agent is a while loop around a model that can call functions. Everything that makes it safe to leave running lives outside the model, in the loop you control. The model can ask to send 400 emails. The router is what refuses.
Excessive agency — an agent holding more permission or autonomy than its task needs — is a named entry in the OWASP Top 10 for LLM Applications. Unbounded step loops are the version of it that shows up first in production. Which is why OpenAI's own Agents SDK ships a max_turns argument that raises MaxTurnsExceeded instead of leaving the limit to the prompt.
Writing "stop after five steps" in the system prompt is not a cap. It's a preference, and the model is free to hold a different one. We're building the hard version by hand so you can see where it lives.
The three supporting files
invoices.csv — sample data. The contact addresses use the .example TLD, reserved by RFC 2606 exactly so documentation can't mail a real person.
invoice_id,account,contact,amount_usd,due_date,status
INV-1041,Northwind Traders,ap@northwind.example,4200.00,2025-01-14,open
INV-1042,Contoso Ltd,billing@contoso.example,980.50,2025-02-03,open
INV-1043,Fabrikam Inc,ap@fabrikam.example,15750.00,2025-03-20,paid
ledger.py — the audit log. Insert-only. One row per model turn, one per tool call.
# ledger.py
import json, os, sqlite3, time
DB = os.environ.get("AGENT_DB", "agent.db")
SCHEMA = """
CREATE TABLE IF NOT EXISTS ledger (
id INTEGER PRIMARY KEY AUTOINCREMENT,
run_id TEXT NOT NULL,
ts REAL NOT NULL,
step INTEGER NOT NULL,
kind TEXT NOT NULL,
payload TEXT NOT NULL
);
"""
def init():
con = sqlite3.connect(DB)
con.executescript(SCHEMA)
con.commit()
con.close()
def write(run_id, step, kind, payload):
con = sqlite3.connect(DB)
con.execute(
"INSERT INTO ledger (run_id, ts, step, kind, payload) VALUES (?,?,?,?,?)",
(run_id, time.time(), step, kind, json.dumps(payload, default=str)),
)
con.commit()
con.close()
if __name__ == "__main__":
init()
print(f"ledger ready at {DB}")
tools.py — one read-only tool, one irreversible tool, and the schemas the model sees. The send tool writes to outbox/, so nothing leaves the machine while you're testing.
# tools.py
import csv, os
from datetime import date
INVOICES = os.environ.get("INVOICES_CSV", "invoices.csv")
APPROVALS_REQUIRED = {"send_dunning_email"}
def list_overdue_invoices(days_overdue: int = 1):
today = date.today()
out = []
with open(INVOICES) as f:
for row in csv.DictReader(f):
overdue = (today - date.fromisoformat(row["due_date"])).days
if row["status"] == "open" and overdue >= days_overdue:
out.append({
"invoice_id": row["invoice_id"],
"account": row["account"],
"contact": row["contact"],
"amount_usd": float(row["amount_usd"]),
"days_overdue": overdue,
})
return {"count": len(out), "invoices": out}
def send_dunning_email(contact: str, invoice_id: str, body: str):
os.makedirs("outbox", exist_ok=True)
path = f"outbox/{invoice_id}.txt"
with open(path, "w") as f:
f.write(f"To: {contact}\nSubject: Invoice {invoice_id}\n\n{body}\n")
return {"queued": path}
REGISTRY = {
"list_overdue_invoices": list_overdue_invoices,
"send_dunning_email": send_dunning_email,
}
SCHEMAS = [
{"type": "function", "function": {
"name": "list_overdue_invoices",
"description": "Return open invoices past their due date.",
"parameters": {
"type": "object",
"properties": {"days_overdue": {"type": "integer",
"description": "Minimum days past due."}},
"required": ["days_overdue"],
"additionalProperties": False,
}}},
{"type": "function", "function": {
"name": "send_dunning_email",
"description": "Queue one payment-reminder email. Irreversible; needs human approval.",
"parameters": {
"type": "object",
"properties": {
"contact": {"type": "string"},
"invoice_id": {"type": "string"},
"body": {"type": "string"},
},
"required": ["contact", "invoice_id", "body"],
"additionalProperties": False,
}}},
]
The schema format and the tool_calls round-trip are documented in OpenAI's function calling guide.
agent.py, the whole loop
Three things in here matter more than the prompt. for step in range(...) is the step cap, and the model has no way to extend it. spend accumulates real token counts from resp.usage against a rate table, and the loop refuses to start another turn once it crosses MAX_USD. APPROVALS_REQUIRED is checked in the router, before the function is ever called.
The prompt is the least load-bearing part of the file.
# agent.py
import json, os, sys, uuid
from openai import OpenAI
import ledger
from tools import REGISTRY, SCHEMAS, APPROVALS_REQUIRED
MODEL = os.environ.get("AGENT_MODEL", "gpt-4o-mini")
MAX_STEPS = int(os.environ.get("AGENT_MAX_STEPS", "8"))
MAX_USD = float(os.environ.get("AGENT_MAX_USD", "0.25"))
# USD per 1M tokens: (input, output). Check the pricing page before trusting these.
PRICES = {"gpt-4o-mini": (0.15, 0.60)}
SYSTEM = (
"You are the accounts-receivable agent for a small company. "
"Find invoices at least 30 days overdue, then call send_dunning_email once per "
"invoice with a short, polite reminder. Then stop and summarise what you queued. "
"Never invent an invoice, an amount or an address."
)
class Halt(Exception):
pass
def cost_usd(model, prompt_tokens, completion_tokens):
inp, outp = PRICES.get(model, (0.0, 0.0))
return (prompt_tokens * inp + completion_tokens * outp) / 1_000_000
def approve(run_id, step, name, args):
if os.environ.get("AGENT_UNATTENDED") == "1":
ledger.write(run_id, step, "pending_approval", {"name": name, "args": args})
return False
print(f"\nAPPROVAL NEEDED -> {name}")
print(json.dumps(args, indent=2))
return input("type 'yes' to run it: ").strip().lower() == "yes"
def main(goal):
ledger.init()
run_id = uuid.uuid4().hex[:12]
client = OpenAI()
spend = 0.0
messages = [{"role": "system", "content": SYSTEM},
{"role": "user", "content": goal}]
ledger.write(run_id, 0, "run_start",
{"model": MODEL, "goal": goal,
"max_steps": MAX_STEPS, "max_usd": MAX_USD})
for step in range(1, MAX_STEPS + 1):
if spend >= MAX_USD:
raise Halt(f"budget cap hit at ${spend:.4f}")
resp = client.chat.completions.create(
model=MODEL, messages=messages, tools=SCHEMAS, tool_choice="auto",
)
u = resp.usage
spend += cost_usd(MODEL, u.prompt_tokens, u.completion_tokens)
msg = resp.choices[0].message
messages.append(msg.model_dump(exclude_none=True))
ledger.write(run_id, step, "model_turn", {
"requested": [c.function.name for c in (msg.tool_calls or [])],
"prompt_tokens": u.prompt_tokens,
"completion_tokens": u.completion_tokens,
"spend_usd": round(spend, 6),
})
if not msg.tool_calls:
print(msg.content)
ledger.write(run_id, step, "run_end", {"spend_usd": round(spend, 6)})
print(f"\nrun {run_id} finished in {step} steps, ${spend:.4f}")
return
for call in msg.tool_calls:
name = call.function.name
args = json.loads(call.function.arguments or "{}")
if name not in REGISTRY:
result = {"error": f"unknown tool {name}"}
elif name in APPROVALS_REQUIRED and not approve(run_id, step, name, args):
result = {"error": "declined; no action taken"}
else:
try:
result = REGISTRY[name](**args)
except Exception as e:
result = {"error": f"{type(e).__name__}: {e}"}
ledger.write(run_id, step, "tool_call",
{"name": name, "args": args, "result": result})
messages.append({"role": "tool", "tool_call_id": call.id,
"content": json.dumps(result, default=str)})
raise Halt(f"step cap of {MAX_STEPS} reached without finishing")
if __name__ == "__main__":
try:
main(" ".join(sys.argv[1:]) or "Chase everything 30+ days overdue.")
except Halt as e:
print(f"HALTED: {e}", file=sys.stderr)
sys.exit(2)
Run it
- Create the project and install the SDK:
mkdir ar-agent && cd ar-agent && python3 -m venv .venv && source .venv/bin/activate && pip install openai. The last line of pip output names the version, e.g.Successfully installed openai-1.x.x. - Set your key from platform.openai.com/api-keys:
export OPENAI_API_KEY="sk-...". Confirm the shell has it withecho ${OPENAI_API_KEY:0:3}— it printssk-and nothing else. - Save the four files into the directory.
lsprints exactlyagent.py invoices.csv ledger.py tools.py. - Create the ledger:
python ledger.py. Printsledger ready at agent.db. - Test the read tool with no model in the way:
python -c "import json,tools; print(json.dumps(tools.list_overdue_invoices(30), indent=2))". You get"count": 2and the INV-1041 and INV-1042 rows. INV-1043 is absent because its status ispaid. - Run the agent:
python agent.py "Chase everything 30+ days overdue."You'll see anAPPROVAL NEEDED -> send_dunning_emailblock with the JSON arguments, twice. Typeyesto both. The run ends with the model's summary and thenrun <12-hex-id> finished in N steps, $0.00xx. - Check what actually got written:
ls outboxprintsINV-1041.txt INV-1042.txt, andcat outbox/INV-1041.txtshows aTo: ap@northwind.exampleheader. Nothing was sent — that's the point of the stub. - Read the audit trail:
sqlite3 agent.db "select step, kind, substr(payload,1,60) from ledger order by id;". Row 1 isrun_start, then alternatingmodel_turn/tool_callrows, ending inrun_end. - Prove the step cap is real:
AGENT_MAX_STEPS=1 python agent.py, thenecho $?. Output isHALTED: step cap of 1 reached without finishingon stderr and exit code2. Nothing new appears inoutbox/. - Prove the budget cap is real:
AGENT_MAX_USD=0.00001 python agent.py. The first turn runs, the second refuses:HALTED: budget cap hit at $0.0002(your figure will differ with token count).
When it doesn't work
No key in the environment. The client raises on construction:
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
You exported it in a different shell, or in a shell that has since closed, or you're running under cron — which does not inherit your interactive environment. Fix: export in the same shell, or set the key in the cron line.
A model name your key can't reach. The SDK raises openai.NotFoundError with a 404 whose message says the model does not exist or you do not have access to it. Swap AGENT_MODEL for a model your account can see.
Then do the second half of the fix: add that model to PRICES. If you don't, PRICES.get(model, (0.0, 0.0)) returns zeros, spend stays at 0.0 forever, and the budget cap silently never fires. A cap that can't trigger is worse than no cap, because you'll believe it.
Tool results appended in the wrong order. A 400 that reads like:
openai.BadRequestError: Error code: 400 - Invalid parameter: messages with role
'tool' must be a response to a preceeding message with 'tool_calls'.
The assistant message carrying tool_calls must go into messages before any role: "tool" message, and you need one tool message per tool_call_id — all of them, even the ones you declined. That's why the declined branch still appends {"error": "declined; no action taken"} instead of skipping.
If you get a 429 with You exceeded your current quota, that's billing, not code.
What a run costs, and where the number comes from
The run prints its own spend because resp.usage gives you prompt_tokens and completion_tokens on every call. The rate table is the only thing you maintain by hand.
| Input | Where it comes from | Trap |
|---|---|---|
prompt_tokens, completion_tokens | resp.usage on each API response | Cached input bills at a different rate on some models; the table is a floor, not an invoice |
| USD per 1M tokens | OpenAI's pricing page — gpt-4o-mini is listed at $0.15 input / $0.60 output per 1M tokens | Rates change; a stale PRICES entry makes your cap wrong in a direction you won't notice |
MAX_USD | You | Checked between turns, so one very long turn can overshoot it |
The arithmetic is unremarkable. At $0.15 per 1M input tokens, a run that consumes 40,000 input tokens costs $0.006 on the input side. Which is why nobody optimises single runs. The number that matters is that figure times how often the thing fires, and that's a scheduling decision, not a prompt decision.
Running it unattended, without waking up to 400 sent emails
Set AGENT_UNATTENDED=1 and approve() stops asking. It writes the proposed action to the ledger as pending_approval, returns False, and the model is told the action was declined. The agent still does all the reading, all the reasoning and all the drafting; a human still authorises every irreversible step, just asynchronously.
The alternative people reach for — let it send, email yourself a summary — isn't approval. It's notification, after the fact, about mail already in a customer's inbox.
A weekly run, wrapped in flock so a slow run can never overlap the next one:
0 9 * * 1 cd /srv/ar-agent && AGENT_UNATTENDED=1 AGENT_MAX_STEPS=8 AGENT_MAX_USD=0.25 OPENAI_API_KEY=sk-... flock -n .lock .venv/bin/python agent.py >> run.log 2>&1
Monday morning, read the queue:
sqlite3 agent.db "select run_id, payload from ledger where kind='pending_approval';"
Two overlapping runs sharing an outbox/ and a ledger is a genuinely annoying class of bug, and flock -n costs you one word to prevent.
What I would not hand to an agent yet
I took Metadata.io from $0 to $15M ARR with humans doing the work, raised $50M along the way, and hold 6 patents in AI-driven marketing. I'm now building the next thing as a zero-human company, so I'm not arguing agents can't run real operations. I'm arguing about which lever you give them.
I would not delegate spend authority. Not the ad budget, not vendor payments, not anything that moves money out. The reason isn't that models are dumb — it's that spend is the one action where a wrong loop compounds instead of just failing. Drafting is safe because the worst case is a bad draft you delete. Sending is one approval away from a bad draft in a customer's inbox with your name on it.
I would not let an agent write to a system of record with no reversal path. Draft into outbox/, into a Slack channel, into a pending_approval table — some staging surface a human clears — until you've watched a few hundred of its proposals and know what its bad ones look like.
And I would not judge an agent by its demo run. Judge it by the ledger after a month: how many steps per task, how many declined approvals, how many HALTED lines in run.log. If you want the longer version of where these systems break in marketing specifically, that's here, and the two-year account of building this way is here.
The agent you just built has two tools. One is read-only, and the other one can't fire without you.
Sources
- OpenAI — Function calling guide — Tool schema shape, tool_calls response, and the required order of assistant/tool messages
- openai-python (official SDK repo) — Client usage, OPENAI_API_KEY default, error classes such as NotFoundError and RateLimitError
- Python docs — sqlite3 — SQLite ships in the standard library, so the ledger needs no extra dependency
- RFC 2606 — Reserved Top Level DNS Names — Why the sample invoices use .example addresses
- flock(1) man page — Preventing overlapping scheduled runs
- OWASP Top 10 for Large Language Model Applications — Excessive agency as a named, documented risk category for agents
- OpenAI Agents SDK — Running agents — max_turns and MaxTurnsExceeded as vendor precedent for an externally enforced step cap
- OpenAI — API keys — Where the reader gets OPENAI_API_KEY
- OpenAI — API pricing — Per-million-token rates used in the PRICES table and the cost arithmetic