feat(python): add x402search ActionProvider

python/coinbase-agentkit/coinbase_agentkit/action_providers/__init__.py

@@ -38,6 +38,7 @@
 from .wow.wow_action_provider import WowActionProvider, wow_action_provider
 from .x402.schemas import X402Config
 from .x402.x402_action_provider import x402_action_provider, x402ActionProvider
+from .x402search.x402search_action_provider import X402SearchActionProvider, x402search_action_provider
 
 __all__ = [
     "AaveActionProvider",
@@ -84,4 +85,6 @@
     "wow_action_provider",
     "x402ActionProvider",
     "x402_action_provider",
+    "X402SearchActionProvider",
+    "x402search_action_provider",
 ]

python/coinbase-agentkit/coinbase_agentkit/action_providers/x402search/README.md

@@ -0,0 +1,37 @@
+# x402search ActionProvider
+
+Natural language search across **14,000+ indexed API services** for Coinbase AgentKit agents.
+
+**Cost:** $0.01 USDC per query — paid automatically via x402 protocol on Base mainnet.
+
+## Why
+
+Coinbase Bazaar is `ls /apis`. x402search is `grep -r "token price"` across all of them.
+
+When an agent needs a data source, it should search by capability — not iterate every Bazaar endpoint. This provider gives AgentKit agents that search natively, paid automatically with no human in the loop.
+
+## Usage
+```python
+from coinbase_agentkit import AgentKit, AgentKitConfig
+from coinbase_agentkit.wallet_providers import CdpWalletProvider
+from coinbase_agentkit.action_providers.x402search import x402search_action_provider
+
+agentkit = AgentKit(AgentKitConfig(
+    wallet_provider=CdpWalletProvider(cdp_config),
+    action_providers=[x402search_action_provider()],
+))
+```
+
+## Best queries
+
+| Query | Results |
+|-------|---------|
+| `crypto` | 112 |
+| `token price` | 88 |
+| `crypto market data` | 10 |
+| `btc price` | 8 |
+
+## Links
+- Live API: https://x402search.xyz
+- MCP package: `x402search-mcp`
+- x402 protocol: https://x402.org

python/coinbase-agentkit/coinbase_agentkit/action_providers/x402search/__init__.py

@@ -0,0 +1,8 @@
+"""x402search action provider."""
+
+from .x402search_action_provider import X402SearchActionProvider, x402search_action_provider
+
+__all__ = [
+    "X402SearchActionProvider",
+    "x402search_action_provider",
+]

python/coinbase-agentkit/coinbase_agentkit/action_providers/x402search/schemas.py

@@ -0,0 +1,29 @@
+"""Schemas for x402search action provider."""
+
+from pydantic import BaseModel, Field
+
+
+class SearchApisSchema(BaseModel):
+    """Input schema for the search_apis action."""
+
+    query: str = Field(
+        ...,
+        description=(
+            "Natural language query to find API services by capability. "
+            "Examples: 'token price', 'crypto market data', 'NFT metadata', "
+            "'weather forecast', 'sentiment analysis', 'btc price'"
+        ),
+        min_length=2,
+        max_length=200,
+    )
+    limit: int = Field(
+        default=5,
+        ge=1,
+        le=10,
+        description="Maximum number of results to return (1-10, default 5).",
+    )
+
+    class Config:
+        """Pydantic config."""
+
+        title = "Parameters for searching API services by capability"

python/coinbase-agentkit/coinbase_agentkit/action_providers/x402search/x402search_action_provider.py

