Payments X402¶

payments_x402 — x402 HTTP 402 agent payment protocol for G6.

Cluster: Uncategorised | Type: component | MCP Tools: 14

Overview¶

Implements the HTTP 402 agent payment protocol for G6, enabling autonomous agents to make and manage cryptocurrency payments. Provides a full payment pipeline: policy evaluation (spending limits, token/chain allowlists, anti-structuring detection), human-in-the-loop (HITL) approval gates, wallet signing (with dry-run adapter for testing), and immutable audit logging.

Enforces safety-first defaults including mandatory HITL for first transactions and high-value payments.

When to use:

• Enabling autonomous agents to pay for external API calls or compute resources
• Enforcing spending policy with HITL approval gates before any real payment
• Auditing payment history with immutable, tamper-evident logs

Example:

from mvp.payments_x402 import PaymentsX402Block, PaymentsX402Input

block = PaymentsX402Block()
result = block.infer(PaymentsX402Input(
    operation="pay",
    amount_wei=1000,
    token="USDC",
    chain="base",
    payee="0xrecipient...",
))
# result.value -> PaymentsX402Output with status, tx_hash, policy_decision

Caveats and known limitations:

• The default wallet adapter is dry-run only; real transactions require an injected production WalletAdapter. A simulated settlement is machine-readable via completion_state="qualified-draft" and audit_fields.dry_run=True / evidence.dry_run=True (not just the 0xdryrun_ hash prefix).
• The real wallet/payment-provider path must pass staging E2E before paid-user launch
• Audit records include full payment details including addresses — consider PII and compliance implications
• No explicit rate limiting on approval requests — an agent could spam the HITL queue
• Modify/delete audit operations are blocked at the protocol level (returns error, never executes)
• verify_facilitator is a string allowlist check only (it compares the facilitator against facilitator_allowlist); it is not cryptographic facilitator attestation.
• KYC is declared but not enforced: kyc_threshold_wei exists on SpendingPolicy but is currently inert. Enforcing it would tighten the HITL approval gate and is a human-reviewed policy decision (see the remediation plan), not an automated capability.
• The MCP surface (payments_x402_mcp/) is narrowed: read-only discovery/query/verify ops are open, HITL approval-lifecycle ops require an operator-authority gate, and pay is never exposed (no settlement over MCP without a certified real wallet).

Works well with: token_budget, work_loop, align_csf

Public API¶

AuditStore¶

Append-only SQLite audit trail for payment transactions.

Constructor:

Methods:

record(fields: dict[str, Any]) -> str¶

Insert an audit record and return the row id as a string.

query(filters: dict[str, Any], limit: int = 100) -> list[dict[str, Any]]¶

SELECT records matching filters (equality on each key).

approval_history() -> list[dict[str, Any]]¶

Return every row carrying an approval_token, oldest first.

export(payer: str, date_range: dict[str, str] | None = None, format: str = 'json') -> list[dict[str, Any]]¶

Export audit records for a given payer, optionally filtered by date range.

check_idempotency(key: str) -> dict[str, Any] | None¶

Return the existing record if idempotency_key matches, else None.

ApprovalStatus(Enum)¶

ApprovalRecord¶

Mutable record tracking a single approval request.

HITLGate¶

Human-in-the-loop gate that mediates payment approvals.

Methods:

restore_pending(records: Iterable[ApprovalRecord]) -> int¶

Seed in-memory state with reconstructed pending approvals.

create_approval(payer: str, payee: str, amount_wei: str, reason: str, token_name: str = '', chain: str = '', facilitator: str = '', nonce: int = 0, idempotency_key: str = '', payer_attested: str = '', payer_attested_user: str = '') -> str¶

Create a new pending approval and return its token.

approve(token: str, approver: str, conditions: dict | None = None, approver_user: str = '') -> Result[ApprovalRecord]¶

Approve a pending request.

reject(token: str, approver: str, reason: str = '') -> Result[ApprovalRecord]¶

Reject a pending request.

cancel(token: str, requester: str) -> Result[ApprovalRecord]¶

Cancel a pending request.

complete(token: str) -> Result[ApprovalRecord]¶

Mark an approved request as completed after successful signing.

check_status(token: str, current_time: str | None = None, timeout_seconds: int = 300) -> Result[ApprovalRecord]¶

Return the current status, marking expired if timed out.

get_pending(payer: str) -> list[ApprovalRecord]¶

Return all pending records for payer.

bulk_approve(tokens: list[str], approver: str, approver_user: str = '') -> Result[list[ApprovalRecord]]¶

Approve every token in tokens atomically.

PaymentsX402Block(AIBlock[PaymentsX402Input, PaymentsX402Output, None])¶

AIBlock orchestrating x402 payment flow.

Methods:

infer(data: PaymentsX402Input) -> Result[PaymentsX402Output]¶

process(inp: PaymentsX402Input) -> Result[PaymentsX402Output]¶

PolicyDecision(Enum)¶

PolicyResult¶

PolicyEngine¶

Constructor:

Methods:

evaluate(inp: PaymentsX402Input, context: dict | None = None, commit: bool = True) -> PolicyResult¶

record_approved(inp: PaymentsX402Input) -> None¶

Commit spend/nonce bookkeeping after a payment is actually signed.

PaymentOperation(Enum)¶

SpendingPolicy¶

X402Header¶

PaymentsX402Input¶

PaymentsX402Output¶

WalletAdapter(Protocol)¶

Signing interface for x402 payment transactions.

Methods:

sign_payment(payee: str, amount_wei: str, token: str, chain: str, nonce: int) -> Result[str]¶

Sign a payment and return the transaction hash.

get_balance(token: str, chain: str) -> Result[str]¶

Return the balance in wei for the given token on chain.

DryRunWalletAdapter¶

Test / dry-run wallet that NEVER makes real transactions.

Methods:

sign_payment(payee: str, amount_wei: str, token: str, chain: str, nonce: int) -> Result[str]¶

get_balance(token: str, chain: str) -> Result[str]¶

X402ParseResult¶

Parsed representation of an x402 payment header.

Functions¶

parse_x402_header(header: dict | None) -> X402ParseResult¶

Parse a raw x402 header dict into an X402ParseResult.

validate_request_matches_header(amount_wei: str, token: str, chain: str, facilitator: str, header: X402ParseResult) -> Result[None]¶

Check that the caller's request parameters match the parsed header.

MCP Tools¶