Python

Complete examples for building autonomous AI agents with x402 payment capabilities using OpenClaw and Claw402.

Installation

# Install OpenClaw and Claw402 provider
pip install openclaw claw402

# For Solana wallet management
pip install solana

Quick Start

Configure Agent Wallet

# Generate Solana wallet for agent
solana-keygen new -o ~/.config/solana/agent.json

# Fund wallet with SOL for transaction fees
solana airdrop 1 ~/.config/solana/agent.json --url devnet

# Set environment variables
export CLAW402_WALLET_PATH=~/.config/solana/agent.json
export CLAW402_NETWORK=devnet

Basic Agent with Payment Capabilities

from openclaw import Agent
from claw402 import X402Provider
import os

# Create agent with x402 payment provider
agent = Agent(
    name="payment_assistant",
    providers=[
        X402Provider(
            network="devnet",
            wallet_path=os.getenv("CLAW402_WALLET_PATH"),
            approval_policy="threshold:0.01_SOL"
        )
    ]
)

# Agent autonomously executes payment from natural language
response = agent.run(
    "Pay the invoice from Acme Corp for 0.05 SOL"
)

print(f"Transaction: {response.tx_signature}")
print(f"Status: {response.status}")

OpenClaw Provider Integration

Initialize Provider with Approval Policies

from claw402 import X402Provider, ApprovalPolicy

# Configure multi-tier approval policies
approval_policy = ApprovalPolicy(
    auto_approve_threshold="0.01 SOL",
    require_confirmation_threshold="0.1 SOL",
    block_above="1.0 SOL"
)

provider = X402Provider(
    network="mainnet-beta",
    wallet_path="~/.config/solana/agent.json",
    approval_policy=approval_policy,
    rpc_url="https://api.mainnet-beta.solana.com"
)

agent = Agent(
    name="treasury_agent",
    providers=[provider]
)

Using MCP Tools Directly

from claw402 import X402Provider

provider = X402Provider(
    network="mainnet-beta",
    wallet_path="~/.config/solana/agent.json"
)

# Initialize payment session
session = await provider.initialize_payment(
    amount="0.1 SOL",
    recipient="[email protected]",
    purpose="API access subscription",
    settlement_constraint="immediate"
)

print(f"Session ID: {session.session_id}")
print(f"Payment address: {session.payment_address}")

# Authorize transaction
tx_result = await provider.authorize_transaction(
    session_id=session.session_id,
    approval_token="auto"  # Requires approval policy
)

print(f"Transaction signature: {tx_result.tx_signature}")

# Verify settlement
settlement = await provider.verify_settlement(
    tx_signature=tx_result.tx_signature
)

print(f"Confirmations: {settlement.confirmations}")
print(f"Settlement state: {settlement.state}")

Autonomous Agent Use Cases

1. Invoice Settlement Agent

from openclaw import Agent
from claw402 import X402Provider
import asyncio

class InvoiceAgent:
    def __init__(self):
        self.agent = Agent(
            name="invoice_settler",
            providers=[
                X402Provider(
                    network="mainnet-beta",
                    wallet_path="~/.config/solana/agent.json",
                    approval_policy="threshold:100_USDC"
                )
            ]
        )

    async def process_invoice(self, invoice):
        """Autonomously process and pay invoice"""

        # Agent validates invoice and determines payment intent
        analysis = self.agent.run(
            f"""
            Review this invoice and determine if payment is warranted:

            Vendor: {invoice.vendor}
            Amount: {invoice.amount}
            Due Date: {invoice.due_date}
            Services: {invoice.description}

            If valid, initiate payment using x402 protocol.
            """
        )

        if analysis.payment_initiated:
            print(f"✅ Invoice {invoice.id} paid")
            print(f"   Transaction: {analysis.tx_signature}")
            return analysis.tx_signature
        else:
            print(f"⚠️  Invoice {invoice.id} requires manual review")
            print(f"   Reason: {analysis.review_reason}")
            return None

    async def batch_process(self, invoices):
        """Process multiple invoices concurrently"""
        tasks = [self.process_invoice(inv) for inv in invoices]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

# Usage
agent = InvoiceAgent()

# Mock invoice data
invoices = [
    Invoice(id="INV-001", vendor="Acme Corp", amount="250 USDC", ...),
    Invoice(id="INV-002", vendor="Beta Inc", amount="500 USDC", ...),
]

results = await agent.batch_process(invoices)

2. SaaS Subscription Agent

from openclaw import Agent
from claw402 import X402Provider
from datetime import datetime, timedelta

