x402-firewall 0.1.0

Spending firewall middleware for x402 payments. Policy engine, approval queue, and audit log for AI agents.

x402-firewall

Spending firewall middleware for x402 payments.

@guard(budget_usd=25.00, require_approval_above=5.00)
async def my_agent():
    fw = get_firewall()
    data = await fw.pay_x402("https://data.example.com/report", max_usd=1.50)

Every x402 payment inside the decorated function is checked against your policy before it executes. Too expensive → blocked. Above the approval threshold → paused until a human approves. Over budget → raises immediately. Everything is audit-logged.

---

Install

pip install x402-firewall

Zero required dependencies.

---

Quickstart

import asyncio
from x402_firewall import guard, get_firewall, PaymentBlocked, PaymentPendingApproval

@guard(
    budget_usd=25.00,           # max total spend per call
    require_approval_above=5.00, # payments ≥ $5 need human approval
    block_vendors=["untrusted-vendor.com"],
)
async def research_agent(query: str):
    fw = get_firewall()

    # This payment is $1.50 — allowed immediately
    result = await fw.pay_x402(
        "https://data.example.com/report",
        max_usd=1.50,
        chain="base",
    )

    # This payment is $8.00 — escalated to approval queue
    # (blocks here until a human approves or timeout)
    premium = await fw.pay_x402(
        "https://premium.example.com/dataset",
        max_usd=8.00,
        chain="base",
    )

    return result, premium

asyncio.run(research_agent("latest AI research"))

---

Policy options

@guard(
    budget_usd=25.00,               # session budget cap
    max_single_payment_usd=10.00,   # hard cap per payment
    require_approval_above=5.00,    # escalate above this amount
    block_vendors=["spam.com", r".*\.evil\.com$"],  # regex supported
    allow_vendors=["trusted.com"],  # allowlist mode (blocks all others)
    allow_chains=["base", "solana"],# only these chains
    block_chains=["ethereum"],      # block specific chains
    max_payments_per_session=10,    # max number of payments
)
async def my_agent(): ...

---

Human approval queue

When a payment exceeds require_approval_above, the agent pauses and waits for a human decision. You can wire up any notification channel.

Webhook (Slack, custom server)

@guard(
    budget_usd=50.00,
    require_approval_above=2.00,
    webhook_url="https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
    approval_timeout=300.0,  # seconds to wait (default: 300)
)
async def my_agent(): ...

The webhook receives a POST with:

{
  "request_id": "abc-123",
  "amount_usd": 8.50,
  "vendor": "premium.example.com",
  "chain": "base",
  "session_id": "sess-4f2a"
}

Your server calls queue.resolve(request_id, approved=True) to unblock the agent.

Custom notification (callback)

from x402_firewall import ApprovalQueue

queue = ApprovalQueue(
    on_pending=lambda req: send_telegram(
        f"Approve payment of ${req.amount_usd:.2f} to {req.vendor}? "
        f"ID: {req.request_id}"
    )
)

@guard(budget_usd=50.00, require_approval_above=2.00, approval_queue=queue)
async def my_agent(): ...

# From your Telegram bot handler:
queue.resolve("abc-123", approved=True, approver="Clarence")

In-process approval (for testing)

from x402_firewall import ApprovalQueue
import asyncio

queue = ApprovalQueue()

@guard(budget_usd=50.00, require_approval_above=2.00, approval_queue=queue)
async def my_agent(): ...

async def main():
    task = asyncio.create_task(my_agent())
    await asyncio.sleep(0.1)          # let the agent reach the payment
    queue.auto_approve()              # approve all pending
    await task

---

Audit log

Every payment decision is recorded — blocked, allowed, escalated, approved, rejected.

from x402_firewall import AuditLog, guard, get_firewall

log = AuditLog(path="/var/log/x402-agent.jsonl")  # or None for in-memory

@guard(budget_usd=25.00, audit_log=log)
async def my_agent(): ...

# After the run:
entries = log.entries_for_session("sess-abc")
print(f"Total spent: ${log.total_spent('sess-abc'):.4f}")

Each JSONL line:

{
  "decision": "allowed",
  "amount_usd": 1.5,
  "vendor": "data.example.com",
  "chain": "base",
  "session_id": "sess-4f2a",
  "timestamp": 1750000000.0,
  "reason": "",
  "approver": ""
}

---

Manual firewall (no decorator)

from x402_firewall import Firewall, FirewallContext, Policy, AuditLog

fw = Firewall(
    policy=Policy(budget_usd=25.00, require_approval_above=5.00),
    audit_log=AuditLog(path="/var/log/x402.jsonl"),
)
ctx = FirewallContext(session_id="my-agent-session")

# Pre-flight check
await fw.check_payment(1.50, "https://data.example.com/r", "base", ctx)

# Or full payment flow
response = await fw.pay_x402("https://data.example.com/r", max_usd=1.50, ctx=ctx)

---

Load from YAML config

# policy.yaml
budget_usd: 25.00
require_approval_above: 5.00
block_vendors:
  - "untrusted.com"
  - ".*\\.spam\\..*"
allow_chains:
  - base
  - solana
max_payments_per_session: 20

import yaml
from x402_firewall import Policy, Firewall

with open("policy.yaml") as f:
    policy = Policy.from_dict(yaml.safe_load(f))

fw = Firewall(policy=policy)

---

EU AI Act compliance

The require_approval_above threshold combined with the audit log gives you Article 14 (human-in-the-loop) compliance for AI agents making financial decisions. Every escalated payment is:

• Paused before execution
• Sent to a designated human approver
• Logged with approver identity and timestamp

This is the documented compliance mechanism for high-risk AI systems under the EU AI Act (enforcement starts August 2026).

For teams needing a managed dashboard, hosted approval queue, and compliance reports: → bonanza-labs.com/firewall

---

License

Apache 2.0 — see LICENSE.

Built by Bonanza Labs.