aifinpay-gate

aifinpay-gate (Python)

Accept payments from AI agents in a Python API. When an agent calls a paid route without paying, the gate answers HTTP 402 with everything it needs to buy a batch of calls; the agent settles on-chain from its own wallet (you receive 99%, AiFinPay 1%, non-custodial) and retries with an AIFP-Receipt. The gate verifies that receipt locally — an Ed25519 signature against the AiFinPay JWKS, no call to AiFinPay per request — and meters the batch.

A port of @aifinpay/gate: the same 402 body, the same checks and the same refusal sentences, held to the Node package by recorded fixtures (tests/test_parity.py).

pip install aifinpay-gate            # PyNaCl is the only dependency
pip install "aifinpay-gate[redis]"   # shared counters across processes

Register

curl -s -X POST https://api.aifinpay.io/v1/merchants -H 'content-type: application/json' \
  -d '{"name":"My API","pay_to":{"evm":"0xYourPayoutWallet"},"settlement_version":"1.4"}'

Keep merchant_id and the one-time merchant_secret, then claim the merchant at https://dash.aifinpay.io with the secret to see payments. One pay_to.evm receives on every EVM network.

FastAPI / Starlette (ASGI)

from fastapi import FastAPI, Request
from aifinpay_gate import AifpGateMiddleware, Gate, Route

gate = Gate("mrch_…", routes=[
    Route("/create", "premium", methods={"POST"}),   # $0.005 per call
    Route("/api/*"),                                  # the section and everything under it
])
app = FastAPI()
app.add_middleware(AifpGateMiddleware, gate=gate)

@app.post("/create")
async def create(request: Request):
    paid = request.state.aifp      # {"agent", "receipt_id", "used", "remaining", ...}
    return {"ok": True, "remaining": paid["remaining"]}

Flask / Django (WSGI)

from aifinpay_gate import AifpGateWSGI, Gate, Route

app.wsgi_app = AifpGateWSGI(app.wsgi_app, Gate("mrch_…", routes=[Route("/api/*")]))
# in a view: request.environ["aifp"]

Both adapters also serve /.well-known/x402.json so agents discover the paid routes before they hit one (serve_discovery=False to turn it off).

Any other framework

from aifinpay_gate import Gate, SimpleRequest

result = gate.decide(SimpleRequest(path, headers, method))
if result is None:        # not a paid route
    ...
elif not result.ok:       # 402 / 403 / 503 — send result.status, result.headers, result.body as JSON
    ...
else:                     # paid: add result.headers to your response; result.aifp is the metering context
    ...

Routes

Route(pattern, tier="standard", weight=None, methods=None, paywall=True)

• pattern is exact, or ends in /* — /api/* covers /api and everything under /api/. The longest matching pattern wins. This is the hosted gateway's matcher; wildcards in the middle (/backtest/*/bid) are refused.
• tier: standard $0.0005, complex $0.002, premium $0.005 per call (weights 1, 4, 10 billing units). weight overrides the units per call.
• paywall=False serves the route free and meters nothing.
• Paths no route matches pass through untouched.

Gate(merchant_id, resource="/api/search", tier="complex") is the single-mount form: every request handed to it is charged against that one resource.

Production: share the counters

The default MemoryStore meters per process. Under gunicorn/uvicorn with several workers, or several pods, each gets a full copy of every batch — a 200-unit batch serves up to 200 × workers calls. Use Redis:

import redis
from aifinpay_gate import Gate, RedisStore

gate = Gate("mrch_…", routes=[...], store=RedisStore(redis.Redis.from_url(REDIS_URL)))

The store increments atomically and compares after the increment, so concurrent requests can never overspend a batch; the counter's TTL is set once and expires with the receipt.

Options

What it guarantees

• Fails closed: if the JWKS cannot be reached, paid routes answer 503, never free.
• Only EdDSA receipts from https://api.aifinpay.io with aud = your merchant id.
• Only quota receipts are spendable; per-call billing receipts are refused.
• An expired receipt gets a 402 (buy again), a wrong one a 403 with a reason.
• Stricter than the Node gate in two places: a receipt without a numeric exp, or without a usable unit_quota, is refused.

Not included, on purpose: free allowances, daily caps and per-agent blocks live in the AiFinPay dashboard, so there is one source of truth for each rule.

Opt-in dashboard request reporting (0.1.2)

Use one GateReporter per merchant per worker, created after worker fork. It uses only Python's standard library. Add its callback to your existing gate; keep the gate's shared Redis quota store and verification settings.

import os
from aifinpay_gate import GateReporter, Gate, Route, RedisStore

merchant_id = os.environ["AIFP_MERCHANT_ID"]
reporter = GateReporter(
    merchant_id=merchant_id,
    merchant_secret=os.environ["AIFP_MERCHANT_SECRET"],
)
gate = Gate(
    merchant_id=merchant_id,
    routes=[Route(pattern="/api/agent/genres/list", tier="standard")],
    store=RedisStore(redis_client),
    on_event=reporter.on_event,
)
# In your existing worker shutdown hook, after stopping new requests:
reporter.close(timeout=10.0)

Register the identical static or prefix route pattern in the dashboard first. Routes mode emits the matched Route.pattern; a single-resource gate must set resource explicitly. Full URLs, queries, fragments, encoded or invalid paths are rejected instead of rewritten. The API rejects unregistered patterns. Keep the merchant secret server-side. Share the reporter across mounts; do not report the same event through multiple reporters or through both hosted and self-hosted analytics. If you already have a callback, compose the two hooks.

Only 402 and gate-admitted serve events are reported; exempt humans and other event kinds are excluded. serve is admission before your handler, not proof of successful HTTP response or delivered content. The payload has only UUID, kind, registered path and UTC timestamp; no headers, receipts, payer IDs, IPs or customer content. This populates request counters, not geographic/user-agent charts or general website visits. Settlements still supply payments/revenue; purchased quota is never backfilled as actual usage.

api_base accepts an HTTPS origin only (default https://api.aifinpay.io); the reporting API must be deployed before enabling this hook. Redirects are blocked before a second request, including302, so secrets stay at that origin. on_event is synchronous enqueue only. The background daemon thread sends batches of at most 50 with a 3 second socket I/O timeout; it never performs HTTP on the gate thread. The memory queue includes in-flight events, holds at most 1000, and drops new events when full. UUIDs and timestamps stay identical on transient retries (at most 5 attempts/15 minutes). Permanent 4xx except 408/429, or redirects, stop reporting and discard the queue. Fix configuration or credentials before creating a replacement reporter.

reporter.stats returns queued, delivered (API acknowledgements including retry duplicates), dropped, retries, last_error (sanitized code), stopped and closed. Monitor these in your own service. An abrupt process stop loses unsent telemetry; reporting failures never alter verification or quotas. reporter.flush(timeout=10.0) requests one attempt per queued event and waits at most the deadline; transient failures remain for background retries. reporter.close(timeout=10.0) stops acceptance, flushes within that deadline, drops unsent events and tells the worker to stop. An already in-flight socket operation may finish later, but it cannot start another request. Neither method promises delivery. A reporter inherited across fork refuses to report and exposes wrong_process; construct a fresh reporter in each worker.