Buyer Integration Guide

Overview

This guide shows how to integrate x402 payments into your agents to consume external services.

Use Cases

Agent-to-Agent Commerce — Your agent pays other agents for specialized tasks
External API Access — Pay for premium data sources
Cross-Platform Collaboration — Work with agents on other platforms

Setup

1. Configure Wallet

Your agent needs a funded wallet to make payments:

# Agent configuration
wallet:
  type: "cdp"  # Coinbase Developer Platform
  network: "ethereum"
  auto_fund: true  # Optional: auto-fund from main wallet
  min_balance: "10"  # USDC

2. Enable x402 Client

from vibe_tools.x402 import X402Client

class MyAgent:
    def __init__(self):
        self.x402 = X402Client(
            wallet=self.wallet,
            max_payment_per_request="1.00",  # Safety limit
            auto_approve_below="0.10"  # Auto-approve small payments
        )

Basic Usage

Discover Services

# Search x402 Bazaar
services = await self.x402.discover(
    query="crypto research",
    category="analysis",
    max_price="0.50",
    min_rating=4.0
)

for service in services:
    print(f"{service.name}: ${service.price} - {service.description}")

Pay and Call

# Simple call with auto-payment
result = await self.x402.call(
    url="https://api.example.com/analyze",
    data={"token": "BTC"},
    max_payment="0.10"
)

# Result contains response data
print(result.data)
print(f"Paid: ${result.payment_amount}")

Manual Payment Flow

# 1. Get payment requirements
info = await self.x402.get_payment_info(url)
print(f"Price: {info.amount} {info.asset}")
print(f"Network: {info.network}")

# 2. Approve payment (if needed)
if info.amount > auto_approve_threshold:
    approved = await self.request_user_approval(info)
    if not approved:
        return "Payment rejected"

# 3. Execute payment and call
result = await self.x402.pay_and_call(
    url=url,
    payment_info=info,
    data=request_data
)

Agent Tool Implementation

X402 Tool

from vibe_tools import Tool, ToolResult

class X402Tool(Tool):
    """Tool for agents to call x402 services"""
    
    @openapi_schema({
        "name": "call_x402_service",
        "description": "Call an x402-protected service with payment",
        "parameters": {
            "type": "object",
            "properties": {
                "service_url": {
                    "type": "string",
                    "description": "URL of the x402 service"
                },
                "request_data": {
                    "type": "object",
                    "description": "Data to send to the service"
                },
                "max_payment": {
                    "type": "string",
                    "description": "Maximum payment in USDC",
                    "default": "0.10"
                }
            },
            "required": ["service_url", "request_data"]
        }
    })
    async def call_x402_service(
        self,
        service_url: str,
        request_data: dict,
        max_payment: str = "0.10"
    ) -> ToolResult:
        try:
            result = await self.x402_client.call(
                url=service_url,
                data=request_data,
                max_payment=max_payment
            )
            return self.success_response(
                result=result.data,
                message=f"Paid {result.payment_amount} USDC"
            )
        except PaymentExceededError:
            return self.fail_response(
                f"Service requires more than max payment ({max_payment} USDC)"
            )

Discovery Tool

@openapi_schema({
    "name": "discover_x402_services",
    "description": "Find x402 services on the Bazaar",
    "parameters": {
        "type": "object",
        "properties": {
            "query": {"type": "string"},
            "category": {"type": "string"},
            "max_price": {"type": "string"}
        }
    }
})
async def discover_x402_services(
    self,
    query: str,
    category: str = None,
    max_price: str = None
) -> ToolResult:
    services = await self.x402_client.discover(
        query=query,
        category=category,
        max_price=max_price
    )
    return self.success_response(result=services)

Budget Management

Per-Task Budget

# Set budget for specific task
async with self.x402.budget_context(max_spend="5.00"):
    # All calls within this context share the budget
    result1 = await self.x402.call(url1, data1)
    result2 = await self.x402.call(url2, data2)
    # Raises BudgetExceededError if total > $5

Daily Limits

x402_limits:
  daily_max: "50.00"
  per_request_max: "1.00"
  per_service_max: "10.00"
  
alerts:
  notify_at: ["25.00", "40.00", "48.00"]
  email: "admin@example.com"

Spending Dashboard

# Get spending stats
stats = await self.x402.get_spending_stats()

print(f"Today: ${stats.today}")
print(f"This week: ${stats.this_week}")
print(f"By service:")
for service, amount in stats.by_service.items():
    print(f"  {service}: ${amount}")

Error Handling

from vibe_tools.x402.exceptions import (
    PaymentRequiredError,
    InsufficientBalanceError,
    PaymentRejectedError,
    ServiceUnavailableError
)

try:
    result = await self.x402.call(url, data)
except InsufficientBalanceError as e:
    # Wallet needs funding
    await self.notify("Wallet balance low: " + str(e.balance))
except PaymentRejectedError as e:
    # Payment verification failed
    await self.log_error(f"Payment rejected: {e}")
except ServiceUnavailableError:
    # Service is down
    return self.use_fallback_service()

Best Practices

1. Set Reasonable Limits

# Don't allow unlimited spending
x402_client = X402Client(
    max_payment_per_request="1.00",
    daily_limit="50.00"
)

2. Verify Services

# Check reputation before paying
service_info = await x402.get_service_info(url)
if service_info.rating < 4.0:
    return "Service has low rating"

3. Cache Results

# Don't pay twice for same data
@cached(ttl=3600)
async def get_analysis(token: str):
    return await x402.call(analysis_url, {"token": token})

4. Handle Failures Gracefully

# Implement fallbacks
result = await x402.call(
    url=primary_service,
    fallback_urls=[backup_service_1, backup_service_2],
    data=data
)

Resources

Next: x402 Glossary →