Skip to content

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