Python x402 Data Access Guide

import argparse
import json
import os
import string
import sys
import time
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from getpass import getpass
from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse

import requests

NETWORK = "eip155:8453"
BASE_USDC_ASSET = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913"
MAX_AMOUNT_ATOMIC = 10_000
PREVIEW_URL = "https://api.sourcestrand.com/v1/preview/crypto/asset-signal?asset=BTC"
PAID_URL = "https://api.sourcestrand.com/v1/paid/crypto/asset-signal?asset=BTC"
EXPECTED_HOST = "api.sourcestrand.com"
MAX_TIMEOUT_SECONDS = 60
EIP3009_DOMAIN_NAME = "USD Coin"
EIP3009_DOMAIN_VERSION = "2"
EIP3009_PRIMARY_TYPE = "TransferWithAuthorization"
EIP3009_TYPE_FIELDS = (
    ("from", "address"), ("to", "address"), ("value", "uint256"),
    ("validAfter", "uint256"), ("validBefore", "uint256"),
    ("nonce", "bytes32"),
)
REQUIRED_EXTENSION_KEY = "bazaar"

def _is_hex(value, length):
    return (
        isinstance(value, str) and len(value) == length and
        value.startswith("0x") and
        all(character in string.hexdigits for character in value[2:])
    )

def _usdc_atomic(price):
    try:
        value = Decimal(str(price).removeprefix("$"))
    except InvalidOperation as error:
        raise RuntimeError("The payment offer returned an invalid price") from error
    atomic = value * Decimal(1_000_000)
    if atomic != atomic.to_integral_value():
        raise RuntimeError("The payment offer uses more than six USDC decimals")
    amount = int(atomic)
    if amount <= 0 or amount > MAX_AMOUNT_ATOMIC:
        raise RuntimeError("The payment offer exceeds this guide's signing cap")
    return amount

def _buyer_hint():
    value = os.getenv("SOURCESTRAND_BUYER_HINT", "").strip()
    if len(value) > 120:
        raise ValueError("SOURCESTRAND_BUYER_HINT must be at most 120 characters")
    return value

def _buyer_headers():
    headers = {"accept": "application/json"}
    buyer_hint = _buyer_hint()
    if buyer_hint:
        headers["x-buyer-hint"] = buyer_hint
    return headers

def _same_url(left, right):
    left_url, right_url = urlparse(left), urlparse(right)
    return (
        left_url.scheme == right_url.scheme == "https" and
        left_url.hostname == right_url.hostname == EXPECTED_HOST and
        left_url.port in (None, 443) and right_url.port in (None, 443) and
        not left_url.username and not right_url.username and
        not left_url.password and not right_url.password and
        not left_url.fragment and not right_url.fragment and
        left_url.path == right_url.path and
        sorted(parse_qsl(left_url.query, keep_blank_values=True)) ==
        sorted(parse_qsl(right_url.query, keep_blank_values=True))
    )

def _attributed_paid_url():
    parsed = urlparse(PAID_URL)
    if (
        parsed.scheme != "https" or parsed.hostname != EXPECTED_HOST or
        parsed.port not in (None, 443) or parsed.username or parsed.password or
        parsed.fragment
    ):
        raise RuntimeError("Refusing a paid URL outside the SourceStrand allowlist")
    query = dict(parse_qsl(parsed.query, keep_blank_values=True))
    buyer_hint = _buyer_hint()
    if buyer_hint:
        query["buyer_hint"] = buyer_hint
    return urlunparse(parsed._replace(query=urlencode(query)))

def preview():
    response = requests.get(
        PREVIEW_URL, headers=_buyer_headers(), timeout=20, allow_redirects=False
    )
    response.raise_for_status()
    payload = response.json()
    if not isinstance(payload, dict):
        raise RuntimeError("SourceStrand preview returned a non-object payload")
    return payload

