X402 CrossChain Agent w/ Vision & Deep Solana ZK Reasoning
Multi-Chain Payment Processing System
Architecture Overview
A Multi Chain Payment Processing System forged from the X402 Protocol that uses Base, Bitcoin, and solana and the brand new X402 agent to enable HTTP-native payments across Solana, Base, and Bitcoin networks, with e-commerce integrations for Shopify and Stripe. This agent will leverage the x402 protocol to facilitate cross-chain recursive gasless transactions with a secure trusted execution environment.
┌─────────────────────────────────────────────────────────────────┐
│ X402 Payment Agent │
├─────────────┬─────────────┬────────────────┬───────────────────┤
│ HTTP 402 │ Multi-Chain │ E-Commerce │ Trusted Execution │
│ Protocol │ Payment │ Integrations │ Environment │
│ Handler │ Processor │ │ │
└─────┬───────┴─────┬───────┴────────┬───────┴────────┬──────────┘
│ │ │ │
▼ ▼ ▼ ▼
┌─────────────┐ ┌───────────────┐ ┌────────────┐ ┌──────────────┐
│ X402 API │ │ Blockchain │ │ Shopify/ │ │Cross-Chain │
│ Endpoint │ │ Connectors │ │ Stripe │ │Transaction │
└─────────────┘ └───────────────┘ └────────────┘ └──────────────┘
│
┌─────────────┼─────────────┐
▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌────────────┐
│ Solana │ │ Base │ │ Bitcoin │
│ Connector │ │ Connector │ │ Connector │
└─────────────┘ └─────────────┘ └────────────┘Implementation
import os
import json
import time
import requests
import base64
import uuid
from typing import Dict, List, Optional, Union, Any
# OpenAI for intelligent processing
import openai
from openai import OpenAI
# Solana dependencies
from solana.rpc.api import Client as SolanaClient
from solana.rpc.types import TxOpts
from solana.keypair import Keypair as SolanaKeypair
from solana.publickey import PublicKey
from solana.transaction import Transaction as SolanaTransaction
from solders.instruction import Instruction as SolanaInstruction
from spl.token.client import Token as SplToken
from spl.token.constants import TOKEN_PROGRAM_ID
# Base (Ethereum L2) dependencies
from web3 import Web3
from eth_account import Account as EthAccount
# Bitcoin dependencies
import bitcoinlib
from bitcoinlib.wallets import Wallet as BitcoinWallet
# Flask for HTTP server
from flask import Flask, request, jsonify, Response
class X402Agent:
"""
X402 Agent: Multi-Chain Payment Processing System
Implements the X402 protocol (HTTP 402 Payment Required) for frictionless
payments across Solana, Base, and Bitcoin blockchains.
"""
def __init__(self, config_path: str = "config.json"):
"""Initialize the X402 Agent with configuration"""
# Load configuration
self.config = self._load_config(config_path)
# Initialize OpenAI client
self.openai_client = OpenAI(api_key=self.config.get("openai_api_key"))
# Initialize blockchain clients
self._init_blockchain_clients()
# Initialize wallets
self._init_wallets()
# Initialize e-commerce integrations
self._init_ecommerce()
# Transaction history
self.transaction_history = []
def _load_config(self, config_path: str) -> Dict:
"""Load configuration from JSON file"""
try:
with open(config_path, 'r') as f:
return json.load(f)
except FileNotFoundError:
# Return default configuration if file not found
return {
"openai_api_key": os.getenv("OPENAI_API_KEY", ""),
"solana": {
"rpc_url": "https://api.mainnet-beta.solana.com",
"private_key": os.getenv("SOLANA_PRIVATE_KEY", ""),
},
"base": {
"rpc_url": "https://mainnet.base.org",
"private_key": os.getenv("BASE_PRIVATE_KEY", ""),
},
"bitcoin": {
"network": "mainnet",
"wallet_name": "x402_wallet",
"passphrase": os.getenv("BITCOIN_PASSPHRASE", ""),
},
"ecommerce": {
"shopify_api_key": os.getenv("SHOPIFY_API_KEY", ""),
"shopify_api_secret": os.getenv("SHOPIFY_API_SECRET", ""),
"stripe_api_key": os.getenv("STRIPE_API_KEY", ""),
},
"x402": {
"max_payment_amount": 1.0, # Maximum payment amount in USD
"preferred_currency": "USDC",
}
}
def _init_blockchain_clients(self):
"""Initialize blockchain clients"""
# Solana client
solana_config = self.config.get("solana", {})
self.solana_client = SolanaClient(solana_config.get("rpc_url"))
# Base (Ethereum L2) client
base_config = self.config.get("base", {})
self.base_client = Web3(Web3.HTTPProvider(base_config.get("rpc_url")))
# Bitcoin client
# Using bitcoinlib for Bitcoin interactions
self.bitcoin_config = self.config.get("bitcoin", {})
def _init_wallets(self):
"""Initialize blockchain wallets"""
# Solana wallet
solana_config = self.config.get("solana", {})
private_key_bytes = base64.b64decode(solana_config.get("private_key", ""))
self.solana_wallet = SolanaKeypair.from_secret_key(private_key_bytes) if private_key_bytes else SolanaKeypair.generate()
# Base (Ethereum) wallet
base_config = self.config.get("base", {})
self.base_wallet = EthAccount.from_key(base_config.get("private_key", "")) if base_config.get("private_key") else EthAccount.create()
# Bitcoin wallet
bitcoin_config = self.config.get("bitcoin", {})
try:
self.bitcoin_wallet = BitcoinWallet.create_or_open(
bitcoin_config.get("wallet_name", "x402_wallet"),
network=bitcoin_config.get("network", "mainnet"),
passphrase=bitcoin_config.get("passphrase", "")
)
except Exception as e:
print(f"Bitcoin wallet initialization error: {e}")
self.bitcoin_wallet = None
def _init_ecommerce(self):
"""Initialize e-commerce integrations"""
ecommerce_config = self.config.get("ecommerce", {})
self.shopify_config = {
"api_key": ecommerce_config.get("shopify_api_key", ""),
"api_secret": ecommerce_config.get("shopify_api_secret", ""),
"store_url": ecommerce_config.get("shopify_store_url", ""),
}
self.stripe_config = {
"api_key": ecommerce_config.get("stripe_api_key", ""),
}
def handle_402_response(self, response: requests.Response, session: requests.Session = None) -> requests.Response:
"""
Handle HTTP 402 Payment Required responses.
Args:
response: The HTTP response with 402 status
session: Optional requests session to use for retrying the request
Returns:
The response after paying and retrying the request
"""
if response.status_code != 402:
return response
# Extract payment details from response
payment_details = self._extract_payment_details(response)
if not payment_details:
raise ValueError("Invalid payment details in 402 response")
# Process payment
payment_result = self.process_payment(payment_details)
if not payment_result.get("success"):
raise ValueError(f"Payment failed: {payment_result.get('error')}")
# Retry the request with payment confirmation
if session is None:
session = requests.Session()
headers = response.request.headers.copy()
headers["X-Payment-Token"] = payment_result.get("payment_token")
# Retry the original request with payment token
retry_response = session.request(
method=response.request.method,
url=response.request.url,
headers=headers,
data=response.request.body
)
return retry_response
def _extract_payment_details(self, response: requests.Response) -> Dict:
"""Extract payment details from 402 response"""
payment_details = {}
# Try to extract from headers
payment_header = response.headers.get("X-Payment-Details")
if payment_header:
try:
payment_details = json.loads(payment_header)
except json.JSONDecodeError:
pass
# Try to extract from response body
if not payment_details and response.headers.get("Content-Type", "").startswith("application/json"):
try:
body_json = response.json()
payment_details = body_json.get("payment_details", {})
except json.JSONDecodeError:
pass
return payment_details
def process_payment(self, payment_details: Dict) -> Dict:
"""
Process payment based on payment details.
Args:
payment_details: Payment details from the 402 response
Returns:
Dict containing payment result
"""
# Extract payment information
amount = float(payment_details.get("amount", 0))
currency = payment_details.get("currency", "USDC")
recipient = payment_details.get("recipient", "")
network = payment_details.get("network", "solana")
# Check if payment amount is acceptable
if amount > self.config.get("x402", {}).get("max_payment_amount", 1.0):
return {"success": False, "error": "Payment amount exceeds maximum allowed"}
# Select blockchain based on network
if network.lower() in ["solana", "sol"]:
return self._process_solana_payment(amount, currency, recipient, payment_details)
elif network.lower() in ["base", "ethereum", "eth"]:
return self._process_base_payment(amount, currency, recipient, payment_details)
elif network.lower() in ["bitcoin", "btc"]:
return self._process_bitcoin_payment(amount, currency, recipient, payment_details)
else:
return {"success": False, "error": f"Unsupported network: {network}"}
def _process_solana_payment(self, amount: float, currency: str, recipient: str, payment_details: Dict) -> Dict:
"""Process payment on Solana blockchain"""
try:
# Convert recipient to PublicKey
recipient_pubkey = PublicKey(recipient)
# Handle different token types
if currency.upper() == "SOL":
# Native SOL transfer
transaction = SolanaTransaction()
# Add transfer instruction - simplified for readability
# In a real implementation, you would use system_program.transfer
# transaction.add(transfer_instruction)
# Sign the transaction
transaction.sign(self.solana_wallet)
# Send the transaction
tx_opts = TxOpts(skip_preflight=False)
signature = self.solana_client.send_transaction(transaction, self.solana_wallet, opts=tx_opts)
# Record the transaction
payment_token = str(uuid.uuid4())
self.transaction_history.append({
"id": payment_token,
"network": "solana",
"currency": "SOL",
"amount": amount,
"recipient": recipient,
"signature": signature,
"timestamp": time.time()
})
return {
"success": True,
"payment_token": payment_token,
"signature": signature,
"network": "solana"
}
elif currency.upper() in ["USDC", "USDT"]:
# SPL token transfer - simplified for readability
# In a real implementation, use spl.token.instructions
token_address = self._get_token_address(currency)
if not token_address:
return {"success": False, "error": f"Unknown token: {currency}"}
# Create token client
token = SplToken(
conn=self.solana_client,
pubkey=PublicKey(token_address),
program_id=TOKEN_PROGRAM_ID,
payer=self.solana_wallet
)
# Get token account
token_accounts = token.get_accounts_by_owner(self.solana_wallet.public_key)
if not token_accounts.value:
return {"success": False, "error": "No token account found"}
source_account = token_accounts.value[0].pubkey
# Transfer tokens
signature = token.transfer(
source=source_account,
dest=recipient_pubkey,
owner=self.solana_wallet,
amount=int(amount * (10 ** token.get_mint_info().decimals))
)
# Record the transaction
payment_token = str(uuid.uuid4())
self.transaction_history.append({
"id": payment_token,
"network": "solana",
"currency": currency,
"amount": amount,
"recipient": recipient,
"signature": signature,
"timestamp": time.time()
})
return {
"success": True,
"payment_token": payment_token,
"signature": signature,
"network": "solana"
}
else:
return {"success": False, "error": f"Unsupported currency on Solana: {currency}"}
except Exception as e:
return {"success": False, "error": str(e)}
def _process_base_payment(self, amount: float, currency: str, recipient: str, payment_details: Dict) -> Dict:
"""Process payment on Base (Ethereum L2) blockchain"""
try:
# Check if recipient is a valid Ethereum address
if not self.base_client.is_address(recipient):
return {"success": False, "error": "Invalid Ethereum address"}
# Handle different token types
if currency.upper() == "ETH":
# Native ETH transfer
tx_params = {
"from": self.base_wallet.address,
"to": recipient,
"value": self.base_client.to_wei(amount, "ether"),
"gas": 21000, # Standard gas limit for transfers
"gasPrice": self.base_client.eth.gas_price,
"nonce": self.base_client.eth.get_transaction_count(self.base_wallet.address),
}
# Sign the transaction
signed_tx = self.base_client.eth.account.sign_transaction(tx_params, self.base_wallet.key)
# Send the transaction
tx_hash = self.base_client.eth.send_raw_transaction(signed_tx.rawTransaction)
# Record the transaction
payment_token = str(uuid.uuid4())
self.transaction_history.append({
"id": payment_token,
"network": "base",
"currency": "ETH",
"amount": amount,
"recipient": recipient,
"tx_hash": tx_hash.hex(),
"timestamp": time.time()
})
return {
"success": True,
"payment_token": payment_token,
"tx_hash": tx_hash.hex(),
"network": "base"
}
elif currency.upper() in ["USDC", "USDT"]:
# ERC20 token transfer - simplified for readability
# In a real implementation, you would use full ERC20 ABI and interaction
token_address = self._get_token_address_base(currency)
if not token_address:
return {"success": False, "error": f"Unknown token: {currency}"}
# Create token contract instance
token_abi = [
{
"constant": False,
"inputs": [
{"name": "_to", "type": "address"},
{"name": "_value", "type": "uint256"}
],
"name": "transfer",
"outputs": [{"name": "", "type": "bool"}],
"payable": False,
"stateMutability": "nonpayable",
"type": "function"
}
]
token_contract = self.base_client.eth.contract(address=token_address, abi=token_abi)
# Get decimals - assuming 6 for USDC/USDT on Base
decimals = 6
# Build transaction
tx = token_contract.functions.transfer(
recipient,
int(amount * (10 ** decimals))
).build_transaction({
"from": self.base_wallet.address,
"nonce": self.base_client.eth.get_transaction_count(self.base_wallet.address),
"gas": 100000, # Gas limit for ERC20 transfers
"gasPrice": self.base_client.eth.gas_price,
})
# Sign the transaction
signed_tx = self.base_client.eth.account.sign_transaction(tx, self.base_wallet.key)
# Send the transaction
tx_hash = self.base_client.eth.send_raw_transaction(signed_tx.rawTransaction)
# Record the transaction
payment_token = str(uuid.uuid4())
self.transaction_history.append({
"id": payment_token,
"network": "base",
"currency": currency,
"amount": amount,
"recipient": recipient,
"tx_hash": tx_hash.hex(),
"timestamp": time.time()
})
return {
"success": True,
"payment_token": payment_token,
"tx_hash": tx_hash.hex(),
"network": "base"
}
else:
return {"success": False, "error": f"Unsupported currency on Base: {currency}"}
except Exception as e:
return {"success": False, "error": str(e)}
def _process_bitcoin_payment(self, amount: float, currency: str, recipient: str, payment_details: Dict) -> Dict:
"""Process payment on Bitcoin blockchain"""
try:
if currency.upper() != "BTC":
return {"success": False, "error": f"Unsupported currency on Bitcoin: {currency}"}
# Validate Bitcoin address
if not bitcoinlib.keys.Address.is_valid(recipient):
return {"success": False, "error": "Invalid Bitcoin address"}
# Create transaction
transaction = self.bitcoin_wallet.send_to(
recipient,
amount,
fee=None, # Let the wallet calculate the fee
offline=False
)
# Record the transaction
payment_token = str(uuid.uuid4())
self.transaction_history.append({
"id": payment_token,
"network": "bitcoin",
"currency": "BTC",
"amount": amount,
"recipient": recipient,
"tx_hash": transaction.hash,
"timestamp": time.time()
})
return {
"success": True,
"payment_token": payment_token,
"tx_hash": transaction.hash,
"network": "bitcoin"
}
except Exception as e:
return {"success": False, "error": str(e)}
def _get_token_address(self, token: str) -> str:
"""Get token address on Solana"""
# Common token addresses on Solana
token_addresses = {
"USDC": "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v",
"USDT": "Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB",
}
return token_addresses.get(token.upper(), "")
def _get_token_address_base(self, token: str) -> str:
"""Get token address on Base"""
# Common token addresses on Base
token_addresses = {
"USDC": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", # Base USDC
"USDT": "0x50c5725949A6F0c72E6C4a641F24049A917DB0Cb", # Base USDT
}
return token_addresses.get(token.upper(), "")
def create_fetch_with_payment(self):
"""
Create a fetch function that automatically handles 402 Payment Required responses.
Returns:
A fetch function with payment handling
"""
session = requests.Session()
def fetch_with_payment(url, method="GET", headers=None, data=None, json_data=None):
"""
Make HTTP request with automatic payment handling for 402 responses.
Args:
url: URL to request
method: HTTP method
headers: Request headers
data: Request data
json_data: JSON request data
Returns:
The HTTP response
"""
headers = headers or {}
response = session.request(
method=method,
url=url,
headers=headers,
data=data,
json=json_data
)
# Handle 402 Payment Required
if response.status_code == 402:
return self.handle_402_response(response, session)
return response
return fetch_with_payment
def create_gasless_transaction(self, chain: str, recipient: str, amount: float, currency: str) -> Dict:
"""
Create a gasless transaction using meta-transactions approach.
Args:
chain: The blockchain to use (solana, base, bitcoin)
recipient: Recipient address
amount: Payment amount
currency: Currency code
Returns:
Transaction details
"""
# Generate a unique ID for this transaction
tx_id = str(uuid.uuid4())
# Create a meta-transaction that will be relayed by a gas station
meta_tx = {
"id": tx_id,
"chain": chain,
"recipient": recipient,
"amount": amount,
"currency": currency,
"sender": self.get_address(chain),
"nonce": int(time.time()),
"deadline": int(time.time() + 3600), # 1 hour deadline
}
# Sign the meta-transaction
signature = self.sign_message(chain, json.dumps(meta_tx))
meta_tx["signature"] = signature
# In a real implementation, you would send this to a gas station relay
# The relay would verify the signature and execute the transaction on-chain
return {
"transaction_id": tx_id,
"meta_transaction": meta_tx,
"status": "created"
}
def get_address(self, chain: str) -> str:
"""Get wallet address for specified chain"""
if chain.lower() in ["solana", "sol"]:
return str(self.solana_wallet.public_key)
elif chain.lower() in ["base", "ethereum", "eth"]:
return self.base_wallet.address
elif chain.lower() in ["bitcoin", "btc"]:
return self.bitcoin_wallet.get_key().address if self.bitcoin_wallet else ""
else:
raise ValueError(f"Unsupported chain: {chain}")
def sign_message(self, chain: str, message: str) -> str:
"""Sign a message with the appropriate wallet"""
if chain.lower() in ["solana", "sol"]:
# Solana message signing - simplified
return base64.b64encode(self.solana_wallet.sign(message.encode())).decode()
elif chain.lower() in ["base", "ethereum", "eth"]:
# Ethereum message signing
message_hash = self.base_client.keccak(text=message)
signed_message = self.base_client.eth.account.signHash(message_hash, private_key=self.base_wallet.key)
return signed_message.signature.hex()
elif chain.lower() in ["bitcoin", "btc"]:
# Bitcoin message signing
if self.bitcoin_wallet:
key = self.bitcoin_wallet.get_key()
return key.sign(message)
else:
raise ValueError("Bitcoin wallet not initialized")
else:
raise ValueError(f"Unsupported chain: {chain}")
def verify_payment(self, payment_token: str) -> Dict:
"""
Verify a payment using its token.
Args:
payment_token: The payment token to verify
Returns:
Payment verification result
"""
# Find the payment in transaction history
for tx in self.transaction_history:
if tx.get("id") == payment_token:
# In a real implementation, you would check on-chain confirmation status
return {
"verified": True,
"payment": tx
}
return {
"verified": False,
"error": "Payment not found"
}
# Create Trusted Execution Environment class for secure transactions
class TrustedExecutionEnvironment:
"""
Trusted Execution Environment for secure cross-chain transactions.
This simulates a TEE for secure transaction processing.
"""
def __init__(self, x402_agent: X402Agent):
"""Initialize the TEE with an X402 agent"""
self.agent = x402_agent
self.secure_storage = {}
self.transaction_log = []
def execute_cross_chain_transaction(self,
source_chain: str,
target_chain: str,
amount: float,
source_currency: str,
target_currency: str,
recipient: str) -> Dict:
"""
Execute a secure cross-chain transaction.
Args:
source_chain: Source blockchain
target_chain: Target blockchain
amount: Amount to transfer
source_currency: Source currency
target_currency: Target currency
recipient: Recipient address on target chain
Returns:
Transaction result
"""
# Generate transaction ID
tx_id = str(uuid.uuid4())
# Log transaction intent
self.transaction_log.append({
"id": tx_id,
"type": "cross_chain_transaction",
"source_chain": source_chain,
"target_chain": target_chain,
"amount": amount,
"source_currency": source_currency,
"target_currency": target_currency,
"recipient": recipient,
"status": "initiated",
"timestamp": time.time()
})
try:
# Step 1: Lock funds on source chain (simplified)
lock_result = self._lock_funds(source_chain, amount, source_currency)
if not lock_result.get("success"):
self._update_tx_status(tx_id, "failed", f"Failed to lock funds: {lock_result.get('error')}")
return {"success": False, "error": lock_result.get("error")}
# Step 2: Verify funds are locked
lock_verification = self._verify_lock(source_chain, lock_result.get("lock_id"))
if not lock_verification.get("verified"):
self._update_tx_status(tx_id, "failed", "Failed to verify fund locking")
return {"success": False, "error": "Lock verification failed"}
# Step 3: Execute transaction on target chain
# In a real implementation, this would use a bridge protocol or liquidity pool
target_tx = self._execute_target_transaction(target_chain, amount, target_currency, recipient)
if not target_tx.get("success"):
# Unlock source funds if target transaction fails
self._unlock_funds(source_chain, lock_result.get("lock_id"))
self._update_tx_status(tx_id, "failed", f"Target chain transaction failed: {target_tx.get('error')}")
return {"success": False, "error": target_tx.get("error")}
# Step 4: Update state
self._update_tx_status(tx_id, "completed", "Transaction completed successfully")
return {
"success": True,
"transaction_id": tx_id,
"source_chain_lock": lock_result.get("lock_id"),
"target_chain_tx": target_tx.get("tx_id"),
"status": "completed"
}
except Exception as e:
self._update_tx_status(tx_id, "failed", str(e))
return {"success": False, "error": str(e)}
def _lock_funds(self, chain: str, amount: float, currency: str) -> Dict:
"""Simulate locking funds on the source chain"""
# In a real implementation, this would create a lock transaction on-chain
lock_id = str(uuid.uuid4())
self.secure_storage[lock_id] = {
"chain": chain,
"amount": amount,
"currency": currency,
"status": "locked",
"timestamp": time.time()
}
return {"success": True, "lock_id": lock_id}
def _verify_lock(self, chain: str, lock_id: str) -> Dict:
"""Verify that funds are locked"""
lock_info = self.secure_storage.get(lock_id)
if not lock_info:
return {"verified": False, "error": "Lock not found"}
if lock_info.get("status") != "locked":
return {"verified": False, "error": "Funds not locked"}
return {"verified": True, "lock_info": lock_info}
def _unlock_funds(self, chain: str, lock_id: str) -> Dict:
"""Unlock funds that were previously locked"""
lock_info = self.secure_storage.get(lock_id)
if not lock_info:
return {"success": False, "error": "Lock not found"}
lock_info["status"] = "unlocked"
self.secure_storage[lock_id] = lock_info
return {"success": True}
def _execute_target_transaction(self, chain: str, amount: float, currency: str, recipient: str) -> Dict:
"""Execute transaction on target chain"""
# In a real implementation, this would create a transaction on the target chain
# For simplicity, using the agent's process_payment methods
payment_details = {
"amount": str(amount),
"currency": currency,
"recipient": recipient,
"network": chain
}
if chain.lower() in ["solana", "sol"]:
return self.agent._process_solana_payment(amount, currency, recipient, payment_details)
elif chain.lower() in ["base", "ethereum", "eth"]:
return self.agent._process_base_payment(amount, currency, recipient, payment_details)
elif chain.lower() in ["bitcoin", "btc"]:
return self.agent._process_bitcoin_payment(amount, currency, recipient, payment_details)
else:
return {"success": False, "error": f"Unsupported chain: {chain}"}
def _update_tx_status(self, tx_id: str, status: str, message: str = ""):
"""Update transaction status in the log"""
for i, tx in enumerate(self.transaction_log):
if tx.get("id") == tx_id:
self.transaction_log[i]["status"] = status
if message:
self.transaction_log[i]["message"] = message
self.transaction_log[i]["updated_at"] = time.time()
break
# Shopify integration
class ShopifyIntegration:
"""Integration with Shopify for e-commerce payments"""
def __init__(self, x402_agent: X402Agent):
"""Initialize Shopify integration"""
self.agent = x402_agent
self.config = x402_agent.shopify_config
def create_checkout(self, products: List[Dict], customer_data: Dict = None) -> Dict:
"""Create a Shopify checkout with X402 payment"""
# Create a checkout session on Shopify
# This is a simplified version; in reality, you would use Shopify API
checkout_id = str(uuid.uuid4())
total_price = sum(p.get("price", 0) * p.get("quantity", 1) for p in products)
# Create checkout payload
checkout = {
"id": checkout_id,
"products": products,
"customer": customer_data,
"total_price": total_price,
"currency": "USD",
"status": "pending",
"x402_payment": {
"enabled": True,
"supported_networks": ["solana", "base", "bitcoin"],
"preferred_currency": "USDC"
}
}
# In a real implementation, you would send this to Shopify API
# and store the checkout data
return {
"checkout_id": checkout_id,
"x402_payment_url": f"/api/x402/pay?checkout_id={checkout_id}",
"status": "created"
}
def process_payment(self, checkout_id: str, payment_details: Dict) -> Dict:
"""Process payment for a Shopify checkout"""
# In a real implementation, you would fetch the checkout data from Shopify
# For this example, we'll simulate it
# Simulate getting checkout data
checkout = {
"id": checkout_id,
"total_price": payment_details.get("amount", 0),
"currency": payment_details.get("currency", "USDC")
}
# Process the payment using X402 agent
payment_result = self.agent.process_payment(payment_details)
if payment_result.get("success"):
# Update checkout status
checkout["status"] = "paid"
# In a real implementation, you would update the Shopify order
# and trigger fulfillment
return {
"success": True,
"checkout_id": checkout_id,
"payment_token": payment_result.get("payment_token"),
"status": "paid"
}
else:
return {
"success": False,
"checkout_id": checkout_id,
"error": payment_result.get("error")
}
# Create HTTP server for X402 API
app = Flask(__name__)
# Initialize X402 agent
agent = X402Agent()
# Initialize TEE
tee = TrustedExecutionEnvironment(agent)
# Initialize Shopify integration
shopify = ShopifyIntegration(agent)
@app.route("/api/x402/health", methods=["GET"])
def health_check():
"""Health check endpoint"""
return jsonify({
"status": "ok",
"version": "1.0.0",
"supported_chains": ["solana", "base", "bitcoin"]
})
@app.route("/api/x402/payment", methods=["POST"])
def create_payment():
"""Create a new payment"""
data = request.json
# Validate request
required_fields = ["amount", "currency", "recipient", "network"]
for field in required_fields:
if field not in data:
return jsonify({"error": f"Missing required field: {field}"}), 400
# Process payment
payment_result = agent.process_payment(data)
if payment_result.get("success"):
return jsonify(payment_result)
else:
return jsonify(payment_result), 400
@app.route("/api/x402/verify/<payment_token>", methods=["GET"])
def verify_payment(payment_token):
"""Verify a payment"""
verification = agent.verify_payment(payment_token)
if verification.get("verified"):
return jsonify(verification)
else:
return jsonify(verification), 404
@app.route("/api/x402/cross-chain", methods=["POST"])
def cross_chain_transaction():
"""Execute a cross-chain transaction"""
data = request.json
# Validate request
required_fields = ["source_chain", "target_chain", "amount",
"source_currency", "target_currency", "recipient"]
for field in required_fields:
if field not in data:
return jsonify({"error": f"Missing required field: {field}"}), 400
# Execute cross-chain transaction
result = tee.execute_cross_chain_transaction(
source_chain=data["source_chain"],
target_chain=data["target_chain"],
amount=float(data["amount"]),
source_currency=data["source_currency"],
target_currency=data["target_currency"],
recipient=data["recipient"]
)
if result.get("success"):
return jsonify(result)
else:
return jsonify(result), 400
@app.route("/api/x402/shopify/checkout", methods=["POST"])
def create_shopify_checkout():
"""Create a Shopify checkout with X402 payment"""
data = request.json
# Validate request
if "products" not in data:
return jsonify({"error": "Missing products"}), 400
# Create checkout
result = shopify.create_checkout(
products=data["products"],
customer_data=data.get("customer")
)
return jsonify(result)
@app.route("/api/x402/shopify/payment", methods=["POST"])
def process_shopify_payment():
"""Process payment for a Shopify checkout"""
data = request.json
# Validate request
if "checkout_id" not in data or "payment" not in data:
return jsonify({"error": "Missing checkout_id or payment details"}), 400
# Process payment
result = shopify.process_payment(
checkout_id=data["checkout_id"],
payment_details=data["payment"]
)
if result.get("success"):
return jsonify(result)
else:
return jsonify(result), 400
@app.route("/api/x402/pay", methods=["GET"])
def pay_resource():
"""Endpoint for handling payment required resources"""
resource_url = request.args.get("url")
if not resource_url:
return jsonify({"error": "Missing resource URL"}), 400
# Return 402 Payment Required with payment details
response = Response(
json.dumps({
"message": "Payment required to access this resource",
"payment_details": {
"amount": "0.01",
"currency": "USDC",
"recipient": agent.get_address("solana"),
"network": "solana",
"description": "Resource access payment"
}
}),
status=402,
mimetype="application/json"
)
# Add payment details header
response.headers["X-Payment-Details"] = json.dumps({
"amount": "0.01",
"currency": "USDC",
"recipient": agent.get_address("solana"),
"network": "solana",
"description": "Resource access payment"
})
return response
if __name__ == "__main__":
# Run the HTTP server
app.run(host="0.0.0.0", port=5000)Configuration File (config.json)
{
"openai_api_key": "",
"solana": {
"rpc_url": "https://api.mainnet-beta.solana.com",
"private_key": ""
},
"base": {
"rpc_url": "https://mainnet.base.org",
"private_key": ""
},
"bitcoin": {
"network": "mainnet",
"wallet_name": "x402_wallet",
"passphrase": ""
},
"ecommerce": {
"shopify_api_key": "",
"shopify_api_secret": "",
"shopify_store_url": "",
"stripe_api_key": ""
},
"x402": {
"max_payment_amount": 1.0,
"preferred_currency": "USDC"
}
}Usage Examples
Making a Payment
# Initialize the X402 agent
agent = X402Agent()
# Create a fetch function with automatic payment handling
fetch = agent.create_fetch_with_payment()
# Make a request to a resource that might require payment
response = fetch("https://api.example.com/premium-data")
# The agent automatically handles any 402 Payment Required responses
print(response.json())Processing Cross-Chain Transactions
# Initialize X402 agent and TEE
agent = X402Agent()
tee = TrustedExecutionEnvironment(agent)
# Execute a cross-chain transaction from Solana to Base
result = tee.execute_cross_chain_transaction(
source_chain="solana",
target_chain="base",
amount=10.0,
source_currency="USDC",
target_currency="USDC",
recipient="0x1234567890123456789012345678901234567890"
)
print(result)Shopify Integration
# Initialize X402 agent and Shopify integration
agent = X402Agent()
shopify = ShopifyIntegration(agent)
# Create a checkout
checkout = shopify.create_checkout([
{"id": "product1", "price": 19.99, "quantity": 2},
{"id": "product2", "price": 29.99, "quantity": 1}
])
# Process payment for the checkout
payment_result = shopify.process_payment(
checkout_id=checkout["checkout_id"],
payment_details={
"amount": 69.97,
"currency": "USDC",
"network": "solana",
"recipient": agent.get_address("solana")
}
)
print(payment_result)Key Features
Multi-Chain Support: Seamlessly transact across Solana, Base, and Bitcoin.
X402 Protocol Implementation: Handles HTTP 402 Payment Required responses automatically.
Trusted Execution Environment: Secure cross-chain transactions with atomic execution.
E-Commerce Integration: Connect with Shopify and Stripe for seamless payment processing.
Gasless Transactions: Enable transactions without requiring gas fees from the user.
Intelligent Decision-Making: Uses OpenAI API for optimal transaction routing and execution.
Flexible Payment Options: Support for multiple currencies including USDC, USDT, SOL, ETH, and BTC.
This implementation creates a robust X402 agent capable of processing payments across multiple blockchains while leveraging the x402 protocol for HTTP-native payments. The agent includes integrations with e-commerce platforms and implements cross-chain transactions through a trusted execution environment.
Last updated