One of our clients, an industrial equipment distributor with about $85 million in revenue, kept its customer relationships in Salesforce and its invoices, payments and credit memos in NetSuite. Disputes were constant – partial shipments, pricing differences, credits an account manager had promised and nobody had processed – so the company put a Python-based AI worker on them. A customer disputed three invoices totaling $47,800, and the account manager's note in Salesforce read: "Customer says these charges should be credited. We agreed to review after reconciling the shipment discrepancies." The worker took that as approval, and tried to post a $47,800 credit memo, with the dispute marked resolved and the credit marked approved. The account manager had agreed to review the dispute. Nobody had approved a credit.
Nothing reached NetSuite, because the decision wasn't the model's to make. Code outside the model rejected the write on three counts: the worker couldn't approve its own credit memo, no finance-approved credit authorization existed, and the dispute was still under review. The finance operations manager found it in the daily review of rejected actions – four rejected credit attempts on the same dispute in eleven minutes.
When an AI agent only reads, a bad answer is an inconvenience. When it can create credit notes in an ERP or update accounts in a CRM, a bad answer turns into a record that finance has to find and reverse. Most of the guardrail advice I come across for this situation lives in the prompt – "never issue a credit above $500," "always ask before deleting." To be honest, I don't think a prompt is a control at all. It is a request. The model can be talked out of honoring it by a confused user, by its own reasoning, or by text someone planted in a document it was asked to read.
So here is the position I take, and the one this article implements. The model never touches a business system. It only produces proposals. A small, deterministic piece of code – I'll call it the tool gateway – decides what happens to each proposal. It can reject it, park it for a person, or execute it, and it records the outcome every time. Everything below works with any model. Whichever SDK you use, a tool call arrives as a tool name and a JSON object of arguments, and that is all the gateway needs.
Narrow, typed tools instead of "call the API"
The guardrail that does the most work is the choice of tools. A generic erp_request(method, path, body) tool hands the model the entire surface of your integration account, and no amount of checking further down the line will recover a sensible boundary from that. Each tool should correspond to one business operation the process actually needs, with arguments specific enough that validating them means something.
Here is a tool for an accounts payable exception worker. It may draft a credit note, but it may not post one. I use Pydantic v2, because its models do two jobs at once: they produce the JSON Schema you send to the model (DraftCreditNote.model_json_schema()), and they validate whatever comes back.
from dataclasses import dataclass
from decimal import Decimal
from enum import Enum
from typing import Annotated, Any, Callable, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator
class Tier(Enum):
AUTO = "auto" # execute and record
APPROVE = "approve" # record a proposal; a person releases it
class DraftCreditNote(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True)
invoice_id: Annotated[str, Field(pattern=r"^INV-\d{6,10}$")]
amount: Annotated[Decimal, Field(gt=0, max_digits=12, decimal_places=2)]
currency: Literal["USD", "EUR", "GBP"]
reason_code: Literal["PRICE_VARIANCE", "SHORT_SHIP", "DUPLICATE", "FREIGHT"]
memo: Annotated[str, Field(max_length=280)] = ""
@field_validator("amount")
@classmethod
def _cents(cls, v: Decimal) -> Decimal:
return v.quantize(Decimal("0.01")) # 120.5 and 120.50 must hash alike
@dataclass(frozen=True)
class ToolSpec:
name: str
args_model: type[BaseModel]
tier: Tier
auto_limit: Decimal | None # above this, a person must approve
hard_limit: Decimal | None # above this, refuse outright
max_calls_per_hour: int
check: Callable[[Any, Any], list[str]] # business rules
plan: Callable[[Any, Any], dict] # what would change (dry run)
execute: Callable[[Any, Any, str], dict] # third arg: idempotency key
shape: Callable[[dict], dict] # what the model gets back
extra="forbid" is easy to overlook, and it matters. Without it, suppose the model invents a post_immediately: true argument. Today it gets silently ignored. Then, one day, someone adds a field by that name to the model for some unrelated reason, and from then on it gets silently honored! The closed enums for reason codes do a similar job, keeping the model from inventing categories your reporting has never heard of.
Business rules belong to the system of record
Schema validation tells you the arguments are well formed. It can't tell you they are true. The model may say the invoice is open and the credit is within the balance – but the gateway should ask the ERP, and never take the model's word for it. These checks run before anything is recorded as a proposal.
def check_credit_note(args: DraftCreditNote, ctx) -> list[str]:
inv = ctx.erp.get_invoice(args.invoice_id) # read-only call, your client
if inv is None:
return [f"invoice {args.invoice_id} not found"]
problems = []
if inv.entity_id not in ctx.allowed_entities:
problems.append("invoice belongs to an entity outside this worker's scope")
if inv.status != "OPEN":
problems.append(f"invoice status is {inv.status}, expected OPEN")
if inv.currency != args.currency:
problems.append(f"currency {args.currency} does not match invoice {inv.currency}")
if args.amount > inv.open_balance:
problems.append(f"amount exceeds open balance {inv.open_balance}")
return problems
The function returns a list of reasons instead of raising an exception, and that is deliberate. The reasons go into the audit log, and they also go back to the model, which can often correct itself on the next turn (the wrong currency, say, or the wrong invoice). What the model can't do is argue the rule away. The rule isn't in its context window. It lives in code the model never sees.
The gateway
The gateway runs its checks in a fixed order. Does the tool exist? Do the arguments validate? Is this a retry of a call already recorded? Do the business rules pass? Is the amount under the hard ceiling? Only then does it decide the tier. A tool marked AUTO still goes to a person when the amount crosses its auto limit. I think that is how value thresholds should work – the same operation, with a different level of oversight depending on what's at stake.
import hashlib
from psycopg.types.json import Jsonb
from pydantic import ValidationError
class DefiniteFailure(Exception):
"""Raised by a tool when the target rejected the write; nothing changed."""
class Gateway:
def __init__(self, conn, tools: dict[str, ToolSpec], ctx):
self.conn, self.tools, self.ctx = conn, tools, ctx # psycopg 3, autocommit=True
def handle(self, run_id: str, name: str, raw_args: dict) -> dict:
spec = self.tools.get(name)
if spec is None:
return self._reject(run_id, name, raw_args, [f"unknown tool {name!r}"])
try:
args = spec.args_model.model_validate(raw_args)
except ValidationError as e:
msgs = [f"{'.'.join(map(str, err['loc']))}: {err['msg']}" for err in e.errors()]
return self._reject(run_id, name, raw_args, msgs)
key = hashlib.sha256(f"{run_id}|{name}|{args.model_dump_json()}".encode()).hexdigest()
if replay := self._replay(spec, key): # a retry gets the original outcome
return replay
problems = spec.check(args, self.ctx)
amount = getattr(args, "amount", None)
if amount is not None and spec.hard_limit is not None and amount > spec.hard_limit:
problems.append(f"amount exceeds hard limit {spec.hard_limit}")
if problems:
return self._reject(run_id, name, raw_args, problems)
tier = spec.tier
if amount is not None and spec.auto_limit is not None and amount > spec.auto_limit:
tier = Tier.APPROVE
return self._record(run_id, spec, args, spec.plan(args, self.ctx), key, tier)
def _replay(self, spec, key) -> dict | None:
prior = self.conn.execute(
"SELECT id, status, result FROM tool_calls WHERE idem_key = %s", (key,)
).fetchone()
if prior is None:
return None
return {"status": prior[1], "call_id": prior[0],
"result": spec.shape(prior[2] or {}), "replayed": True}
def _reject(self, run_id, name, raw_args, reasons) -> dict:
self.conn.execute(
"INSERT INTO tool_calls (run_id, tool, args, status, reasons) "
"VALUES (%s, %s, %s, 'rejected', %s)",
(run_id, name, Jsonb(raw_args), Jsonb(reasons)),
)
return {"status": "rejected", "reasons": reasons}
Rejected calls are logged with the raw arguments the model actually sent, and not a cleaned-up version. Later, when you're deciding whether the worker has earned more autonomy, the rejections will be some of the most useful rows you have.
At the distributor, this is where the real lesson turned up. A two-week sample held 186 rejected calls: 91 credits with no financial approval behind them, 38 credits larger than the approved adjustment, because the model used the invoice total, 29 duplicate attempts from the worker retrying rejected calls, 17 against the wrong customer or invoice, because account matching in the CRM was ambiguous, and 11 into a closed accounting period. The worker usually found the right customer and the right invoices. What it got wrong, over and over, was whether it was allowed to act – it read phrases such as "agreed to review", "will take care of" and "customer expects a credit" as formal approval. When half the rejections point the same way, you are looking at a systematic misreading, and the fix is a change in design.
We made three changes. First, we took the approval decision away from the model altogether. The tool now required a reference to an existing approval record, and the gateway looked that record up for itself, checked the approver's authority and checked the amount – the model never got to say whether something was approved. Second, we stopped retrying business-rule rejections. A rejected financial action went to a person, and only transient infrastructure failures were retried, under the idempotency key. Third, we restricted the integration user's permissions in NetSuite, so the ERP enforced the same limits even if the gateway had a bug.
Idempotency and rate limits in one transaction
Agents retry. The orchestration loop retries on timeouts, the model re-issues a call it believes failed, and a worker process restarts in the middle of a run. The key computed above is a hash of the run, the tool and the canonical validated arguments. (model_dump_json() serializes fields in the order they are declared, and the _cents validator gives each amount exactly one spelling – so 120.5 and 120.50 come out the same – which means the same arguments always produce the same string.) Within one run, an identical proposal is the same proposal. Treating it as a new one is precisely how you get the double credit note you were trying to prevent.
The gateway looks up that key before it runs the business rules, and the order is important. Say the first attempt's write landed and reduced the invoice's open balance. If the retry were checked against the rules again, it would now fail – and a model that's told "amount exceeds open balance" will helpfully lower the amount and try again. That is a new proposal, with a new key, and a second credit note. A retry should simply get back the answer the original call got.
The rate limit has to be counted in the same step as the insert. Otherwise two calls arriving at the same moment can both see 49 out of 50, and both go ahead. This version assumes PostgreSQL. A transaction-scoped advisory lock for each tool makes the count happen one call at a time, and the unique constraint on idem_key is the backstop.
def _record(self, run_id, spec, args, plan, key, tier) -> dict:
status = "executing" if tier is Tier.AUTO else "pending_approval"
with self.conn.transaction():
self.conn.execute("SELECT pg_advisory_xact_lock(hashtext(%s))", (spec.name,))
if replay := self._replay(spec, key): # a concurrent twin got here first
return replay
(recent,) = self.conn.execute(
"SELECT count(*) FROM tool_calls WHERE tool = %s "
"AND status <> 'rejected' AND created_at > now() - interval '1 hour'",
(spec.name,),
).fetchone()
if recent >= spec.max_calls_per_hour:
return self._reject(run_id, spec.name, args.model_dump(mode="json"),
["hourly limit reached for this tool"])
(call_id,) = self.conn.execute(
"INSERT INTO tool_calls (run_id, tool, args, idem_key, status, plan) "
"VALUES (%s, %s, %s, %s, %s, %s) RETURNING id",
(run_id, spec.name, Jsonb(args.model_dump(mode="json")), key, status, Jsonb(plan)),
).fetchone()
if tier is Tier.APPROVE:
return {"status": "pending_approval", "call_id": call_id, "plan": plan}
return self._run(call_id, spec, args, key)
Proposals waiting for approval count toward the hourly limit, and that is on purpose. A fault that floods the approval queue is still a fault, and the people doing the approving are a limited resource too.
The key also travels downstream. If the target API accepts an idempotency key, the tool's execute function passes it along. If it doesn't, use the closest thing the system offers – for example, writing the key to an external ID field and upserting on it (Salesforce supports upsert by external ID), or looking the key up before creating anything. A key kept only on the gateway's side can't protect you from the case that matters most: a write that succeeded on the far side just as your connection dropped.
Executing, and admitting what you don't know
That case gets its own status. When the target system rejects a write outright, nothing changed, and the call simply failed. When the request times out, you don't know what happened – and pretending you do is how duplicates get made.
def _run(self, call_id, spec, args, key) -> dict:
try:
result, status = spec.execute(args, self.ctx, key), "succeeded"
except DefiniteFailure as exc:
result, status = {"error": str(exc)}, "failed"
except Exception as exc: # timeout, reset: outcome unknown
result, status = {"error": type(exc).__name__}, "unknown"
self.conn.execute(
"UPDATE tool_calls SET status = %s, result = %s, updated_at = now() WHERE id = %s",
(status, Jsonb(result), call_id),
)
return {"status": status, "call_id": call_id, "result": spec.shape(result)}
Handlers have to return dicts that can be turned into JSON, so convert any Decimal to a string. An unknown row replays to the model as unknown, which stops it from retrying blindly, and a reconciliation job or a person then settles it against the target system.
Approval and dry runs
The plan column is the dry run. Every proposal stores what it would change, worked out by the tool's plan function from the current state of the system, and that is what an approver reads. In the first weeks of a worker's life, I set every write tool's tier to APPROVE, so the whole thing runs in shadow – real proposals and real plans, but nothing executed without a person. Comparing those proposals with what people actually decided is the evidence you need before loosening anything. The policy side of that – sitting down with finance or compliance to decide which actions an agent may take on its own – I've covered in a longer piece on designing guardrails and human checkpoints for AI agents.
When a proposal is approved, the gateway executes the arguments it stored. It must never ask the model to generate them again. It also has to run the business rules again, because the invoice may well have been paid in the hour the proposal sat in the queue.
def approve(self, call_id: int, approver: str) -> dict:
with self.conn.transaction():
row = self.conn.execute(
"SELECT tool, args, idem_key FROM tool_calls "
"WHERE id = %s AND status = 'pending_approval' FOR UPDATE", (call_id,)
).fetchone()
if row is None:
raise LookupError(f"call {call_id} is not awaiting approval")
spec = self.tools[row[0]]
args = spec.args_model.model_validate(row[1])
if problems := spec.check(args, self.ctx):
self.conn.execute(
"UPDATE tool_calls SET status = 'rejected', reasons = %s, "
"updated_at = now() WHERE id = %s", (Jsonb(problems), call_id))
return {"status": "rejected", "reasons": problems}
self.conn.execute(
"UPDATE tool_calls SET status = 'executing', approved_by = %s, "
"updated_at = now() WHERE id = %s", (approver, call_id))
return self._run(call_id, spec, args, row[2])
The FOR UPDATE, together with the status condition, means that if two approvers click at the same moment, the proposal still executes only once.
The audit table
One table carries the whole lifecycle of a call – what the model asked for, why it was refused, what it would have changed, who approved it, and what the target system sent back.
CREATE TABLE tool_calls (
id bigserial PRIMARY KEY,
run_id text NOT NULL,
tool text NOT NULL,
args jsonb NOT NULL,
idem_key text UNIQUE, -- NULL for rejections; NULLs don't collide
status text NOT NULL CHECK (status IN ('rejected', 'pending_approval',
'executing', 'succeeded', 'failed', 'unknown')),
reasons jsonb,
plan jsonb,
result jsonb,
approved_by text,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX ON tool_calls (tool, created_at);
CREATE INDEX ON tool_calls (run_id);
Rows in this table change, since a call's status moves forward over time. If your auditors need evidence that nothing was tampered with, also write each status change to an append-only events table, and take away UPDATE and DELETE on that table from the application's database role.
Treat what tools return as hostile
Prompt injection sits first on the OWASP Top 10 for LLM applications, and agents that can act are exactly where it hurts. The invoice PDF your worker reads could contain text telling it to issue a credit against a different invoice, and the model may well comply. The design above doesn't rely on the model resisting. Whatever it proposes still has to land inside an allowed entity, an open balance, an amount limit and a rate limit – and, above a threshold, in front of a human. That is the reason I put the controls outside the model in the first place.
A few more habits limit what an injection can reach. The shape function returns only the fields the process needs, so a CRM record with forty fields, including a free-text notes field, doesn't flow back into the model's context. Free text the agent writes, like memo, is capped in length, because another agent or another system may read it later. And no tool lets data read from a document decide where money or records go. An agent may match an invoice to a supplier, but it never sources bank details or changes master data – and the integration account it runs under shouldn't have permission to do either.
None of this makes an agent trustworthy on its own. What it does is keep the agent's mistakes small, visible and reversible. That is what lets you widen its autonomy one tool at a time, based on evidence, and not on hope.
I'm Jayant Chaudhary, founder of eProxim, where we build integrations, and increasingly the AI workers that act through them, for implementation partners and the firms they serve.
Top comments (0)