def _payment_intent(paid_url):
    parsed = urlparse(paid_url)
    paid_query = dict(parse_qsl(parsed.query, keep_blank_values=True))
    intent_query = {"resource_url": paid_url}
    if paid_query.get("buyer_hint"):
        intent_query["buyer_hint"] = paid_query["buyer_hint"]
    intent_url = urlunparse((
        "https", parsed.netloc, "/v1/payment/intent", "",
        urlencode(intent_query), ""
    ))
    response = requests.get(
        intent_url, headers=_buyer_headers(),
        timeout=20, allow_redirects=False
    )
    response.raise_for_status()
    intent = response.json()
    try:
        required_amount = int(intent["required_amount_atomic"])
        pay_to = str(intent["pay_to"])
        intent_price_atomic = _usdc_atomic(intent["price"])
    except (KeyError, TypeError, ValueError) as error:
        raise RuntimeError("Payment intent omitted amount, price, or recipient") from error
    if (
        intent.get("status") != "approval_required" or
        intent.get("payment_protocol") != "x402" or
        not _same_url(str(intent.get("resource_url", "")), paid_url) or
        intent.get("network") != NETWORK or
        str(intent.get("asset_address", "")).lower() != BASE_USDC_ASSET.lower() or
        required_amount != MAX_AMOUNT_ATOMIC or
        intent_price_atomic != MAX_AMOUNT_ATOMIC or
        not _is_hex(pay_to, 42)
    ):
        raise RuntimeError("Payment intent does not match the approved purchase")
    return {"amount": required_amount, "pay_to": pay_to, "body": intent}

def source_strand_base_usdc_only(
    version, requirements, expected_amount=MAX_AMOUNT_ATOMIC, expected_pay_to=None
):
    if version != 2 or len(requirements) != 1:
        return []
    allowed = []
    for requirement in requirements:
        try:
            amount = int(requirement.get_amount())
            pay_to = str(requirement.pay_to)
            extra = requirement.extra
            max_timeout_seconds = requirement.max_timeout_seconds
        except (AttributeError, TypeError, ValueError):
            continue
        required_extra_keys = {"name", "version", "price"}
        if (
            not isinstance(extra, dict) or
            not required_extra_keys.issubset(extra) or
            set(extra) - (required_extra_keys | {"assetTransferMethod"}) or
            extra.get("assetTransferMethod", "eip3009") != "eip3009"
        ):
            continue
        try:
            extra_price_atomic = _usdc_atomic(extra.get("price"))
        except (RuntimeError, TypeError, ValueError):
            continue
        if (
            requirement.scheme == "exact" and
            requirement.network == NETWORK and
            str(requirement.asset).lower() == BASE_USDC_ASSET.lower() and
            amount == expected_amount and 0 < amount <= MAX_AMOUNT_ATOMIC and
            expected_pay_to is not None and
            pay_to.lower() == expected_pay_to.lower() and
            max_timeout_seconds == MAX_TIMEOUT_SECONDS and
            extra.get("name") == EIP3009_DOMAIN_NAME and
            extra.get("version") == EIP3009_DOMAIN_VERSION and
            extra_price_atomic == expected_amount
        ):
            allowed.append(requirement)
    return allowed

class GuardedEIP3009Signer:
    def __init__(self, delegate, payer, amount, pay_to):
        self._delegate = delegate
        self._payer = payer
        self._amount = amount
        self._pay_to = pay_to
        self.sign_calls = 0

    @property
    def address(self):
        return str(self._delegate.address)

    def sign_typed_data(self, domain, types, primary_type, message):
        if self.sign_calls != 0:
            raise RuntimeError("Signer refused more than one typed-data signature")
        self.sign_calls += 1
        try:
            fields = tuple(
                (str(field.name), str(field.type))
                for field in types[EIP3009_PRIMARY_TYPE]
            )
            valid_before = int(message["validBefore"])
            nonce = message["nonce"]
        except (AttributeError, KeyError, TypeError, ValueError) as error:
            raise RuntimeError("Signer received malformed EIP-3009 typed data") from error
        now_seconds = int(time.time())
        if (
            primary_type != EIP3009_PRIMARY_TYPE or
            set(types) != {EIP3009_PRIMARY_TYPE} or
            fields != EIP3009_TYPE_FIELDS or
            getattr(domain, "name", None) != EIP3009_DOMAIN_NAME or
            getattr(domain, "version", None) != EIP3009_DOMAIN_VERSION or
            getattr(domain, "chain_id", None) != 8453 or
            str(getattr(domain, "verifying_contract", "")).lower() !=
                BASE_USDC_ASSET.lower() or
            set(message) !=
                {"from", "to", "value", "validAfter", "validBefore", "nonce"} or
            str(message.get("from", "")).lower() != self._payer.lower() or
            str(message.get("to", "")).lower() != self._pay_to.lower() or
            message.get("value") != self._amount or
            message.get("validAfter") != 0 or
            valid_before <= now_seconds or
            valid_before > now_seconds + MAX_TIMEOUT_SECONDS + 5 or
            not isinstance(nonce, bytes) or len(nonce) != 32
        ):
            raise RuntimeError("Signer refused typed data outside the approved payment")
        signature = self._delegate.sign_typed_data(
            domain, types, primary_type, message
        )
        if not isinstance(signature, bytes) or len(signature) != 65:
            raise RuntimeError("Wallet returned an invalid EIP-3009 signature")
        return signature