class SubscriptionAgent:
    def __init__(self):
        self.provider = X402Provider(
            network="mainnet-beta",
            wallet_path="~/.config/solana/agent.json",
            approval_policy="threshold:50_USDC"
        )

        self.agent = Agent(
            name="subscription_manager",
            providers=[self.provider]
        )

    async def check_and_renew_subscriptions(self):
        """Check active subscriptions and renew if expiring soon"""

        # Query agent's active subscriptions
        subscriptions = await self.get_active_subscriptions()

        for sub in subscriptions:
            if self.should_renew(sub):
                await self.renew_subscription(sub)

    def should_renew(self, subscription):
        """Determine if subscription should be renewed"""
        days_until_expiry = (subscription.expires_at - datetime.now()).days
        return days_until_expiry <= 7  # Renew within 7 days

    async def renew_subscription(self, subscription):
        """Autonomously renew subscription"""

        response = self.agent.run(
            f"""
            Renew the subscription for {subscription.service_name}.

            Cost: {subscription.monthly_cost}
            Provider: {subscription.provider_wallet}
            Duration: 1 month

            Use x402 protocol to process payment.
            """
        )

        if response.success:
            print(f"✅ Renewed {subscription.service_name}")
            print(f"   Valid until: {response.new_expiry_date}")
        else:
            print(f"❌ Failed to renew {subscription.service_name}")

# Schedule agent to run daily
agent = SubscriptionAgent()
await agent.check_and_renew_subscriptions()

3. Resource Allocation Agent

from openclaw import Agent
from claw402 import X402Provider

class ResourceAgent:
    def __init__(self):
        self.agent = Agent(
            name="resource_allocator",
            providers=[
                X402Provider(
                    network="mainnet-beta",
                    wallet_path="~/.config/solana/agent.json",
                    approval_policy="threshold:10_USDC"
                )
            ]
        )

    async def allocate_compute_resources(self, workload_specs):
        """Autonomously provision and pay for cloud compute"""

        response = self.agent.run(
            f"""
            I need to provision cloud compute resources for the following workload:

            CPU: {workload_specs.cpu_cores} cores
            Memory: {workload_specs.memory_gb} GB
            Duration: {workload_specs.duration_hours} hours
            Provider: {workload_specs.preferred_provider}

            1. Estimate the cost using x402.estimate_fees
            2. Check my balance using x402.query_balance
            3. If sufficient funds, initialize payment using x402.initialize_payment
            4. Authorize the transaction

            Report the provisioned resource details.
            """
        )

        if response.success:
            print(f"✅ Resources provisioned")
            print(f"   Instance ID: {response.instance_id}")
            print(f"   Cost: {response.total_cost}")
            print(f"   TX: {response.tx_signature}")
            return response.instance_id
        else:
            print(f"❌ Provisioning failed: {response.error}")
            return None

# Usage
agent = ResourceAgent()
instance_id = await agent.allocate_compute_resources(
    WorkloadSpecs(
        cpu_cores=8,
        memory_gb=32,
        duration_hours=24,
        preferred_provider="aws_sol_compute"
    )
)

Standalone MCP Server Mode

Launch Claw402 as MCP Server

# Start Claw402 MCP server
claw402 serve \
  --network mainnet-beta \
  --wallet ~/.config/solana/agent.json \
  --port 3402 \
  --approval-policy-file ./approval-policy.yaml

Connect Custom Agent to MCP Server

from mcp import MCPClient

# Connect to Claw402 MCP server
client = MCPClient("http://localhost:3402")

# List available tools
tools = await client.list_tools()
print(f"Available tools: {[tool.name for tool in tools]}")

# Call MCP tool directly
result = await client.call_tool(
    "x402.initialize_payment",
    {
        "amount": "0.1 SOL",
        "recipient": "[email protected]",
        "purpose": "API access subscription"
    }
)

print(f"Session ID: {result.session_id}")

WebSocket Transaction Monitoring

Real-Time Transaction Updates

import asyncio
import websockets
import json

async def monitor_transactions(wallet_address, api_key):
    """Monitor agent wallet for real-time transaction updates"""

    uri = "wss://api.claw402.xyz/v1/transactions/stream"

    async with websockets.connect(uri) as websocket:
        # Authenticate
        await websocket.send(json.dumps({
            "type": "authenticate",
            "api_key": api_key
        }))

        # Subscribe to wallet
        await websocket.send(json.dumps({
            "type": "subscribe",
            "wallet": wallet_address
        }))

        print(f"Monitoring transactions for {wallet_address}")

        # Listen for updates
        async for message in websocket:
            tx = json.loads(message)
            print(f"Transaction update: {tx['signature']}")
            print(f"  Status: {tx['status']}")
            print(f"  Confirmations: {tx['confirmations']}")

            if tx['status'] == 'finalized':
                print(f"  ✅ Transaction finalized")

