Python
Complete examples for accepting x402 payments with Interface402 using Python.
Installation
pip install requests python-dotenv websocket-client
# For async support:
pip install aiohttp websocketsBasic Payment Verification
Verify a Payment
import requests
import os
from dotenv import load_dotenv
load_dotenv()
def verify_payment(payment_proof: str, amount: int, recipient: str) -> dict:
"""Verify an x402 payment using Interface402"""
url = 'https://api.interface402.dev/v1/payments/verify'
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {os.getenv("API_KEY")}'
}
payload = {
'payment_proof': payment_proof,
'amount': amount,
'recipient': recipient
}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
data = response.json()
if data['verified']:
print(f'✅ Payment verified!')
print(f'Transaction ID: {data["transaction_id"]}')
print(f'Amount: {data["amount"]}')
else:
print('❌ Payment invalid')
return data
except requests.exceptions.RequestException as e:
print(f'Error verifying payment: {e}')
raise
# Usage
result = verify_payment(
payment_proof='BASE64_ENCODED_PAYMENT_PROOF',
amount=1000000,
recipient='YOUR_WALLET_ADDRESS'
)Flask Integration
Protect an API Endpoint
from flask import Flask, request, jsonify
import requests
import os
app = Flask(__name__)
def verify_payment_with_interface402(payment_proof: str, amount: int, recipient: str) -> bool:
"""Helper function to verify payments"""
try:
response = requests.post(
'https://api.interface402.dev/v1/payments/verify',
headers={
'Content-Type': 'application/json',
'Authorization': f'Bearer {os.getenv("API_KEY")}'
},
json={
'payment_proof': payment_proof,
'amount': amount,
'recipient': recipient
}
)
response.raise_for_status()
data = response.json()
return data.get('verified', False)
except Exception as e:
print(f'Payment verification error: {e}')
return False
@app.route('/api/premium-content')
def premium_content():
"""Protected endpoint that requires x402 payment"""
payment_proof = request.headers.get('X-Payment-Proof')
REQUIRED_AMOUNT = 1000000 # 0.001 SOL in lamports
MY_WALLET = 'YOUR_WALLET_ADDRESS'
if not payment_proof:
return jsonify({
'error': 'Payment Required',
'amount': REQUIRED_AMOUNT,
'recipient': MY_WALLET,
'message': 'Please provide X-Payment-Proof header'
}), 402
# Verify payment
verified = verify_payment_with_interface402(
payment_proof,
REQUIRED_AMOUNT,
MY_WALLET
)
if verified:
return jsonify({
'data': 'Your premium content here',
'message': 'Access granted'
})
else:
return jsonify({
'error': 'Invalid payment',
'message': 'Payment verification failed'
}), 402
if __name__ == '__main__':
app.run(port=3000)Payment Middleware
from flask import request, jsonify
from functools import wraps
def require_payment(amount: int, recipient: str):
"""Decorator to protect endpoints with x402 payments"""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
payment_proof = request.headers.get('X-Payment-Proof')
if not payment_proof:
return jsonify({
'error': 'Payment Required',
'amount': amount,
'recipient': recipient
}), 402
try:
response = requests.post(
'https://api.interface402.dev/v1/payments/verify',
headers={
'Content-Type': 'application/json',
'Authorization': f'Bearer {os.getenv("API_KEY")}'
},
json={
'payment_proof': payment_proof,
'amount': amount,
'recipient': recipient
}
)
data = response.json()
if data.get('verified'):
# Attach payment info to request
request.payment = data
return f(*args, **kwargs)
else:
return jsonify({'error': 'Invalid payment'}), 402
except Exception as e:
return jsonify({'error': 'Payment verification failed'}), 500
return decorated_function
return decorator
# Usage
@app.route('/api/paid-content')
@require_payment(amount=1000000, recipient='YOUR_WALLET_ADDRESS')
def paid_content():
return jsonify({'data': 'Your paid content'})FastAPI Integration
Modern Async API
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
import httpx
import os
app = FastAPI()
class PaymentVerification(BaseModel):
payment_proof: str
amount: int
recipient: str
async def verify_payment_async(payment_proof: str, amount: int, recipient: str) -> bool:
"""Async payment verification"""
async with httpx.AsyncClient() as client:
try:
response = await client.post(
'https://api.interface402.dev/v1/payments/verify',
headers={
'Content-Type': 'application/json',
'Authorization': f'Bearer {os.getenv("API_KEY")}'
},
json={
'payment_proof': payment_proof,
'amount': amount,
'recipient': recipient
}
)
response.raise_for_status()
data = response.json()
return data.get('verified', False)
except Exception as e:
print(f'Error: {e}')
return False
@app.get('/api/premium-data')
async def get_premium_data(x_payment_proof: str = Header(None)):
"""Protected endpoint"""
REQUIRED_AMOUNT = 1000000
MY_WALLET = 'YOUR_WALLET_ADDRESS'
if not x_payment_proof:
raise HTTPException(
status_code=402,
detail={
'error': 'Payment Required',
'amount': REQUIRED_AMOUNT,
'recipient': MY_WALLET
}
)
verified = await verify_payment_async(
x_payment_proof,
REQUIRED_AMOUNT,
MY_WALLET
)
if not verified:
raise HTTPException(status_code=402, detail='Invalid payment')
return {'data': 'Your premium data'}
@app.post('/api/verify-payment')
async def verify_payment_endpoint(verification: PaymentVerification):
"""Endpoint to verify a payment"""
verified = await verify_payment_async(
verification.payment_proof,
verification.amount,
verification.recipient
)
if verified:
return {'success': True, 'message': 'Payment verified'}
else:
raise HTTPException(status_code=402, detail='Payment verification failed')WebSocket for Real-Time Notifications
Basic WebSocket Connection
import websocket
import json
import threading
class PaymentStream:
def __init__(self, api_key: str, wallet_address: str):
self.api_key = api_key
self.wallet_address = wallet_address
self.ws = None
def on_message(self, ws, message):
"""Handle incoming payment notifications"""
payment = json.loads(message)
print(f'💰 Payment received!')
print(f' Amount: {payment["amount"]}')
print(f' From: {payment["sender"]}')
print(f' Transaction: {payment["transaction_id"]}')
# Do something with the payment
# e.g., unlock content, send email, etc.
def on_error(self, ws, error):
print(f'Error: {error}')
def on_close(self, ws, close_status_code, close_msg):
print('Connection closed, reconnecting...')
# Reconnect after 5 seconds
import time
time.sleep(5)
self.connect()
def on_open(self, ws):
print('Connected to payment stream')
# Authenticate
ws.send(json.dumps({
'type': 'authenticate',
'apiKey': self.api_key
}))
# Subscribe to payments for your wallet
ws.send(json.dumps({
'type': 'subscribe',
'wallet': self.wallet_address
}))
def connect(self):
"""Connect to payment stream"""
websocket.enableTrace(False)
self.ws = websocket.WebSocketApp(
'wss://api.interface402.dev/v1/payments/stream',
on_open=self.on_open,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close
)
# Run in background thread
wst = threading.Thread(target=self.ws.run_forever)
wst.daemon = True
wst.start()
def disconnect(self):
if self.ws:
self.ws.close()
# Usage
stream = PaymentStream(
api_key=os.getenv('API_KEY'),
wallet_address='YOUR_WALLET_ADDRESS'
)
stream.connect()
# Keep running
import time
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
stream.disconnect()Async WebSocket with websockets
import asyncio
import websockets
import json
import os
async def stream_payments(api_key: str, wallet_address: str):
"""Stream payments using async websockets"""
uri = 'wss://api.interface402.dev/v1/payments/stream'
async with websockets.connect(uri) as websocket:
# Authenticate
await websocket.send(json.dumps({
'type': 'authenticate',
'apiKey': api_key
}))
# Subscribe to payments
await websocket.send(json.dumps({
'type': 'subscribe',
'wallet': wallet_address
}))
print('Listening for payments...')
# Listen for messages
async for message in websocket:
payment = json.loads(message)
print(f'Payment: {payment["amount"]} from {payment["sender"]}')
# Usage
asyncio.run(stream_payments(
os.getenv('API_KEY'),
'YOUR_WALLET_ADDRESS'
))Error Handling
Robust Error Handling with Retry
import time
from typing import Optional
class PaymentVerificationError(Exception):
def __init__(self, code: str, message: str, details: dict = None):
self.code = code
self.message = message
self.details = details or {}
super().__init__(self.message)
def verify_payment_with_retry(
payment_proof: str,
amount: int,
recipient: str,
max_retries: int = 3
) -> Optional[dict]:
"""Verify payment with automatic retry logic"""
for attempt in range(max_retries):
try:
response = requests.post(
'https://api.interface402.dev/v1/payments/verify',
headers={
'Content-Type': 'application/json',
'Authorization': f'Bearer {os.getenv("API_KEY")}'
},
json={
'payment_proof': payment_proof,
'amount': amount,
'recipient': recipient
}
)
data = response.json()
if not response.ok:
error = data.get('error', {})
raise PaymentVerificationError(
code=error.get('code', 'UNKNOWN_ERROR'),
message=error.get('message', 'Verification failed'),
details=error.get('details', {})
)
return data
except PaymentVerificationError as e:
# Don't retry on certain errors
if e.code in ['INVALID_PAYMENT_PROOF', 'INVALID_AMOUNT']:
print(f'Error: {e.message}')
raise
# Retry on rate limits or network errors
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f'Retrying in {wait_time}s...')
time.sleep(wait_time)
continue
raise
except requests.RequestException as e:
if attempt < max_retries - 1:
time.sleep(2 ** attempt)
continue
raise
raise Exception(f'Failed after {max_retries} attempts')
# Usage
try:
result = verify_payment_with_retry(
payment_proof='BASE64_ENCODED_PAYMENT_PROOF',
amount=1000000,
recipient='YOUR_WALLET_ADDRESS'
)
print('Payment verified:', result['transaction_id'])
except PaymentVerificationError as e:
print(f'Verification failed: {e.message}')Django Integration
Django View with Payment Protection
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
import requests
import os
@require_http_methods(["GET"])
def premium_content_view(request):
"""Django view protected by x402 payment"""
payment_proof = request.headers.get('X-Payment-Proof')
REQUIRED_AMOUNT = 1000000
MY_WALLET = 'YOUR_WALLET_ADDRESS'
if not payment_proof:
return JsonResponse({
'error': 'Payment Required',
'amount': REQUIRED_AMOUNT,
'recipient': MY_WALLET
}, status=402)
try:
response = requests.post(
'https://api.interface402.dev/v1/payments/verify',
headers={
'Content-Type': 'application/json',
'Authorization': f'Bearer {os.getenv("API_KEY")}'
},
json={
'payment_proof': payment_proof,
'amount': REQUIRED_AMOUNT,
'recipient': MY_WALLET
}
)
data = response.json()
if data.get('verified'):
return JsonResponse({
'data': 'Your premium content',
'message': 'Access granted'
})
else:
return JsonResponse({
'error': 'Invalid payment'
}, status=402)
except Exception as e:
return JsonResponse({
'error': 'Payment verification failed'
}, status=500)Complete Example: Payment-Protected API
from flask import Flask, request, jsonify
import requests
import os
from datetime import datetime
app = Flask(__name__)
# In-memory cache (use Redis in production)
verified_payments = set()
def verify_payment(payment_proof: str, amount: int, recipient: str) -> dict:
"""Verify payment with Interface402"""
response = requests.post(
'https://api.interface402.dev/v1/payments/verify',
headers={
'Content-Type': 'application/json',
'Authorization': f'Bearer {os.getenv("API_KEY")}'
},
json={
'payment_proof': payment_proof,
'amount': amount,
'recipient': recipient
}
)
response.raise_for_status()
return response.json()
@app.route('/api/verify-payment', methods=['POST'])
def verify_payment_endpoint():
"""Verify a payment"""
data = request.json
payment_proof = data.get('payment_proof')
amount = data.get('amount')
try:
result = verify_payment(
payment_proof,
amount,
'YOUR_WALLET_ADDRESS'
)
if result['verified']:
# Cache the transaction ID
verified_payments.add(result['transaction_id'])
return jsonify({
'success': True,
'transaction_id': result['transaction_id']
})
return jsonify({
'success': False,
'error': 'Payment verification failed'
}), 402
except Exception as e:
return jsonify({
'success': False,
'error': str(e)
}), 500
@app.route('/api/content/<content_id>')
def get_content(content_id):
"""Get paid content"""
transaction_id = request.headers.get('X-Transaction-Id')
if not transaction_id or transaction_id not in verified_payments:
return jsonify({
'error': 'Payment Required',
'message': 'Valid payment required to access this content'
}), 402
# Return the content
return jsonify({
'content_id': content_id,
'content': 'Your premium content here',
'accessed_at': datetime.utcnow().isoformat()
})
if __name__ == '__main__':
app.run(port=3000)Async/Await Examples
Async Payment Verification
import asyncio
import aiohttp
import os
async def verify_payment_async(
payment_proof: str,
amount: int,
recipient: str
) -> dict:
"""Async payment verification"""
url = 'https://api.interface402.dev/v1/payments/verify'
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {os.getenv("API_KEY")}'
}
payload = {
'payment_proof': payment_proof,
'amount': amount,
'recipient': recipient
}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=payload) as response:
data = await response.json()
return data
async def main():
"""Verify multiple payments concurrently"""
payments = [
('PROOF_1', 1000000, 'WALLET_1'),
('PROOF_2', 2000000, 'WALLET_2'),
('PROOF_3', 3000000, 'WALLET_3')
]
tasks = [
verify_payment_async(proof, amount, wallet)
for proof, amount, wallet in payments
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f'Payment {i+1} failed: {result}')
else:
print(f'Payment {i+1} verified: {result["transaction_id"]}')
# Run
asyncio.run(main())Best Practices
Environment Variables: Always store API keys in environment variables
Error Handling: Implement retry logic for network errors and rate limits
Logging: Log all payment verifications for debugging and reconciliation
Validation: Always validate payment amounts match expectations
Security: Never expose API keys in client-side code
Caching: Cache verified payments to avoid redundant verifications
Testing: Test with small amounts before going live
Next Steps
See JavaScript examples for JavaScript/TypeScript integration
Check out the API Reference for all endpoints
Read about error handling best practices
Last updated