X402

import os
import asyncio
from solders.keypair import Keypair
from x402 import x402Client, prefer_network
from x402.http.clients.httpx import x402HttpxClient
from x402.mechanisms.svm import KeypairSigner
from x402.mechanisms.svm.exact import ExactSvmScheme

async def main():
    # 1. Initialize signer from base58 private key
    keypair = Keypair.from_base58_string(os.environ["SOLANA_SIGNER_KEY"])
    signer = KeypairSigner(keypair)

    # 2. Configure x402 client for Solana Mainnet
    NETWORK = "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp"
    x402 = x402Client()
    x402.register(NETWORK, ExactSvmScheme(signer=signer))
    x402.register_policy(prefer_network(NETWORK))

    # 3. The async client handles the 402 → sign → retry loop
    async with x402HttpxClient(x402, timeout=120.0) as client:
        resp = await client.post(
            "https://api.markdown2pdf.ai/markdown",
            json={
                "data": {"text_body": "# Hello from X402"},
                "options": {"document_name": "hello.pdf"},
            },
        )
        resp.raise_for_status()
        print("Paid tx proof:", resp.headers.get("payment-response"))
        print("Poll path:", resp.json()["path"])

if __name__ == "__main__":
    asyncio.run(main())

import httpx
from x402 import x402ClientSync
from x402.http.utils import decode_payment_required_header, encode_payment_signature_header
from x402.mechanisms.svm.exact import ExactSvmScheme

# 1. First request returns 402
r = httpx.post(url, json=payload)
assert r.status_code == 402
x402_offer = next(o for o in r.json()["offers"] if "x402" in o["payment_methods"])
accepts = x402_offer["metadata"]["x402"]["accepts"][0]  # amount, payTo, network, ...

# 2. Build + sign a PaymentPayload. x402ClientSync.create_payment_payload picks the
# requirement and calls ExactSvmScheme.create_payment_payload to sign the transfer.
# signer and NETWORK are set up as in the example above.
x402 = x402ClientSync()
x402.register(NETWORK, ExactSvmScheme(signer=signer))
payment_required = decode_payment_required_header(r.headers["payment-required"])
signed_payload = x402.create_payment_payload(payment_required)
payment_header = encode_payment_signature_header(signed_payload)

# 3. Re-POST with PAYMENT-SIGNATURE (X-PAYMENT also works)
r = httpx.post(url, json=payload, headers={"PAYMENT-SIGNATURE": payment_header})
r.raise_for_status()