def _validate_payment_context(required, paid_url):
    resource = getattr(required, "resource", None)
    if (
        resource is None or
        not _same_url(str(getattr(resource, "url", "")), paid_url) or
        getattr(resource, "mime_type", None) != "application/json"
    ):
        raise RuntimeError("402 challenge resource does not match the approved URL")
    extensions = getattr(required, "extensions", None)
    if (
        not isinstance(extensions, dict) or
        set(extensions) != {REQUIRED_EXTENSION_KEY} or
        not isinstance(extensions.get(REQUIRED_EXTENSION_KEY), dict)
    ):
        raise RuntimeError("402 challenge contains unexpected payment extensions")

def _same_requirement(left, right):
    try:
        return (
            left.scheme == right.scheme and left.network == right.network and
            str(left.asset).lower() == str(right.asset).lower() and
            str(left.get_amount()) == str(right.get_amount()) and
            str(left.pay_to).lower() == str(right.pay_to).lower() and
            left.max_timeout_seconds == right.max_timeout_seconds and
            left.extra == right.extra
        )
    except (AttributeError, TypeError, ValueError):
        return False

def _same_resource(left, right):
    if left is None or right is None:
        return False
    fields = ("url", "description", "mime_type", "service_name", "tags", "icon_url")
    return all(
        getattr(left, field, None) == getattr(right, field, None)
        for field in fields
    )

def _validate_created_payload(
    payment_payload, required, paid_url, payer, amount, pay_to, created_after
):
    _validate_payment_context(required, paid_url)
    accepted = getattr(payment_payload, "accepted", None)
    accepts = getattr(required, "accepts", None)
    if (
        getattr(payment_payload, "x402_version", None) != 2 or
        not isinstance(accepts, list) or len(accepts) != 1 or
        accepted is None or not _same_requirement(accepted, accepts[0]) or
        not source_strand_base_usdc_only(2, [accepted], amount, pay_to)
    ):
        raise RuntimeError("Created payload changed the sole approved requirement")
    if (
        not _same_resource(payment_payload.resource, required.resource) or
        not _same_url(str(payment_payload.resource.url), paid_url) or
        payment_payload.extensions != required.extensions
    ):
        raise RuntimeError("Created payload changed the approved resource or extensions")
    inner = getattr(payment_payload, "payload", None)
    if not isinstance(inner, dict) or set(inner) != {"authorization", "signature"}:
        raise RuntimeError("Created payload is not an exact EIP-3009 payload")
    authorization = inner.get("authorization")
    signature = inner.get("signature")
    if (
        not isinstance(authorization, dict) or
        set(authorization) !=
            {"from", "to", "value", "validAfter", "validBefore", "nonce"} or
        not _is_hex(signature, 132) or
        str(authorization.get("from", "")).lower() != payer.lower() or
        str(authorization.get("to", "")).lower() != pay_to.lower() or
        authorization.get("value") != str(amount) or
        authorization.get("validAfter") != "0" or
        not _is_hex(authorization.get("nonce"), 66)
    ):
        raise RuntimeError("Created EIP-3009 authorization drifted from approval")
    valid_before_text = authorization.get("validBefore")
    if not isinstance(valid_before_text, str) or not valid_before_text.isdecimal():
        raise RuntimeError("Created EIP-3009 authorization has an invalid timeout")
    valid_before = int(valid_before_text)
    if (
        valid_before <= int(time.time()) or
        valid_before > created_after + MAX_TIMEOUT_SECONDS + 5
    ):
        raise RuntimeError("Created EIP-3009 authorization exceeds the approved timeout")

