Autonomous Agent Credit Purchase Reference Implementation

// Reference implementation: x402 credit purchase for autonomous agents.
//
// This script exercises the full `/api/external/credits/purchase` flow:
//   1. POST without a payment header to receive the 402 `PAYMENT-REQUIRED`
//      challenge.
//   2. Parse the challenge, pick a supported acceptance (Base USDC here),
//      and sign an EIP-3009 `transferWithAuthorization` authorization via
//      EIP-712 typed-data signing.
//   3. Base64-encode the signed envelope and retry the POST with the
//      `X-PAYMENT` header set.
//
// Environment:
//   TEST_WALLET           JSON literal of the form
//                         {"address":"0x...","private_key":"0x..."}
//                         (64-hex private key, 0x-prefixed or not).
//   CREDITS_ENDPOINT      Optional override. Defaults to production.
//   PURCHASE_CREDITS      Optional override. Defaults to 500.
//   NETWORK_CAIP2         Optional override. Defaults to "eip155:8453".
//   TOKEN_SYMBOL          Optional override. Defaults to "USDC".
//
// Node >= 22. Dependencies: viem.

import { privateKeyToAccount } from "viem/accounts";

const ENDPOINT =
  process.env.CREDITS_ENDPOINT ||
  "https://www.agentpmt.com/api/external/credits/purchase";
const CREDITS = Number(process.env.PURCHASE_CREDITS || 500);
const NETWORK = process.env.NETWORK_CAIP2 || "eip155:8453";
const TOKEN = (process.env.TOKEN_SYMBOL || "USDC").toUpperCase();

const TRANSFER_WITH_AUTHORIZATION_TYPES = {
  TransferWithAuthorization: [
    { name: "from", type: "address" },
    { name: "to", type: "address" },
    { name: "value", type: "uint256" },
    { name: "validAfter", type: "uint256" },
    { name: "validBefore", type: "uint256" },
    { name: "nonce", type: "bytes32" },
  ],
};

function loadWallet() {
  const raw = process.env.TEST_WALLET;
  if (!raw) {
    throw new Error("TEST_WALLET env var is required (JSON with address + private_key).");
  }
  const parsed = JSON.parse(raw);
  const address = String(parsed.address || "").toLowerCase();
  const pkRaw = String(parsed.private_key || "");
  const privateKey = pkRaw.startsWith("0x") ? pkRaw : `0x${pkRaw}`;
  if (!/^0x[a-fA-F0-9]{64}$/.test(privateKey)) {
    throw new Error("TEST_WALLET.private_key must be a 32-byte hex string.");
  }
  if (!/^0x[a-fA-F0-9]{40}$/.test(address)) {
    throw new Error("TEST_WALLET.address must be a 20-byte hex address.");
  }
  return { address, privateKey };
}

function b64encodeJson(value) {
  return Buffer.from(JSON.stringify(value)).toString("base64");
}

function b64decodeJson(value) {
  return JSON.parse(Buffer.from(value, "base64").toString("utf8"));
}

function caip2ToChainId(network) {
  // CAIP-2 identifiers look like `eip155:<chainId>`. The purchase endpoint
  // advertises chains in this form.
  const match = /^eip155:(\d+)$/.exec(network);
  if (!match) {
    throw new Error(`Unsupported network identifier: ${network}`);
  }
  return Number(match[1]);
}

