#!/usr/bin/env python3 """ Bridge402 Payment Integration Examples This file demonstrates how to integrate x402 payments with the Bridge402 API. It shows different payment scenarios and error handling. Note: These examples use mock payment data for demonstration purposes. In production, you would use the actual x402-python library to generate real EIP-3009 authorizations. """ import asyncio import json import base64 import logging from typing import Dict, Any, Optional import httpx # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class Bridge402PaymentClient: """Client for handling Bridge402 payments and API interactions""" def __init__(self, api_url: str): self.api_url = api_url.rstrip('/') def create_mock_payment(self, minutes: int, resource_path: str) -> str: """ Create a mock x402 payment for demonstration purposes. In production, you would use the x402-python library: ```python from x402 import X402Client from x402.schemes.exact import ExactScheme client = X402Client(private_key, network="base") payment = client.create_payment( scheme=ExactScheme(), amount=minutes * 0.002 * 1_000_000, # USDC atomic units asset="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", pay_to="0x745221d90e41193ee85c9d4679e40a9B57898A47", resource=f"{self.api_url}{resource_path}", description=f"Bridge402 access for {minutes} minutes" ) return payment.encode() ``` """ # Mock payment structure (NOT for production use) mock_payment = { "x402Version": 1, "scheme": "exact", "network": "base", "payload": { "signature": "0x" + "0" * 130, "authorization": "0x" + "0" * 130, } } payment_json = json.dumps(mock_payment) return base64.b64encode(payment_json.encode()).decode() async def get_payment_requirements(self, minutes: int, resource_path: str = "/stream") -> Dict[str, Any]: """Get payment requirements without providing payment""" logger.info(f"Getting payment requirements for {minutes} minutes...") async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.post( f"{self.api_url}/connect", params={"duration_min": minutes} # No X-PAYMENT header - this triggers 402 response ) if response.status_code == 402: data = response.json() logger.info("✅ Payment requirements received") return data else: logger.error(f"❌ Unexpected response: {response.status_code}") return {} except Exception as e: logger.error(f"❌ Error getting requirements: {e}") return {} async def connect_with_payment(self, minutes: int) -> Optional[Dict[str, Any]]: """Connect to Bridge402 with payment""" logger.info(f"Connecting with payment for {minutes} minutes...") payment_data = self.create_mock_payment(minutes, "/stream") headers = { "X-PAYMENT": payment_data, "Content-Type": "application/json" } async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.post( f"{self.api_url}/connect", params={"duration_min": minutes}, headers=headers ) if response.status_code == 200: data = response.json() logger.info("✅ Connection successful") return data elif response.status_code == 402: logger.warning("⚠️ Payment verification failed (expected with mock data)") return None else: logger.error(f"❌ Connection failed: {response.status_code} - {response.text}") return None except Exception as e: logger.error(f"❌ Connection error: {e}") return None async def extend_session(self, access_token: str, additional_minutes: int) -> Optional[Dict[str, Any]]: """Extend an existing session""" logger.info(f"Extending session by {additional_minutes} minutes...") payment_data = self.create_mock_payment(additional_minutes, "/stream") headers = { "X-SESSION": access_token, "X-PAYMENT": payment_data, "Content-Type": "application/json" } async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.post( f"{self.api_url}/extend", params={"duration_min": additional_minutes}, headers=headers ) if response.status_code == 200: data = response.json() logger.info("✅ Session extended successfully") return data elif response.status_code == 402: logger.warning("⚠️ Payment verification failed for extension") return None else: logger.error(f"❌ Extension failed: {response.status_code} - {response.text}") return None except Exception as e: logger.error(f"❌ Extension error: {e}") return None async def get_earnings_transcript(self, symbol: str, year: int, quarter: int) -> Optional[Dict[str, Any]]: """Get earnings transcript with payment""" logger.info(f"Getting {symbol} Q{quarter} {year} earnings transcript...") payment_data = self.create_mock_payment(1, "/earnings-transcript") # $0.05 for transcript headers = { "X-PAYMENT": payment_data, "Content-Type": "application/json" } async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.post( f"{self.api_url}/earnings-transcript", params={ "symbol": symbol, "year": year, "quarter": quarter }, headers=headers ) if response.status_code == 200: data = response.json() logger.info("✅ Transcript retrieved successfully") return data elif response.status_code == 402: logger.warning("⚠️ Payment verification failed for transcript") return None else: logger.error(f"❌ Transcript request failed: {response.status_code} - {response.text}") return None except Exception as e: logger.error(f"❌ Transcript error: {e}") return None async def demo_payment_flow(): """Demonstrate the complete payment flow""" client = Bridge402PaymentClient("http://localhost:8081") print("=" * 60) print("Bridge402 Payment Flow Demo") print("=" * 60) # 1. Get payment requirements print("\n1. Getting payment requirements...") requirements = await client.get_payment_requirements(10) if requirements: print(f" Rate: ${requirements['accepts'][0]['maxAmountRequired']} USDC") print(f" Description: {requirements['accepts'][0]['description']}") print(f" Network: {requirements['accepts'][0]['network']}") print(f" Asset: {requirements['accepts'][0]['asset']}") # 2. Connect with payment print("\n2. Connecting with payment...") session = await client.connect_with_payment(10) if session: print(f" Session ID: {session.get('access_token', 'N/A')[:20]}...") print(f" Expires at: {session.get('expires_at', 'N/A')}") # 3. Extend session if session: print("\n3. Extending session...") extended = await client.extend_session(session['access_token'], 5) if extended: print(f" New expiry: {extended.get('expires_at', 'N/A')}") # 4. Get earnings transcript print("\n4. Getting earnings transcript...") transcript = await client.get_earnings_transcript("AAPL", 2024, 1) if transcript: print(f" Symbol: {transcript.get('symbol')}") print(f" Quarter: Q{transcript.get('quarter')} {transcript.get('year')}") print(f" Provider: {transcript.get('metadata', {}).get('provider')}") async def demo_error_handling(): """Demonstrate error handling scenarios""" client = Bridge402PaymentClient("http://localhost:8081") print("\n" + "=" * 60) print("Error Handling Demo") print("=" * 60) # Test invalid session extension print("\n1. Testing invalid session extension...") invalid_token = base64.b64encode(json.dumps({ "session_id": "invalid_session", "expiry": 0 }).encode()).decode() result = await client.extend_session(invalid_token, 5) if not result: print(" ✅ Correctly handled invalid session") # Test invalid transcript parameters print("\n2. Testing invalid transcript parameters...") result = await client.get_earnings_transcript("INVALID", 1999, 5) if not result: print(" ✅ Correctly handled invalid parameters") if __name__ == "__main__": asyncio.run(demo_payment_flow()) asyncio.run(demo_error_handling())