def _payment_required(parser, paid_url, intent):
    response = requests.get(
        paid_url, headers=_buyer_headers(), timeout=20, allow_redirects=False
    )
    if response.status_code != 402:
        raise RuntimeError("Expected a 402 challenge before initializing a signer")
    try:
        body = response.json()
    except ValueError:
        body = None
    required = parser.get_payment_required_response(
        lambda name: response.headers.get(name), body
    )
    _validate_payment_context(required, paid_url)
    allowed = source_strand_base_usdc_only(
        required.x402_version, required.accepts,
        intent["amount"], intent["pay_to"]
    )
    if not allowed:
        raise RuntimeError("402 challenge does not match amount, recipient, Base, and USDC")
    return required

def _wallet_account():
    from eth_account import Account

    private_key = os.environ.pop("EVM_PRIVATE_KEY", "").strip()
    if not private_key:
        private_key = getpass("Dedicated hot-wallet private key (input hidden): ").strip()
    if not private_key:
        raise RuntimeError("A dedicated hot-wallet private key is required for --buy")
    try:
        return Account.from_key(private_key)
    finally:
        private_key = ""

def _validate_settled_receipt(receipt, payer, expected_amount, asset, pay_to):
    if receipt.success is not True:
        raise RuntimeError("NOT_SETTLED: facilitator reported an unsuccessful settlement")
    extra = getattr(receipt, "extra", None)
    if (
        receipt.network != NETWORK or
        not _is_hex(receipt.transaction, 66) or
        not isinstance(receipt.payer, str) or
        receipt.payer.lower() != payer.lower() or
        not isinstance(receipt.amount, str) or
        receipt.amount != str(expected_amount) or
        not isinstance(extra, dict) or
        not isinstance(extra.get("asset"), str) or
        extra["asset"].lower() != asset.lower() or
        not isinstance(extra.get("payTo"), str) or
        extra["payTo"].lower() != pay_to.lower()
    ):
        raise RuntimeError(
            "SETTLED_RECEIPT_MISMATCH: payment may be settled; do not retry"
        )

def _send_signed_request_once(paid_url, payment_headers):
    session = requests.Session()
    session.mount("https://", requests.adapters.HTTPAdapter(max_retries=0))
    try:
        return session.get(
            paid_url,
            headers={**_buyer_headers(), **payment_headers},
            timeout=30,
            allow_redirects=False,
        )
    finally:
        session.close()

def _raise_if_reconciliation_required(response):
    if response.status_code != 409:
        return
    try:
        body = response.json()
    except ValueError:
        return
    if (
        not isinstance(body, dict) or
        body.get("error") != "PAYMENT_RECONCILIATION_REQUIRED"
    ):
        return
    safe_characters = set(string.ascii_letters + string.digits + "-_.:")
    def safe_token(value):
        token = "".join(
            character for character in str(value) if character in safe_characters
        )[:100]
        return token or "unknown"
    reconciliation_id = safe_token(body.get("reconciliation_id", "unknown"))
    reason = safe_token(body.get("reason", "unknown"))
    raise RuntimeError(
        "PAYMENT_RECONCILIATION_REQUIRED: id=" + reconciliation_id +
        " reason=" + reason +
        "; preserve this response, reconcile Base USDC activity, and do not retry"
    )

