main.py

"""x402 requests client example - sync HTTP with automatic payment handling."""

import os
import sys

from dotenv import load_dotenv
from eth_account import Account

from x402 import x402ClientSync
from x402.http import x402HTTPClientSync
from x402.http.clients import x402_requests
from x402.mechanisms.evm import EthAccountSigner
from x402.mechanisms.evm.exact.register import register_exact_evm_client
from x402.mechanisms.svm import KeypairSigner
from x402.mechanisms.svm.exact.register import register_exact_svm_client

# Load environment variables
load_dotenv()


def validate_environment() -> tuple[str | None, str | None, str, str]:
    """Validate required environment variables.

    Returns:
        Tuple of (evm_private_key, svm_private_key, base_url, endpoint_path).

    Raises:
        SystemExit: If required environment variables are missing.
    """
    evm_private_key = os.getenv("EVM_PRIVATE_KEY")
    svm_private_key = os.getenv("SVM_PRIVATE_KEY")
    base_url = os.getenv("RESOURCE_SERVER_URL")
    endpoint_path = os.getenv("ENDPOINT_PATH")

    missing = []
    if not evm_private_key and not svm_private_key:
        missing.append("EVM_PRIVATE_KEY or SVM_PRIVATE_KEY")
    if not base_url:
        missing.append("RESOURCE_SERVER_URL")
    if not endpoint_path:
        missing.append("ENDPOINT_PATH")

    if missing:
        print(f"Error: Missing required environment variables: {', '.join(missing)}")
        print("Please copy .env-local to .env and fill in the values.")
        sys.exit(1)

    return evm_private_key, svm_private_key, base_url, endpoint_path


def main() -> None:
    """Main entry point demonstrating requests with x402 payments."""
    # Validate environment
    evm_private_key, svm_private_key, base_url, endpoint_path = validate_environment()

    # Create x402 client (sync variant for requests)
    client = x402ClientSync()

    # Register EVM payment scheme if private key provided
    if evm_private_key:
        account = Account.from_key(evm_private_key)
        register_exact_evm_client(client, EthAccountSigner(account))
        print(f"Initialized EVM account: {account.address}")

    # Register SVM payment scheme if private key provided
    if svm_private_key:
        svm_signer = KeypairSigner.from_base58(svm_private_key)
        register_exact_svm_client(client, svm_signer)
        print(f"Initialized SVM account: {svm_signer.address}")

    # Create HTTP client helper for payment response extraction (sync)
    http_client = x402HTTPClientSync(client)

    # Build full URL
    url = f"{base_url}{endpoint_path}"
    print(f"Making request to: {url}\n")

    # Make request using context manager for proper cleanup
    with x402_requests(client) as session:
        response = session.get(url)

        print(f"Response status: {response.status_code}")
        print(f"Response body: {response.text}")

        # Extract and print payment response if present
        try:
            settle_response = http_client.get_payment_settle_response(
                lambda name: response.headers.get(name)
            )
            print(f"\nPayment response: {settle_response.model_dump_json(indent=2)}")
        except ValueError:
            print("\nNo payment response header found")


if __name__ == "__main__":
    main()