Gil Allouche
← All writing
TutorialRunning a company with AI

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.

September 4, 2026·Gil Allouche·11 min read
A build-along. Every step was run start to finish; commands and outputs are what the machine actually printed.

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:

NeedDetail
Python 3.11 or newer, plus the SDKpip install openai — the official one, repo here
An OpenAI API keyplatform.openai.com/api-keys. Any key on a paid account works
sqlite3 on the command lineOnly 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.

Agent loop showing where the step cap, budget cap, approval gate and ledger sitGoalModel turnStep + $ capsTool routerApproval gateAppend-only ledgerHALT, exit 2next step

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

  1. 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.
  2. Set your key from platform.openai.com/api-keys: export OPENAI_API_KEY="sk-...". Confirm the shell has it with echo ${OPENAI_API_KEY:0:3} — it prints sk- and nothing else.
  3. Save the four files into the directory. ls prints exactly agent.py invoices.csv ledger.py tools.py.
  4. Create the ledger: python ledger.py. Prints ledger ready at agent.db.
  5. 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": 2 and the INV-1041 and INV-1042 rows. INV-1043 is absent because its status is paid.
  6. Run the agent: python agent.py "Chase everything 30+ days overdue." You'll see an APPROVAL NEEDED -> send_dunning_email block with the JSON arguments, twice. Type yes to both. The run ends with the model's summary and then run <12-hex-id> finished in N steps, $0.00xx.
  7. Check what actually got written: ls outbox prints INV-1041.txt INV-1042.txt, and cat outbox/INV-1041.txt shows a To: ap@northwind.example header. Nothing was sent — that's the point of the stub.
  8. Read the audit trail: sqlite3 agent.db "select step, kind, substr(payload,1,60) from ledger order by id;". Row 1 is run_start, then alternating model_turn / tool_call rows, ending in run_end.
  9. Prove the step cap is real: AGENT_MAX_STEPS=1 python agent.py, then echo $?. Output is HALTED: step cap of 1 reached without finishing on stderr and exit code 2. Nothing new appears in outbox/.
  10. 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.

InputWhere it comes fromTrap
prompt_tokens, completion_tokensresp.usage on each API responseCached input bills at a different rate on some models; the table is a floor, not an invoice
USD per 1M tokensOpenAI's pricing page — gpt-4o-mini is listed at $0.15 input / $0.60 output per 1M tokensRates change; a stale PRICES entry makes your cap wrong in a direction you won't notice
MAX_USDYouChecked 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

  1. OpenAI — Function calling guideTool schema shape, tool_calls response, and the required order of assistant/tool messages
  2. openai-python (official SDK repo)Client usage, OPENAI_API_KEY default, error classes such as NotFoundError and RateLimitError
  3. Python docs — sqlite3SQLite ships in the standard library, so the ledger needs no extra dependency
  4. RFC 2606 — Reserved Top Level DNS NamesWhy the sample invoices use .example addresses
  5. flock(1) man pagePreventing overlapping scheduled runs
  6. OWASP Top 10 for Large Language Model ApplicationsExcessive agency as a named, documented risk category for agents
  7. OpenAI Agents SDK — Running agentsmax_turns and MaxTurnsExceeded as vendor precedent for an externally enforced step cap
  8. OpenAI — API keysWhere the reader gets OPENAI_API_KEY
  9. OpenAI — API pricingPer-million-token rates used in the PRICES table and the cost arithmetic

Related