Python

pip install flask requests

import base64, json, os

import requests
from flask import Flask, Response, g, request

API = "https://api.pennipay.eu/v1"
SITE_ID = os.environ["PENNI_SITE_ID"]
AUTH = {"Authorization": f"Bearer {os.environ['PENNI_API_KEY']}"}
FIELDS = ("scheme", "network", "asset", "payTo", "amount")

app = Flask(__name__)


def b64(obj):
    return base64.b64encode(json.dumps(obj).encode()).decode()


def payment_required(required, error):
    body = {**required, "error": error}
    return Response(json.dumps(body), 402, {"PAYMENT-REQUIRED": b64(body), "Content-Type": "application/json"})


@app.before_request
def penni():
    # 1. Price of this path (cache it for up to 60 seconds in production)
    config = requests.get(f"{API}/sites/{SITE_ID}/config", params={"path": request.path}, headers=AUTH, timeout=10).json()
    if config.get("blocked"):
        return Response('{"error": "Payments are temporarily unavailable"}', 503, {"Retry-After": "300"})
    if config["match"]["free"] or not config["accepts"]:
        return None  # free path, or nothing to pay with: serve as usual

    # 2. No payment yet: answer 402 with the price
    required = {"x402Version": 2, "resource": {"url": request.url}, "accepts": config["accepts"]}
    header = request.headers.get("PAYMENT-SIGNATURE")
    if not header:
        return payment_required(required, "PAYMENT-SIGNATURE header is required")

    # 3. The payment must match one of your current prices exactly
    payment = json.loads(base64.b64decode(header))
    accepted = payment.get("accepted") or {}
    chosen = next((a for a in config["accepts"] if all(a.get(k) == accepted.get(k) for k in FIELDS)), None)
    if chosen is None:
        return payment_required(required, "Payment does not match the current price")

    # 4. Verify, then settle on-chain
    body = {"x402Version": 2, "paymentPayload": payment, "paymentRequirements": chosen}
    verified = requests.post(f"{API}/x402/verify", json=body, headers=AUTH, timeout=10).json()
    if not verified.get("isValid"):
        return payment_required(required, verified.get("invalidReason", "invalid_payment"))
    for _ in range(2):  # Cardano may answer settlement_pending once: send the same body again
        settled = requests.post(f"{API}/x402/settle", json=body, headers=AUTH, timeout=90).json()
        if settled.get("errorReason") != "settlement_pending":
            break
    if not settled.get("success"):
        return payment_required(required, settled.get("errorReason", "settle_failed"))
    g.payment_response = b64(settled)


@app.after_request
def receipt(response):
    if "payment_response" in g:
        response.headers["PAYMENT-RESPONSE"] = g.payment_response
    return response


@app.get("/premium/report")
def report():
    return {"report": "..."}