import time
from claw402 import X402Provider
async def call_tool_with_backoff(tool_name, params, max_retries=3):
for attempt in range(max_retries):
result = await provider.call_tool(tool_name, params)
if result.error and result.error.code == "RATE_LIMIT_EXCEEDED":
retry_after = result.error.retry_after or (2 ** attempt)
await asyncio.sleep(retry_after)
continue
return result
raise Exception("Max retries exceeded")
def check_rate_limit(result):
if result.rate_limit.remaining < result.rate_limit.limit * 0.1:
print(f"Warning: {result.rate_limit.remaining} tool calls remaining")
# Implement agent throttling if approaching limit
if result.rate_limit.remaining < 5:
# Delay next operation
time.sleep(2)
# Bad: Multiple sessions for same recipient
for invoice in invoices:
await provider.initialize_payment(invoice.amount, invoice.recipient)
# Good: Aggregate amounts per recipient
amounts_by_recipient = aggregate_invoices(invoices)
for recipient, total_amount in amounts_by_recipient.items():
await provider.initialize_payment(total_amount, recipient)
from datetime import datetime, timedelta
class BalanceCache:
def __init__(self, ttl_seconds=60):
self.cache = {}
self.ttl = timedelta(seconds=ttl_seconds)
async def get_balance(self, wallet, provider):
if wallet in self.cache:
cached_balance, timestamp = self.cache[wallet]
if datetime.now() - timestamp < self.ttl:
return cached_balance
balance = await provider.query_balance(wallet)
self.cache[wallet] = (balance, datetime.now())
return balance
import uuid
# Generate unique idempotency key per payment intent
idempotency_key = str(uuid.uuid4())
result = await provider.initialize_payment(
amount="0.1 SOL",
recipient="[email protected]",
idempotency_key=idempotency_key
)
# Safe to retry with same key - will not create duplicate session
if result.error:
result = await provider.initialize_payment(
amount="0.1 SOL",
recipient="[email protected]",
idempotency_key=idempotency_key # Same key
)