From a hand-rolled approval flow
You check "does this need approval?" inside the tool, park the arguments in a dict or a
table, and run the tool once someone flips a flag. That code usually has two bugs: a
replay runs the action again, and the approved call can run with other arguments.
flowchart LR
subgraph Before
B1["tool checks a flag"] --> B2["APPROVED ticket?"]
B2 -- yes --> B3["runs, with any arguments,<br/>every time"]
end
subgraph After
A1["confirm= rule"] --> A2["DeferredApproval + store"]
A2 --> A3["runs once, with the<br/>reviewed arguments"]
end
Mapping
| Before |
After |
if amount > 100 and not approved: inside the tool |
confirm=lambda call, ctx: ... on the tool |
a PENDING dict or table |
an ApprovalStore (SQLiteApprovalStore, RedisApprovalStore, SQLAlchemyApprovalStore, DynamoDBApprovalStore; Database stores) |
APPROVED.add(ticket) |
store.decide(approval_id, Decision(...)) |
| re-calling the tool after approval |
re-running the same call (same run id, same call id) |
| "needs approval" as a return value |
ToolResult(status="pending"), error.details["approval_id"] |
| editing the stored arguments |
Decision(edited_args=...), validated again |
Before
PENDING: dict[str, dict[str, Any]] = {} # ticket id -> arguments
APPROVED: set[str] = set()
def refund_before(ticket: str, order_id: str, amount: float) -> str:
if amount > 100 and ticket not in APPROVED:
PENDING[ticket] = {"order_id": order_id, "amount": amount}
return "needs approval"
refunds.append(amount) # runs again on every replay, with whatever amount it is given
return f"refunded {amount}"
print(refund_before("t1", "o-1", 500))
APPROVED.add("t1")
print(refund_before("t1", "o-1", 500), refund_before("t1", "o-1", 5000)) # replay, bigger
After
registry = Registry()
@registry.tool(confirm=lambda call, ctx: float(call.arguments["amount"]) > 100)
def refund(order_id: str, amount: float) -> str:
"""Refund an order."""
refunds.append(amount)
return f"refunded {amount}"
store = MemoryApprovalStore() # SQLiteApprovalStore / RedisApprovalStore in production
executor = Executor(registry, approval=DeferredApproval(store))
ctx = executor.context(run_id="run_t1")
call = ToolCall(name="refund", arguments={"order_id": "o-1", "amount": 500}, id="c1")
first = await executor.run_call(call, ctx)
approval_id = first.error.details["approval_id"] if first.error else ""
await store.decide(approval_id, Decision(approved=True, decided_by="ana"))
print(first.status, (await executor.run_call(call, ctx)).value)
print((await executor.run_call(call, ctx)).status) # replay: a new review, no refund
bigger = ToolCall(name="refund", arguments={"order_id": "o-1", "amount": 5000}, id="c1")
swapped = await executor.run_call(bigger, ctx)
print(swapped.error.kind if swapped.error else "", refunds) # refused, one refund
Output
needs approval
refunded 500 refunded 5000
pending refunded 500.0
pending
tool-usage-error [500.0]
The old flow paid twice, the second time ten times more. The new one paid once.
Gained and lost
| Gained |
Lost |
| one decision runs one call, once |
the approval check inside the tool body (move it to confirm=) |
| the decision is bound to the run, call id, tool and arguments |
free-form tickets: keep yours in ApprovalRequest.metadata or your own table, keyed by approval id |
| the same flow for frameworks, MCP and the HTTP server |
|
Checklist
- [ ] Remove the approval check from the tool body; add
confirm=.
- [ ] Use a durable store if more than one process runs.
- [ ] Store
run_id and the call id with the approval id, so the resume matches.
- [ ] Reviewers decide through your UI, the HTTP server or a chat app.
- [ ] Delete the
PENDING and APPROVED tables once nothing reads them.
Files
| File |
Does |
examples/guide_migrate_approval.py |
the old flow and its bugs, then the new one |
tests/docs/test_migration.py |
runs it offline and checks the output |