x402 Discovery Query

pip install swarms httpx

import asyncio
from typing import List, Optional, Dict, Any
from swarms import Agent
import httpx


async def query_x402_services(
    limit: Optional[int] = None,
    max_price: Optional[int] = None,
    offset: int = 0,
    base_url: str = "https://api.cdp.coinbase.com",
) -> Dict[str, Any]:
    """
    Query x402 discovery services from the Coinbase CDP API.

    Args:
        limit: Optional maximum number of services to return. If None, returns all available.
        max_price: Optional maximum price in atomic units to filter by. Only services with
                   maxAmountRequired <= max_price will be included.
        offset: Pagination offset for the API request. Defaults to 0.
        base_url: Base URL for the API. Defaults to Coinbase CDP API.

    Returns:
        Dict containing the API response with 'items' list and pagination info.

    Raises:
        httpx.HTTPError: If the HTTP request fails.
        httpx.RequestError: If there's a network error.
    """
    url = f"{base_url}/platform/v2/x402/discovery/resources"
    params = {"offset": offset}

    # If both limit and max_price are specified, fetch more services to account for filtering
    api_limit = limit
    if limit is not None and max_price is not None:
        # Fetch 5x the limit to account for services that might be filtered out
        api_limit = limit * 5

    if api_limit is not None:
        params["limit"] = api_limit

    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.get(url, params=params)
        response.raise_for_status()
        data = response.json()

    # Filter by price if max_price is specified
    if max_price is not None and "items" in data:
        filtered_items = []
        for item in data.get("items", []):
            # Check if any payment option in 'accepts' has maxAmountRequired <= max_price
            accepts = item.get("accepts", [])
            for accept in accepts:
                max_amount_str = accept.get("maxAmountRequired", "")
                if max_amount_str:
                    try:
                        max_amount = int(max_amount_str)
                        if max_amount <= max_price:
                            filtered_items.append(item)
                            break  # Only add item once if any payment option matches
                    except (ValueError, TypeError):
                        continue

        # Apply limit to filtered results if specified
        if limit is not None:
            filtered_items = filtered_items[:limit]

        data["items"] = filtered_items
        # Update pagination total if we filtered
        if "pagination" in data:
            data["pagination"]["total"] = len(filtered_items)

    return data


def get_x402_services_sync(
    limit: Optional[int] = None,
    max_price: Optional[int] = None,
    offset: int = 0,
) -> str:
    """
    Synchronous wrapper for get_x402_services that returns a formatted string.

    Args:
        limit: Optional maximum number of services to return.
        max_price: Optional maximum price in atomic units to filter by.
        offset: Pagination offset for the API request. Defaults to 0.

    Returns:
        JSON-formatted string of service dictionaries matching the criteria.
    """
    async def get_x402_services():
        result = await query_x402_services(
            limit=limit, max_price=max_price, offset=offset
        )
        return result.get("items", [])
    
    services = asyncio.run(get_x402_services())
    return str(services)


# Initialize the agent with the discovery tool
agent = Agent(
    agent_name="X402-Discovery-Agent",
    agent_description="A agent that queries the x402 discovery services from the Coinbase CDP API.",
    model_name="claude-haiku-4-5",
    dynamic_temperature_enabled=True,
    max_loops=1,
    dynamic_context_window=True,
    tools=[get_x402_services_sync],
    top_p=None,
    temperature=None,
    tool_call_summary=True,
)

if __name__ == "__main__":
    # Run the agent
    out = agent.run(
        task="Summarize the first 10 services under 100000 atomic units (e.g., $0.10 USDC)"
    )
    print(out)

result = await query_x402_services()
print(f"Found {len(result['items'])} services")

# Get first 10 services under 100000 atomic units ($0.10 USDC with 6 decimals)
services = await get_x402_services(limit=10, max_price=100000)
for service in services:
    print(service["resource"])

# The agent will automatically call the tool and provide a summary
out = agent.run(
    task="Find and summarize 5 affordable services under 50000 atomic units"
)
print(out)