Python

Complete examples for accepting x402 payments with Oura402 using Python.

Installation

pip install requests python-dotenv websocket-client
# For async support:
pip install aiohttp websockets

Basic 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 Oura402"""

    url = 'https://api.oura402.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_oura402(payment_proof: str, amount: int, recipient: str) -> bool:
    """Helper function to verify payments"""

    try:
        response = requests.post(
            'https://api.oura402.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_oura402(
        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.oura402.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.oura402.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.oura402.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.oura402.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.oura402.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.oura402.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 Oura402"""
    response = requests.post(
        'https://api.oura402.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.oura402.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

1. Environment Variables: Always store API keys in environment variables

2. Error Handling: Implement retry logic for network errors and rate limits

3. Logging: Log all payment verifications for debugging and reconciliation

4. Validation: Always validate payment amounts match expectations

5. Security: Never expose API keys in client-side code

6. Caching: Cache verified payments to avoid redundant verifications

7. 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