@@ -0,0 +1,150 @@
+"""x402search action provider."""
+
+import json
+from collections import defaultdict
+from typing import Any
+from urllib.parse import urlparse
+
+from x402.http.clients.requests import x402_requests
+from x402.mechanisms.evm import EthAccountSigner
+from x402.mechanisms.evm.exact.register import register_exact_evm_client
+from x402 import x402ClientSync
+
+from ...network import Network
+from ...wallet_providers import WalletProvider
+from ...wallet_providers.evm_wallet_provider import EvmWalletProvider
+from ..action_decorator import create_action
+from ..action_provider import ActionProvider
+from .schemas import SearchApisSchema
+
+X402SEARCH_URL = "https://x402search.xyz/v1/search"
+
+
+class X402SearchActionProvider(ActionProvider[WalletProvider]):
+    """Provides natural language search across 14,000+ indexed API services via x402search.
+
+    Coinbase Bazaar lists services. x402search searches them by capability.
+    Cost: $0.01 USDC per query via x402 protocol on Base mainnet.
+    """
+
+    def __init__(self):
+        super().__init__("x402search", [])
+
+    @create_action(
+        name="search_apis",
+        description="""Search for API services and data providers by natural language capability query.
+
+Searches 14,000+ indexed APIs across crypto, DeFi, NFT, weather, finance, AI, and more.
+Returns matching services with names, descriptions, and endpoints.
+Cost: $0.01 USDC per query via x402 protocol on Base mainnet — paid automatically.
+
+Use this when an agent needs to discover what APIs exist for a given capability.
+Coinbase Bazaar lists services. x402search searches them by capability — they are complementary.
+
+Examples:
+- search_apis("token price") -> 88 results including CoinGecko, CryptoCompare
+- search_apis("crypto market data") -> 10 focused results
+- search_apis("NFT metadata") -> NFT-related APIs
+- search_apis("btc price") -> 8 targeted results
+""",
+        schema=SearchApisSchema,
+    )
+    def search_apis(self, wallet_provider: WalletProvider, args: dict[str, Any]) -> str:
+        """Search x402search for API services matching the natural language query.
+
+        Args:
+            wallet_provider: The wallet provider used to authorize x402 payment.
+            args: Input arguments containing query and optional limit.
+
+        Returns:
+            str: A JSON string containing matched API services or error details.
+
+        """
+        validated = SearchApisSchema(**args)
+
+        if not isinstance(wallet_provider, EvmWalletProvider):
+            return json.dumps({
+                "success": False,
+                "error": "x402search requires an EvmWalletProvider with USDC on Base mainnet (eip155:8453).",
+            })
+
+        try:
+            client = x402ClientSync()
+            signer = wallet_provider.to_signer()
+            register_exact_evm_client(client, EthAccountSigner(signer))
+            session = x402_requests(client)
+
+            response = session.get(
+                X402SEARCH_URL,
+                params={"q": validated.query, "limit": 50},
+                timeout=15,
+            )
+            response.raise_for_status()
+            data = response.json()
+
+        except Exception as e:
+            return json.dumps({
+                "success": False,
+                "error": f"x402search request failed: {e}",
+                "hint": "Ensure the wallet has USDC on Base mainnet (eip155:8453).",
+            })
+
+        results = data.get("results", data) if isinstance(data, dict) else data
+        if not isinstance(results, list):
+            return json.dumps({"success": False, "error": "Unexpected response format."})
+
+        domain_counts: dict[str, int] = defaultdict(int)
+        deduped = []
+        for r in results:
+            url = r.get("resource_url", "")
+            domain = urlparse(url).netloc
+            if domain_counts[domain] < 2:
+                deduped.append(r)
+                domain_counts[domain] += 1
+        results = deduped[: validated.limit]
+
+        if not results:
+            return json.dumps({
+                "success": True,
+                "query": validated.query,
+                "results": [],
+                "message": (
+                    f"No APIs found for '{validated.query}'. "
+                    "Try broader terms: 'crypto' returns 112 results, 'token price' returns 88."
+                ),
+            })
+
+        formatted = []
+        for r in results:
+            accepts = r.get("accepts", [])
+            entry: dict[str, Any] = {
+                "url": r.get("resource_url", ""),
+                "rank": r.get("rank"),
+            }
+            if accepts:
+                entry["accepts"] = [
+                    {k: v for k, v in a.items() if k in ("network", "max_amount", "payTo")}
+                    for a in accepts
+                ]
+            formatted.append(entry)
+
+        return json.dumps({
+            "success": True,
+            "query": validated.query,
+            "count": len(formatted),
+            "results": formatted,
+        }, indent=2)
+
+    def supports_network(self, network: Network) -> bool:
+        """x402search works with any EVM wallet; payment settles on Base mainnet."""
+        return True
+
+
+def x402search_action_provider() -> X402SearchActionProvider:
+    """Create a new x402search action provider.
+
+    Returns:
+        X402SearchActionProvider: A new x402search action provider instance.
+
+    """
+    return X402SearchActionProvider()