async function main() {
  const wallet = loadWallet();
  const account = privateKeyToAccount(wallet.privateKey);
  if (account.address.toLowerCase() !== wallet.address) {
    throw new Error(
      `Address mismatch: env=${wallet.address} derived=${account.address.toLowerCase()}`,
    );
  }
  console.log("payer:", account.address);

  // Step 1: trigger the 402 challenge.
  const initialBody = {
    wallet_address: account.address.toLowerCase(),
    credits: CREDITS,
    payment_method: "x402",
  };
  const challenge = await fetch(ENDPOINT, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify(initialBody),
  });
  if (challenge.status !== 402) {
    throw new Error(`Expected 402 challenge, got ${challenge.status}: ${await challenge.text()}`);
  }
  const required = challenge.headers.get("payment-required") || challenge.headers.get("PAYMENT-REQUIRED");
  if (!required) {
    throw new Error("PAYMENT-REQUIRED header missing from 402 response");
  }
  const requiredPayload = b64decodeJson(required);
  const acceptance = requiredPayload.accepts.find(
    (a) => a.network === NETWORK && a.extra && a.extra.name,
  );
  if (!acceptance) {
    throw new Error(`No acceptance for network ${NETWORK}`);
  }
  console.log(
    `acceptance: ${TOKEN} on ${acceptance.network} amount=${acceptance.amount} payTo=${acceptance.payTo}`,
  );

  // Step 2: sign the EIP-3009 authorization via EIP-712 typed data.
  const nowSeconds = Math.floor(Date.now() / 1000);
  const validAfter = 0n;
  const validBefore = BigInt(nowSeconds + Math.min(240, acceptance.maxTimeoutSeconds - 60));
  const nonceBytes = crypto.getRandomValues(new Uint8Array(32));
  const nonce = `0x${Array.from(nonceBytes, (b) => b.toString(16).padStart(2, "0")).join("")}`;

  const domain = {
    name: acceptance.extra.name,
    version: acceptance.extra.version,
    chainId: caip2ToChainId(acceptance.network),
    verifyingContract: acceptance.asset,
  };
  const message = {
    from: account.address,
    to: acceptance.payTo,
    value: BigInt(acceptance.amount),
    validAfter,
    validBefore,
    nonce,
  };
  const signature = await account.signTypedData({
    domain,
    types: TRANSFER_WITH_AUTHORIZATION_TYPES,
    primaryType: "TransferWithAuthorization",
    message,
  });

  // Step 3: retry with the base64-encoded envelope in the `X-PAYMENT` header.
  const envelope = {
    x402Version: 2,
    scheme: "exact",
    network: acceptance.network,
    asset: acceptance.asset,
    payload: {
      signature,
      authorization: {
        from: account.address,
        to: acceptance.payTo,
        value: acceptance.amount,
        validAfter: validAfter.toString(),
        validBefore: validBefore.toString(),
        nonce,
      },
    },
  };
  const xPaymentHeader = b64encodeJson(envelope);

  const settlement = await fetch(ENDPOINT, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "X-PAYMENT": xPaymentHeader,
    },
    body: JSON.stringify(initialBody),
  });
  const settlementBody = await settlement.text();
  console.log("settlement status:", settlement.status);
  console.log("settlement body  :", settlementBody);
  const paymentResponseHeader =
    settlement.headers.get("payment-response") || settlement.headers.get("PAYMENT-RESPONSE");
  if (paymentResponseHeader) {
    console.log("settlement header:", JSON.stringify(b64decodeJson(paymentResponseHeader), null, 2));
  }

  if (settlement.status !== 200) {
    process.exitCode = 1;
  }
}

main().catch((error) => {
  console.error("FATAL:", error?.stack || error);
  process.exitCode = 1;
});

// Reference implementation: signed credit-balance query for autonomous agents.
//
// Uses the two-step EIP-191 flow:
//   1. POST /api/external/auth/session with the wallet address to obtain a
//      session_nonce.
//   2. Build the canonical signing message, sign it with personal-sign, and
//      POST /api/external/credits/balance with the envelope fields.
//
// Environment:
//   TEST_WALLET        JSON literal {"address":"0x...","private_key":"0x..."}
//   BASE_URL           Optional override. Defaults to production.
//
// Node >= 22. Dependencies: viem.

import { privateKeyToAccount } from "viem/accounts";

const BASE_URL = process.env.BASE_URL || "https://www.agentpmt.com";

