Here is a deliberately small implementation. The point is not the code style; it is that every failure path returns without touching the ledger. A call that arrives without a valid token is rejected, not downgraded, not partially executed, not queued for later.
from dataclasses import dataclass
from decimal import Decimal
@dataclass(frozen=True)
class Confirmation:
token_id: str
txn_id: str
amount_minor: int
currency: str
payee_id: str
subject_id: str # customer or reviewer identity
channel: str
class ConfirmationError(Exception):
"""Raised when the gate refuses execution. No ledger writes occur."""
def execute_payment(txn_id: str, payee_id: str, amount_minor: int,
currency: str, token: str | None) -> str:
# 1. Gather the exact facts the token must match.
facts = look_up_transaction(txn_id) # server-side, never model-supplied
if facts is None:
raise ConfirmationError("unknown_transaction")
# 2. Missing token: reject before any side effect.
if not token:
audit("payment.blocked", reason="missing_token", txn_id=txn_id)
raise ConfirmationError("confirmation_required")
conf = verify_signature_and_load(token) # fails closed on bad signature
if conf is None:
audit("payment.blocked", reason="invalid_token", txn_id=txn_id)
raise ConfirmationError("invalid_confirmation")
# 3. Bind check: the token must describe this exact transaction.
if (conf.txn_id != txn_id
or conf.amount_minor != facts.amount_minor
or conf.currency != facts.currency
or conf.payee_id != payee_id
or amount_minor != facts.amount_minor
or currency != facts.currency):
audit("payment.blocked", reason="token_mismatch", txn_id=txn_id)
raise ConfirmationError("confirmation_mismatch")
# 4. Freshness and authority.
if now() > conf.expires_at:
audit("payment.blocked", reason="expired", txn_id=txn_id)
raise ConfirmationError("confirmation_expired")
if not authorize_subject(conf.subject_id, facts.required_role):
audit("payment.blocked", reason="unauthorized_subject", txn_id=txn_id)
raise ConfirmationError("unauthorized")
# 5. Consume token and write ledger in one transaction.
with ledger.transaction():
if not consume_token_once(conf.token_id):
audit("payment.blocked", reason="token_replay", txn_id=txn_id)
raise ConfirmationError("confirmation_reused")
ledger.post(txn_id, payee_id, facts.amount_minor, currency)
audit("payment.executed", txn_id=txn_id, token_id=conf.token_id,
subject=conf.subject_id, channel=conf.channel)
return "posting_created"
Three things to notice, because they are the difference between a gate and a decoration.
The facts come from look_up_transaction, not from the model's arguments. The model may propose an amount, but the gate compares the proposal to the server's record and to the token. A model that "rounds" 490.00 to 500.00 gets a mismatch error.
The token is consumed inside the same transaction as the ledger post. If the process dies between consumption and posting, you roll back and the customer can re-confirm. If you consume first and post later, a crash leaves a burned token and an unpaid invoice, which generates support tickets and tempts someone to loosen the check.
The rejection paths raise rather than return a truthy value. That matters because of how agents read tool results. Agno's implementation shows the pattern of a tool that will not execute until a human confirms, surfacing a structured requirement object to the application (the Agno guide). Returning an error the orchestrator can read, and that the audit log records, keeps the failure visible instead of letting the agent narrate success.