AI agents orchestrate complex workflows β calling LLMs, scraping pages, querying databases, sending emails. Each call costs real money. Without a governance layer, a single buggy loop can burn through your budget before anyone notices.
agent-gov is an open-source reverse proxy that intercepts every tool call your agents make, enforces budgets in real time, and auto-s out-of-control agents. Built as a FastAPI service with SQLite persistence, running 45 tests in 0.3 seconds.
This post walks through the architecture: the proxy pattern, the four-stage decision tree, cost tracking with a tool registry, multi-tenancy via workspaces, and the lazy auto-reset pattern.
Every AI agent tool call passes through agent-gov before reaching the actual tool. The agent sends a POST /proxy/call
with its API key, tool name, and estimated cost. agent-gov validates, budgets, and logs β then returns a 200 to approve or a 429 to reject.
class ToolCall(BaseModel):
agent_key: str = Field(...)
tool_name: str = Field(...)
estimated_cost: float = Field(0.0, ge=0)
The proxy doesn't execute the tool itself β it guards access. The agent only proceeds if the proxy returns 200. This is the gatekeeper pattern: a lightweight decision layer between the agent and the outside world.
Agent -> POST /proxy/call -> agent-gov -> 200/429 -> Agent decides
|
Calls actual tool
|
v
OpenAI / Browser / API
Why a proxy instead of a library? A library can be monkey-patched, removed, or forgotten. A proxy is a network boundary that agents must cross β it can't be bypassed.
Every proxy call runs through a four-stage pipeline:
@app.post("/proxy/call")
async def proxy_tool_call(call: ToolCall):
key_hash = db.hash_key(call.agent_key)
agent = await db.get_agent(key_hash)
if agent is None:
raise HTTPException(status_code=401, detail="Invalid API key")
if agent["d"]:
raise HTTPException(status_code=429,
detail=f"Agent '{agent['name']}' is d.")
agent = await db.check_and_reset_budget(agent)
registered_tool = await db.get_tool(call.tool_name)
actual_cost = (registered_tool["cost_per_call"]
if registered_tool else call.estimated_cost)
new_total = agent["spent_today"] + actual_cost
if new_total > agent["daily_budget"]:
await db._agent(key_hash)
raise HTTPException(status_code=429,
detail="Budget exceeded β agent auto-d.")
updated = await db.update_agent_spend(key_hash, actual_cost)
await db.log_cost_event(key_hash, agent["name"], call.tool_name, actual_cost)
return {"status": "approved", ...}
| Stage | Check | Exit |
|---|---|---|
| Auth | ||
| Does the API key hash match? | 401 β Invalid key | |
| Is the agent d? | 429 β Agent d | |
| Reset | ||
| New day since last call? | (silent) | |
| Budget | ||
| Would this exceed the daily cap? | 429 + auto- | |
| Log | ||
| INSERT cost event | 200 β Approved |
The trickiest design decision was cost determination. Trusting the agent's estimated_cost
is fragile β agents can under-report.
agent-gov uses a tool registry: an UPSERT-able table of known tools with real per-call costs.
registered_tool = await db.get_tool(call.tool_name)
actual_cost = (registered_tool["cost_per_call"]
if registered_tool else call.estimated_cost)
If the tool is registered, its true cost is used. The response includes a cost_source
field so clients know which path was taken.
The test proves an agent can't lie its way past governance: an agent with a $100 budget claiming a $1 estimate for a tool registered at $500/call gets blocked with 429.
v0.5 introduced workspaces β isolated tenants with their own agents, tools, and cost events. Each workspace gets a unique ID and API key. Every database row carries a workspace_id
FK column.
Schema migration uses PRAGMA table_info
to add columns only when missing β SQLite doesn't support IF NOT EXISTS
for ALTER TABLE
.
Tests verify workspace isolation: two workspaces, agents in each, neither can see the other's data.
Instead of a midnight cron job creating a thundering herd, agent-gov uses lazy evaluation: every proxy call checks if a reset is needed.
async def check_and_reset_budget(agent: dict) -> dict:
today = date.today().isoformat()
if agent["last_reset"] == today:
return agent
if agent["d"]:
return agent
return await reset_daily_budget(agent["key_hash"])
An agent that makes no calls doesn't need a reset. The thundering herd becomes a gentle trickle.
The next evolution: per-tool budget caps, webhook-based alerts, and a management API. But the foundation β a simple, testable, async governance proxy β is solid.
agent-gov is open source and MIT licensed. 45 tests. Zero database setup.