"""Pandrosion agent client. No private-key handling; no payment on import. Read-only demo: python agent_client.py For purchases, explicitly call create_order(), save its receipt, and use your authorized wallet. pay_x402() accepts a signer supplied by the caller. """ from decimal import Decimal from hashlib import sha256 import json import os from pathlib import Path from urllib.parse import urlsplit from urllib.request import Request, build_opener, HTTPRedirectHandler ORIGIN = 'https://ivanbesevic.com' BASE = ORIGIN + '/api/payments' BASE_USDC = '0x833589fcd6edb6e08f4c7c32d4f71b54bda02913' RECIPIENT = '0xb74eecbf007c2c86258b85772a1c7efecb8bb966' class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None # Never forward a private token through a redirect. def checked_url(url): p = urlsplit(url) if p.scheme != 'https' or p.netloc != 'ivanbesevic.com' or not p.path.startswith('/api/payments/') or p.query or p.fragment: raise ValueError('Unexpected API destination') return url def request_json(url, body=None, token=None): headers = {'Accept':'application/json', 'Content-Type':'application/json'} if token: headers['Authorization'] = 'Bearer ' + token request = Request(checked_url(url), headers=headers, data=None if body is None else json.dumps(body).encode()) with build_opener(NoRedirect()).open(request, timeout=45) as response: return json.load(response) def save_private(path, data): # O_EXCL prevents replacing an existing receipt, including via symlink. fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, 'w') as file: json.dump(data, file, indent=2) def create_order(receipt_path, asset_id='base:USDC', protocol='x402-v2', max_usd='49.00', product_id='pandrosion-455'): if Path(receipt_path).exists(): raise ValueError('Receipt already exists; resume that order instead of creating another.') if protocol not in ('x402-v2', 'ivanb-invoice-v1', 'stripe-checkout-v1'): raise ValueError('Unknown protocol') if product_id not in ('pandrosion-455','pandrosion-601'): raise ValueError('Unknown product') product = request_json(BASE + '/v1/products/' + product_id) if Decimal(product['price_usd']) > Decimal(max_usd): raise ValueError('Published price exceeds the authorized budget') route = {'x402-v2':'/v1/x402/orders', 'ivanb-invoice-v1':'/v1/orders', 'stripe-checkout-v1':'/v1/stripe/orders'}[protocol] body = {'product_id':product_id} if protocol == 'stripe-checkout-v1' else {'asset_id':asset_id, 'product_id':product_id} order = request_json(BASE + route, body) save_private(receipt_path, order) # Must succeed before any signing. if order.get('product_id') != product_id: raise ValueError('Unexpected product; no payment initiated') if Decimal(order['price_usd']) > Decimal(max_usd): raise ValueError('Quoted price exceeds budget; no payment was initiated') return order def status(receipt_path): order = json.loads(Path(receipt_path).read_text()) return request_json(order['status_url'], token=order['access_token']) def submit_transaction(receipt_path, transaction_id): order = json.loads(Path(receipt_path).read_text()) if order.get('protocol') == 'x402-v2': raise ValueError('Use pay_x402 for this order') if order.get('protocol') == 'stripe-checkout-v1': raise ValueError('Complete the authorized Stripe checkout, then poll status; no transaction hash is needed') return request_json(order['submit_url'], {'transaction_id':transaction_id}, order['access_token']) def prepare_stripe(receipt_path, *, max_usd): """Returns a Stripe-hosted URL; never handles cards or pays automatically.""" receipt = json.loads(Path(receipt_path).read_text()) current = status(receipt_path) if current.get('protocol') != 'stripe-checkout-v1': raise ValueError('This is not a Stripe order') if not Decimal('0') < Decimal(current['price_usd']) <= Decimal(max_usd): raise ValueError('Payment exceeds the authorized budget') if current['state'] == 'paid': return current prepared = request_json(current['payment']['prepare_url'], {}, receipt['access_token']) target = urlsplit(prepared['payment'].get('checkout_url') or '') if target.scheme != 'https' or target.netloc != 'checkout.stripe.com': raise ValueError('Unexpected Stripe checkout destination') return prepared async def pay_x402(receipt_path, signer, *, max_usd): """Explicitly authorized payment only. Never pass an unlimited budget. Install x402[evm,httpx]==2.26.0 in the buyer's environment. The signer is an authorized wallet interface; this client neither obtains nor stores its key. """ import httpx from x402 import x402Client from x402.http.clients.httpx import x402AsyncTransport from x402.mechanisms.evm.exact.client import ExactEvmScheme receipt = json.loads(Path(receipt_path).read_text()) current = status(receipt_path) if current['state'] != 'awaiting_payment': return current # Pending/paid/reviewed/expired orders must not pay again. p = current['payment'] if current.get('protocol') != 'x402-v2' or p['network_id'] != 'eip155:8453' or p['contract'].lower() != BASE_USDC or p['recipient'].lower() != RECIPIENT: raise ValueError('Unexpected payment destination or protocol') if not Decimal('0') < Decimal(current['price_usd']) <= Decimal(max_usd): raise ValueError('Payment exceeds the authorized budget') target = checked_url(p['x402_url']) sdk = x402Client().register('eip155:8453', ExactEvmScheme(signer)) sdk.set_spend_controls({'allowed_assets':[{'network':'eip155:8453','asset':BASE_USDC, 'max_amount_per_payment':p['amount_atomic']}]}) async def validate_terms(context): terms = context.selected_requirements if terms.pay_to.lower() != RECIPIENT or terms.amount != p['amount_atomic'] or terms.asset.lower() != BASE_USDC or str(terms.network) != 'eip155:8453': raise ValueError('Payment terms changed; abort before signing') sdk.on_before_payment_creation(validate_terms) async with httpx.AsyncClient(transport=x402AsyncTransport(sdk), follow_redirects=False, timeout=55) as client: response = await client.get(target, headers={'Authorization':'Bearer '+receipt['access_token']}) response.raise_for_status() return response.json() # 202 means poll status(); it does not mean paid. def download(receipt_path, file_id, output_path): receipt = json.loads(Path(receipt_path).read_text()) current = status(receipt_path) if current['state'] != 'paid': raise ValueError('Payment is not finalized; poll status at >=20-second intervals') artifact = current['downloads'][file_id] request = Request(checked_url(artifact['url']), headers={'Authorization':'Bearer '+receipt['access_token']}) destination = Path(output_path) digest, total, created = sha256(), 0, False try: with build_opener(NoRedirect()).open(request, timeout=45) as response: with destination.open('xb') as output: created = True while chunk := response.read(65536): total += len(chunk) if total > artifact['bytes']: raise ValueError('Download exceeds advertised size') digest.update(chunk); output.write(chunk) if total != artifact['bytes'] or digest.hexdigest() != artifact['sha256']: raise ValueError('Downloaded file failed integrity verification') except Exception: if created: destination.unlink() # Only the incomplete file created by this call. raise return destination if __name__ == '__main__': product = request_json(BASE + '/v1/products/pandrosion-455') capabilities = request_json(BASE + '/v1/x402') print(json.dumps({'product':product['id'],'price_usd':product['price_usd'], 'x402_available':capabilities['available'],'no_payment_initiated':True}, indent=2))