# Usage
await monitor_transactions(
    wallet_address="YOUR_AGENT_WALLET",
    api_key="YOUR_API_KEY"
)

Error Handling for Agent Workflows

Robust Transaction Execution

from claw402 import X402Provider, TransactionError
import asyncio

class ResilientAgent:
    def __init__(self):
        self.provider = X402Provider(
            network="mainnet-beta",
            wallet_path="~/.config/solana/agent.json"
        )

    async def execute_payment_with_retry(
        self,
        amount,
        recipient,
        max_retries=3
    ):
        """Execute payment with automatic retry and error handling"""

        for attempt in range(max_retries):
            try:
                # Initialize payment session
                session = await self.provider.initialize_payment(
                    amount=amount,
                    recipient=recipient,
                    idempotency_key=f"payment_{recipient}_{amount}"
                )

                # Authorize transaction
                tx = await self.provider.authorize_transaction(
                    session_id=session.session_id
                )

                # Wait for settlement
                settlement = await self.wait_for_settlement(tx.tx_signature)

                if settlement.state == "finalized":
                    return tx.tx_signature
                else:
                    raise TransactionError("Settlement failed")

            except TransactionError as e:
                if e.code == "INSUFFICIENT_BALANCE":
                    print(f"❌ Insufficient balance")
                    return None

                if e.code == "RATE_LIMIT_EXCEEDED":
                    retry_after = e.retry_after or (2 ** attempt)
                    print(f"⏳ Rate limited, retrying in {retry_after}s")
                    await asyncio.sleep(retry_after)
                    continue

                if attempt < max_retries - 1:
                    print(f"⚠️  Attempt {attempt + 1} failed, retrying...")
                    await asyncio.sleep(2 ** attempt)
                    continue

                print(f"❌ Payment failed after {max_retries} attempts: {e}")
                return None

    async def wait_for_settlement(self, tx_signature, timeout=60):
        """Wait for transaction to finalize"""
        start = asyncio.get_event_loop().time()

        while asyncio.get_event_loop().time() - start < timeout:
            settlement = await self.provider.verify_settlement(tx_signature)

            if settlement.confirmations >= 32:  # Solana finality
                return settlement

            await asyncio.sleep(2)

        raise TransactionError("Settlement timeout")

# Usage
agent = ResilientAgent()
tx_signature = await agent.execute_payment_with_retry(
    amount="0.05 SOL",
    recipient="[email protected]"
)

Advanced Agent Patterns

Multi-Agent Coordination

from openclaw import Agent
from claw402 import X402Provider
import asyncio

class TreasuryCoordinator:
    def __init__(self):
        # Create specialized agents
        self.invoice_agent = Agent(
            name="invoice_processor",
            providers=[X402Provider(
                network="mainnet-beta",
                wallet_path="~/.config/solana/invoice_agent.json",
                approval_policy="threshold:100_USDC"
            )]
        )

        self.subscription_agent = Agent(
            name="subscription_manager",
            providers=[X402Provider(
                network="mainnet-beta",
                wallet_path="~/.config/solana/subscription_agent.json",
                approval_policy="threshold:50_USDC"
            )]
        )

        self.resource_agent = Agent(
            name="resource_allocator",
            providers=[X402Provider(
                network="mainnet-beta",
                wallet_path="~/.config/solana/resource_agent.json",
                approval_policy="threshold:25_USDC"
            )]
        )

    async def coordinate_treasury_operations(self):
        """Run all treasury agents concurrently"""

        tasks = [
            self.invoice_agent.run("Process pending invoices"),
            self.subscription_agent.run("Check and renew subscriptions"),
            self.resource_agent.run("Optimize resource allocation")
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        for agent_name, result in zip(
            ["invoice", "subscription", "resource"],
            results
        ):
            if isinstance(result, Exception):
                print(f"❌ {agent_name} agent failed: {result}")
            else:
                print(f"✅ {agent_name} agent completed")

# Usage
coordinator = TreasuryCoordinator()
await coordinator.coordinate_treasury_operations()

Best Practices

1. Approval Policies: Always configure approval policies to limit agent transaction exposure

2. Idempotency Keys: Use idempotency keys for all payment operations to prevent duplicates

3. Error Handling: Implement retry logic with exponential backoff for transient errors

4. Monitoring: Set up WebSocket streams to monitor agent transactions in real-time

5. Wallet Security: Store agent wallet keys in secure key management systems (HSM, Vault)

6. Audit Logging: Log all agent payment decisions and transactions for compliance

7. Testing: Test agent workflows on devnet before deploying to mainnet

Next Steps

• See JavaScript/MCP examples for TypeScript integration

• Check out MCP Tool Reference for complete tool schemas

• Read Authentication for security best practices