swarms/examples/guides/x402_examples/agent_integration/x402_discovery_query.py

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.

    Example:
        ```python
        # Get all services
        result = await query_x402_services()
        print(f"Found {len(result['items'])} services")

        # Get first 10 services under 100000 atomic units
        result = await query_x402_services(limit=10, max_price=100000)
        ```
    """
    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
    # This ensures we can return the requested number after filtering by price
    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 filter_services_by_price(
    services: List[Dict[str, Any]], max_price: int
) -> List[Dict[str, Any]]:
    """
    Filter services by maximum price in atomic units.

    Args:
        services: List of service dictionaries from the API.
        max_price: Maximum price in atomic units. Only services with at least one
                   payment option where maxAmountRequired <= max_price will be included.

    Returns:
        List of filtered service dictionaries.

    Example:
        ```python
        all_services = result["items"]
        affordable = filter_services_by_price(all_services, max_price=100000)
        ```
    """
    filtered = []
    for item in services:
        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.append(item)
                        break  # Only add item once if any payment option matches
                except (ValueError, TypeError):
                    continue
    return filtered


def limit_services(
    services: List[Dict[str, Any]], max_count: int
) -> List[Dict[str, Any]]:
    """
    Limit the number of services returned.

    Args:
        services: List of service dictionaries.
        max_count: Maximum number of services to return.

    Returns:
        List containing at most max_count services.

    Example:
        ```python
        all_services = result["items"]
        limited = limit_services(all_services, max_count=10)
        ```
    """
    return services[:max_count]


async def get_x402_services(
    limit: Optional[int] = None,
    max_price: Optional[int] = None,
    offset: int = 0,
) -> List[Dict[str, Any]]:
    """
    Get x402 services with optional filtering by count and price.

    This is a convenience function that queries the API and applies filters.

    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:
        List of service dictionaries matching the criteria.

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

    return result.get("items", [])


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.

    Example:
        ```python
        # Get first 10 services under $0.10 USDC
        services_str = get_x402_services_sync(limit=10, max_price=100000)
        print(services_str)
        ```
    """
    services = asyncio.run(
        get_x402_services(
            limit=limit, max_price=max_price, offset=offset
        )
    )
    return str(services)



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=0.0,
    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)