function loadWallet() {
  const raw = process.env.TEST_WALLET;
  if (!raw) {
    throw new Error("TEST_WALLET env var is required (JSON with address + private_key).");
  }
  const parsed = JSON.parse(raw);
  const address = String(parsed.address || "").toLowerCase();
  const pkRaw = String(parsed.private_key || "");
  const privateKey = pkRaw.startsWith("0x") ? pkRaw : `0x${pkRaw}`;
  if (!/^0x[a-fA-F0-9]{64}$/.test(privateKey)) {
    throw new Error("TEST_WALLET.private_key must be a 32-byte hex string.");
  }
  return { address, privateKey };
}

async function main() {
  const wallet = loadWallet();
  const account = privateKeyToAccount(wallet.privateKey);

  // 1) Open a wallet session to receive a session nonce.
  const sessionResp = await fetch(`${BASE_URL}/api/external/auth/session`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ wallet_address: wallet.address }),
  });
  if (!sessionResp.ok) {
    throw new Error(`Session creation failed: ${sessionResp.status} ${await sessionResp.text()}`);
  }
  const { session_nonce: sessionNonce } = await sessionResp.json();
  if (!sessionNonce) {
    throw new Error("Session response missing session_nonce");
  }

  // 2) Build the canonical EIP-191 message. The balance action uses an
  //    empty payload and a dash placeholder for the product field.
  const requestId = crypto.randomUUID();
  const canonicalMessage = [
    "agentpmt-external",
    `wallet:${wallet.address}`,
    `session:${sessionNonce}`,
    `request:${requestId}`,
    "action:balance",
    "product:-",
    "payload:",
  ].join("\n");
  const signature = await account.signMessage({ message: canonicalMessage });

  const balanceResp = await fetch(`${BASE_URL}/api/external/credits/balance`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      wallet_address: wallet.address,
      session_nonce: sessionNonce,
      request_id: requestId,
      signature,
    }),
  });
  console.log("balance status:", balanceResp.status);
  console.log("balance body  :", await balanceResp.text());
  if (!balanceResp.ok) {
    process.exitCode = 1;
  }
}

main().catch((error) => {
  console.error("FATAL:", error?.stack || error);
  process.exitCode = 1;
});

"""Reference implementation: x402 credit purchase for autonomous agents.

Mirrors ``buy-credits-example.mjs`` so Python-based agents can follow an
identical flow without reading TypeScript source.

Environment:
    TEST_WALLET           JSON literal ``{"address": "0x...", "private_key": "0x..."}``.
    CREDITS_ENDPOINT      Optional override. Defaults to production.
    PURCHASE_CREDITS      Optional override. Defaults to 500.
    NETWORK_CAIP2         Optional override. Defaults to ``eip155:8453``.
    TOKEN_SYMBOL          Optional override. Defaults to ``USDC``.

Dependencies: ``eth_account>=0.11``, ``requests>=2.32``. Python 3.11+.
"""

from __future__ import annotations

import base64
import json
import os
import re
import secrets
import sys
import time
from typing import Any

import requests
from eth_account import Account
from eth_account.messages import encode_typed_data


DEFAULT_ENDPOINT = "https://www.agentpmt.com/api/external/credits/purchase"

TRANSFER_WITH_AUTHORIZATION_TYPES = {
    "TransferWithAuthorization": [
        {"name": "from", "type": "address"},
        {"name": "to", "type": "address"},
        {"name": "value", "type": "uint256"},
        {"name": "validAfter", "type": "uint256"},
        {"name": "validBefore", "type": "uint256"},
        {"name": "nonce", "type": "bytes32"},
    ],
}


def _load_wallet() -> dict[str, str]:
    raw = os.environ.get("TEST_WALLET")
    if not raw:
        raise SystemExit("TEST_WALLET env var is required.")
    parsed = json.loads(raw)
    address = str(parsed["address"]).lower()
    pk = str(parsed["private_key"])
    if not pk.startswith("0x"):
        pk = "0x" + pk
    if not re.fullmatch(r"0x[0-9a-fA-F]{64}", pk):
        raise SystemExit("TEST_WALLET.private_key must be a 32-byte hex value.")
    if not re.fullmatch(r"0x[0-9a-fA-F]{40}", address):
        raise SystemExit("TEST_WALLET.address must be a 20-byte hex address.")
    return {"address": address, "private_key": pk}