def buy_once():
    from x402 import max_amount, x402ClientSync
    from x402.http import x402HTTPClientSync
    from x402.mechanisms.evm import EthAccountSigner
    from x402.mechanisms.evm.exact.register import register_exact_evm_client

    preview()
    paid_url = _attributed_paid_url()
    intent = _payment_intent(paid_url)
    preflight_parser = x402HTTPClientSync(x402ClientSync())
    required = _payment_required(preflight_parser, paid_url, intent)
    account = _wallet_account()

    client = x402ClientSync()
    def approved_purchase_only(version, requirements):
        return source_strand_base_usdc_only(
            version, requirements, intent["amount"], intent["pay_to"]
        )
    client.register_policy(approved_purchase_only)
    client.register_policy(max_amount(MAX_AMOUNT_ATOMIC))
    guarded_signer = GuardedEIP3009Signer(
        EthAccountSigner(account), account.address,
        intent["amount"], intent["pay_to"]
    )
    register_exact_evm_client(client, guarded_signer, networks=NETWORK)
    parser = x402HTTPClientSync(client)
    audit = {
        "authorized_at": datetime.now(timezone.utc).isoformat(),
        "resource": paid_url,
        "payer": account.address,
        "pay_to": intent["pay_to"],
        "amount_atomic": intent["amount"],
        "network": NETWORK,
    }
    print(json.dumps({"payment_approval": audit}, indent=2), file=sys.stderr)

    try:
        created_after = int(time.time())
        payment_payload = client.create_payment_payload(required)
        if guarded_signer.sign_calls != 1:
            raise RuntimeError("Checkout did not create exactly one approved signature")
        _validate_created_payload(
            payment_payload, required, paid_url, account.address,
            intent["amount"], intent["pay_to"], created_after
        )
        payment_headers = parser.encode_payment_signature_header(payment_payload)
        if (
            not isinstance(payment_headers, dict) or
            set(payment_headers) != {"PAYMENT-SIGNATURE"} or
            not isinstance(payment_headers["PAYMENT-SIGNATURE"], str) or
            not payment_headers["PAYMENT-SIGNATURE"]
        ):
            raise RuntimeError("x402 client returned an invalid payment header")
    except Exception as error:
        raise RuntimeError(
            "PRE_SIGN_FAILURE: no signed request was sent; fix the error before rerunning"
        ) from error

    try:
        response = _send_signed_request_once(paid_url, payment_headers)
    except Exception as error:
        raise RuntimeError(
            "UNKNOWN_SETTLEMENT: a signed request was sent. Check the wallet's Base "
            "USDC activity using payment_approval before any retry."
        ) from error

    _raise_if_reconciliation_required(response)

    try:
        receipt = parser.get_payment_settle_response(
            lambda name: response.headers.get(name)
        )
    except Exception as error:
        raise RuntimeError(
            "UNKNOWN_SETTLEMENT: PAYMENT-RESPONSE was missing or invalid. Reconcile "
            "the recorded payer, recipient, amount, resource, and UTC time before retrying."
        ) from error

    receipt_data = receipt.model_dump(mode="json")
    print(json.dumps({"payment_receipt": receipt_data}, indent=2), file=sys.stderr)
    _validate_settled_receipt(
        receipt, account.address, intent["amount"],
        BASE_USDC_ASSET, intent["pay_to"]
    )
    if not response.ok:
        raise RuntimeError(
            "PAYMENT_SETTLED_RESOURCE_ERROR: keep the receipt and do not pay again"
        )
    try:
        data = response.json()
    except ValueError as error:
        raise RuntimeError(
            "PAYMENT_SETTLED_DATA_ERROR: keep the receipt and do not pay again"
        ) from error
    if not isinstance(data, dict):
        raise RuntimeError(
            "PAYMENT_SETTLED_DATA_ERROR: paid response was not a JSON object"
        )
    return {"data": data, "payment_receipt": receipt_data}

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--buy", action="store_true",
        help="Explicitly authorize one capped Base USDC payment"
    )
    args = parser.parse_args()
    result = buy_once() if args.buy else preview()
    print(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()

• Use only the canonical HTTPS host api.sourcestrand.com.
• Accept only x402 version 2 with the exact scheme.
• Restrict the network to Base mainnet, eip155:8453.
• Restrict the asset to native Base USDC, 0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913.
• Require payment-intent amount and recipient to match the actual 402 challenge exactly.
• Apply a positive per-payment cap before the signer can approve a requirement.
• Keep the one-shot signer guard: verify the exact EIP-712 domain and EIP-3009 message, then allow one signature only.
• Disable redirects for preview and payment-enabled requests.
• Disable transport retries for the signed request and send it exactly once.
• Require a parsed receipt with success=true, matching payer, amount, asset, recipient, network, and transaction hash.
• Treat a timeout or parse failure after signing as unknown settlement; never retry blindly.

If the script reports UNKNOWN_SETTLEMENT, keep the emitted payment_approval record, inspect that payer's Base USDC activity for the recorded recipient, amount, resource, and UTC time, and do not run --buy again until the wallet history is reconciled. If the server returns HTTP 409 PAYMENT_RECONCILIATION_REQUIRED, it deliberately omits PAYMENT-RESPONSE: preserve the sanitized reconciliation_id and reason, reconcile on-chain activity, and never resend blindly. Never save or share the key or payment signature.

Verify protocol behavior against the official x402 buyer guide and the asset address against Circle's USDC contract list.