test_x402search.py

@@ -0,0 +1,103 @@
+"""Test script for the x402search action provider.
+
+This script demonstrates how to use the X402SearchActionProvider with a mock
+wallet provider, following the same test patterns used in the repo.
+
+Usage:
+    cd python/coinbase-agentkit && pip install -e .
+    python ../../test_x402search.py
+"""
+
+import json
+import sys
+from unittest.mock import Mock, patch
+
+# ---------------------------------------------------------------------------
+# Mock wallet provider (mimics EvmWalletProvider)
+# ---------------------------------------------------------------------------
+
+from coinbase_agentkit.wallet_providers.evm_wallet_provider import EvmWalletProvider
+
+MOCK_ADDRESS = "0x1234567890123456789012345678901234567890"
+
+mock_wallet = Mock(spec=EvmWalletProvider)
+mock_wallet.get_address.return_value = MOCK_ADDRESS
+mock_wallet.to_signer.return_value = Mock()  # EthAccountSigner-compatible
+
+# ---------------------------------------------------------------------------
+# Import and instantiate the provider
+# ---------------------------------------------------------------------------
+
+from coinbase_agentkit.action_providers.x402search import (
+    X402SearchActionProvider,
+    x402search_action_provider,
+)
+
+provider = x402search_action_provider()
+print(f"Provider name : {provider.name}")
+print(f"Actions       : {[a.name for a in provider._actions]}")
+print()
+
+# ---------------------------------------------------------------------------
+# Call search_apis with a mocked HTTP response (no real network needed)
+# ---------------------------------------------------------------------------
+
+MOCK_SEARCH_RESPONSE = {
+    "results": [
+        {
+            "name": "CoinGecko Price API",
+            "description": "Real-time crypto price data for 10,000+ tokens",
+            "url": "https://api.coingecko.com/api/v3",
+        },
+        {
+            "name": "CryptoCompare",
+            "description": "Comprehensive cryptocurrency market data",
+            "url": "https://min-api.cryptocompare.com",
+        },
+        {
+            "name": "Messari",
+            "description": "Crypto asset metrics and market data",
+            "url": "https://data.messari.io/api/v1",
+        },
+    ]
+}
+
+
+def make_mock_response(data: dict, status_code: int = 200):
+    """Return a requests.Response-like mock."""
+    resp = Mock()
+    resp.status_code = status_code
+    resp.json.return_value = data
+    resp.raise_for_status = Mock()  # no-op for 200
+    return resp
+
+
+# Patch x402_requests so no real HTTP or payment happens
+with patch("coinbase_agentkit.action_providers.x402search.x402search_action_provider.x402_requests") as mock_x402_requests, \
+     patch("coinbase_agentkit.action_providers.x402search.x402search_action_provider.x402ClientSync"), \
+     patch("coinbase_agentkit.action_providers.x402search.x402search_action_provider.EthAccountSigner"), \
+     patch("coinbase_agentkit.action_providers.x402search.x402search_action_provider.register_exact_evm_client"):
+
+    mock_session = Mock()
+    mock_session.get.return_value = make_mock_response(MOCK_SEARCH_RESPONSE)
+    mock_x402_requests.return_value = mock_session
+
+    result = provider.search_apis(mock_wallet, {"query": "crypto price feed"})
+
+print("search_apis(query='crypto price feed') result:")
+print(result)
+print()
+
+parsed = json.loads(result)
+assert parsed["success"] is True, f"Expected success=True, got: {parsed}"
+assert parsed["count"] == 3, f"Expected 3 results, got {parsed['count']}"
+assert parsed["results"][0]["name"] == "CoinGecko Price API"
+
+print(f"  success : {parsed['success']}")
+print(f"  query   : {parsed['query']}")
+print(f"  count   : {parsed['count']}")
+for r in parsed["results"]:
+    print(f"    - {r['name']}: {r['url']}")
+
+print()
+print("All assertions passed.")