def _b64_json(value: Any) -> str:
    return base64.b64encode(json.dumps(value, separators=(",", ":")).encode("utf-8")).decode("ascii")


def _parse_required_header(header_value: str) -> dict[str, Any]:
    return json.loads(base64.b64decode(header_value.encode("ascii")).decode("utf-8"))


def _caip2_to_chain_id(network: str) -> int:
    match = re.fullmatch(r"eip155:(\d+)", network)
    if not match:
        raise SystemExit(f"Unsupported network identifier: {network}")
    return int(match.group(1))


def main() -> int:
    endpoint = os.environ.get("CREDITS_ENDPOINT", DEFAULT_ENDPOINT)
    credits = int(os.environ.get("PURCHASE_CREDITS", "500"))
    network = os.environ.get("NETWORK_CAIP2", "eip155:8453")
    token_symbol = os.environ.get("TOKEN_SYMBOL", "USDC").upper()

    wallet = _load_wallet()
    account = Account.from_key(wallet["private_key"])
    if account.address.lower() != wallet["address"]:
        raise SystemExit(
            f"Address mismatch: env={wallet['address']} derived={account.address.lower()}"
        )
    print("payer:", account.address)

    # Step 1: trigger the 402 challenge.
    initial_body: dict[str, Any] = {
        "wallet_address": wallet["address"],
        "credits": credits,
        "payment_method": "x402",
    }
    challenge = requests.post(
        endpoint,
        json=initial_body,
        headers={"Content-Type": "application/json"},
        timeout=30,
    )
    if challenge.status_code != 402:
        raise SystemExit(f"Expected 402 challenge, got {challenge.status_code}: {challenge.text}")
    required_header = challenge.headers.get("PAYMENT-REQUIRED") or challenge.headers.get(
        "payment-required"
    )
    if not required_header:
        raise SystemExit("PAYMENT-REQUIRED header missing from 402 response")
    required_payload = _parse_required_header(required_header)

    acceptance = next(
        (
            a
            for a in required_payload["accepts"]
            if a["network"] == network and a.get("extra", {}).get("name")
        ),
        None,
    )
    if not acceptance:
        raise SystemExit(f"No acceptance for network {network} token {token_symbol}")
    print(
        f"acceptance: {token_symbol} on {acceptance['network']} "
        f"amount={acceptance['amount']} payTo={acceptance['payTo']}"
    )

    # Step 2: sign the EIP-3009 authorization via EIP-712 typed data.
    now_seconds = int(time.time())
    valid_after = 0
    valid_before = now_seconds + min(240, acceptance["maxTimeoutSeconds"] - 60)
    nonce_hex = "0x" + secrets.token_hex(32)

    typed_data = {
        "types": {
            "EIP712Domain": [
                {"name": "name", "type": "string"},
                {"name": "version", "type": "string"},
                {"name": "chainId", "type": "uint256"},
                {"name": "verifyingContract", "type": "address"},
            ],
            **TRANSFER_WITH_AUTHORIZATION_TYPES,
        },
        "primaryType": "TransferWithAuthorization",
        "domain": {
            "name": acceptance["extra"]["name"],
            "version": acceptance["extra"]["version"],
            "chainId": _caip2_to_chain_id(acceptance["network"]),
            "verifyingContract": acceptance["asset"],
        },
        "message": {
            "from": account.address,
            "to": acceptance["payTo"],
            "value": int(acceptance["amount"]),
            "validAfter": valid_after,
            "validBefore": valid_before,
            "nonce": nonce_hex,
        },
    }
    signable = encode_typed_data(full_message=typed_data)
    signed = account.sign_message(signable)
    signature_hex = signed.signature.hex()
    if not signature_hex.startswith("0x"):
        signature_hex = "0x" + signature_hex

    # Step 3: retry with the base64-encoded envelope in the `X-PAYMENT` header.
    envelope = {
        "x402Version": 2,
        "scheme": "exact",
        "network": acceptance["network"],
        "asset": acceptance["asset"],
        "payload": {
            "signature": signature_hex,
            "authorization": {
                "from": account.address,
                "to": acceptance["payTo"],
                "value": acceptance["amount"],
                "validAfter": str(valid_after),
                "validBefore": str(valid_before),
                "nonce": nonce_hex,
            },
        },
    }
    header_value = _b64_json(envelope)

    settlement = requests.post(
        endpoint,
        json=initial_body,
        headers={"Content-Type": "application/json", "X-PAYMENT": header_value},
        timeout=180,
    )
    print("settlement status:", settlement.status_code)
    print("settlement body  :", settlement.text)
    payment_response = settlement.headers.get("PAYMENT-RESPONSE") or settlement.headers.get(
        "payment-response"
    )
    if payment_response:
        print(
            "settlement header:",
            json.dumps(_parse_required_header(payment_response), indent=2),
        )

    return 0 if settlement.status_code == 200 else 1


if __name__ == "__main__":
    sys.exit(main())

"""Reference implementation: signed credit-balance query for autonomous agents.

Mirrors ``check-credits-balance-example.mjs``.

Environment:
    TEST_WALLET        JSON literal ``{"address": "0x...", "private_key": "0x..."}``.
    BASE_URL           Optional override. Defaults to production.

Dependencies: ``eth_account>=0.11``, ``requests>=2.32``. Python 3.11+.
"""

from __future__ import annotations

import json
import os
import sys
import uuid

import requests
from eth_account import Account
from eth_account.messages import encode_defunct


DEFAULT_BASE_URL = "https://www.agentpmt.com"


def _load_wallet() -> dict[str, str]:
    raw = os.environ.get("TEST_WALLET")
    if not raw:
        raise SystemExit("TEST_WALLET env var is required.")
    parsed = json.loads(raw)
    address = str(parsed["address"]).lower()
    pk = str(parsed["private_key"])
    if not pk.startswith("0x"):
        pk = "0x" + pk
    return {"address": address, "private_key": pk}


def main() -> int:
    base_url = os.environ.get("BASE_URL", DEFAULT_BASE_URL)
    wallet = _load_wallet()
    account = Account.from_key(wallet["private_key"])

    # 1) Open a wallet session.
    session_resp = requests.post(
        f"{base_url}/api/external/auth/session",
        json={"wallet_address": wallet["address"]},
        headers={"Content-Type": "application/json"},
        timeout=30,
    )
    if not session_resp.ok:
        raise SystemExit(
            f"Session creation failed: {session_resp.status_code} {session_resp.text}"
        )
    session_nonce = session_resp.json().get("session_nonce")
    if not session_nonce:
        raise SystemExit("Session response missing session_nonce")

    # 2) Build and sign the canonical EIP-191 balance message.
    request_id = str(uuid.uuid4())
    canonical_message = "\n".join(
        [
            "agentpmt-external",
            f"wallet:{wallet['address']}",
            f"session:{session_nonce}",
            f"request:{request_id}",
            "action:balance",
            "product:-",
            "payload:",
        ]
    )
    signable = encode_defunct(text=canonical_message)
    signed = account.sign_message(signable)
    signature_hex = signed.signature.hex()
    if not signature_hex.startswith("0x"):
        signature_hex = "0x" + signature_hex

    balance_resp = requests.post(
        f"{base_url}/api/external/credits/balance",
        json={
            "wallet_address": wallet["address"],
            "session_nonce": session_nonce,
            "request_id": request_id,
            "signature": signature_hex,
        },
        headers={"Content-Type": "application/json"},
        timeout=30,
    )
    print("balance status:", balance_resp.status_code)
    print("balance body  :", balance_resp.text)
    return 0 if balance_resp.ok else 1


if __name__ == "__main__":
    sys.exit(main())