diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/dexscreener/README.md b/python/coinbase-agentkit/coinbase_agentkit/action_providers/dexscreener/README.md new file mode 100644 index 000000000..52507c378 --- /dev/null +++ b/python/coinbase-agentkit/coinbase_agentkit/action_providers/dexscreener/README.md @@ -0,0 +1,33 @@ +# DexScreener Action Provider + +This action provider gives AI agents access to real-time DEX data from [DexScreener](https://dexscreener.com). + +## Features + +- **Search tokens** by name, symbol, or contract address +- **Get trading pairs** for any token contract +- **Get pair details** with price, volume, and liquidity +- **Discover latest tokens** on any supported chain (Base, Ethereum, Solana, etc.) + +## No API Key Required + +DexScreener's public API is free and requires no authentication. + +## Usage + +```python +from coinbase_agentkit import AgentKit +from coinbase_agentkit.action_providers.dexscreener import dexscreener + +agentkit = AgentKit() +agentkit.add_action_provider(dexscreener()) +``` + +## Actions + +| Action | Description | +|--------|-------------| +| `search_tokens` | Search by name/symbol/address | +| `get_token_pairs` | Get all pairs for a token | +| `get_pair_details` | Detailed pair info | +| `get_latest_base_tokens` | Trending tokens on a chain | diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/dexscreener/__init__.py b/python/coinbase-agentkit/coinbase_agentkit/action_providers/dexscreener/__init__.py new file mode 100644 index 000000000..e2ea599f6 --- /dev/null +++ b/python/coinbase-agentkit/coinbase_agentkit/action_providers/dexscreener/__init__.py @@ -0,0 +1,5 @@ +"""DexScreener action provider.""" + +from .dexscreener_action_provider import DexScreenerActionProvider, dexscreener + +__all__ = ["DexScreenerActionProvider", "dexscreener"] diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/dexscreener/dexscreener_action_provider.py b/python/coinbase-agentkit/coinbase_agentkit/action_providers/dexscreener/dexscreener_action_provider.py new file mode 100644 index 000000000..e3b2555bf --- /dev/null +++ b/python/coinbase-agentkit/coinbase_agentkit/action_providers/dexscreener/dexscreener_action_provider.py @@ -0,0 +1,169 @@ +"""DexScreener action provider for token and DEX data.""" + +import json +from typing import Any + +import requests + +from ...wallet_providers import WalletProvider +from ..action_decorator import create_action +from ..action_provider import ActionProvider +from .schemas import GetLatestTokensSchema, GetPairSchema, GetTokenPairsSchema, SearchTokenSchema + + +class DexScreenerActionProvider(ActionProvider[WalletProvider]): + """Action provider for DexScreener DEX data.""" + + def __init__(self): + super().__init__("dexscreener", []) + + @create_action( + name="search_tokens", + description="""Search for tokens on DexScreener by name, symbol, or contract address. +Returns top matching tokens with price, volume, and liquidity data. + +Inputs: +- query: Search term (token name, symbol, or contract address) + +Example: search for "PEPE" token""", + schema=SearchTokenSchema, + ) + def search_tokens(self, args: dict[str, Any]) -> str: + """Search for tokens on DexScreener.""" + validated = SearchTokenSchema(**args) + try: + resp = requests.get( + f"https://api.dexscreener.com/latest/dex/search/?q={validated.query}", + timeout=10, + ) + if not resp.ok: + return json.dumps({"success": False, "error": f"API returned {resp.status_code}"}) + + data = resp.json() + pairs = data.get("pairs", [])[:5] # Top 5 results + + results = [] + for pair in pairs: + results.append({ + "chain": pair.get("chainId"), + "dex": pair.get("dexId"), + "base_token": pair.get("baseToken", {}).get("symbol"), + "quote_token": pair.get("quoteToken", {}).get("symbol"), + "price_usd": pair.get("priceUsd"), + "price_native": pair.get("priceNative"), + "volume_24h": pair.get("volume", {}).get("h24"), + "liquidity_usd": pair.get("liquidity", {}).get("usd"), + "price_change_24h": pair.get("priceChange", {}).get("h24"), + "pair_address": pair.get("pairAddress"), + }) + + return json.dumps({"success": True, "count": len(results), "pairs": results}, indent=2) + except Exception as e: + return json.dumps({"success": False, "error": str(e)}) + + @create_action( + name="get_token_pairs", + description="""Get all trading pairs for a token contract address on DexScreener. +Returns DEX pairs with price, volume, liquidity, and fee info. + +Inputs: +- token_address: The contract address of the token""", + schema=GetTokenPairsSchema, + ) + def get_token_pairs(self, args: dict[str, Any]) -> str: + """Get trading pairs for a token.""" + validated = GetTokenPairsSchema(**args) + try: + resp = requests.get( + f"https://api.dexscreener.com/tokens/v1/{validated.token_address}", + timeout=10, + ) + if not resp.ok: + return json.dumps({"success": False, "error": f"API returned {resp.status_code}"}) + + pairs = resp.json()[:10] # Top 10 pairs + + results = [] + for pair in pairs: + txns = pair.get("txns", {}).get("h24", {}) + results.append({ + "chain": pair.get("chainId"), + "dex": pair.get("dexId"), + "pair_address": pair.get("pairAddress"), + "base_token": pair.get("baseToken", {}).get("symbol"), + "price_usd": pair.get("priceUsd"), + "volume_24h": pair.get("volume", {}).get("h24"), + "liquidity_usd": pair.get("liquidity", {}).get("usd"), + "buys_24h": txns.get("buys"), + "sells_24h": txns.get("sells"), + "pair_created_at": pair.get("pairCreatedAt"), + }) + + return json.dumps({"success": True, "count": len(results), "pairs": results}, indent=2) + except Exception as e: + return json.dumps({"success": False, "error": str(e)}) + + @create_action( + name="get_pair_details", + description="""Get detailed info for a specific DEX trading pair on DexScreener. + +Inputs: +- pair_address: The address of the trading pair +- chain_id: The chain (e.g., 'base', 'ethereum', 'solana')""", + schema=GetPairSchema, + ) + def get_pair_details(self, args: dict[str, Any]) -> str: + """Get pair details.""" + validated = GetPairSchema(**args) + try: + resp = requests.get( + f"https://api.dexscreener.com/pairs/v1/{validated.chain_id}/{validated.pair_address}", + timeout=10, + ) + if not resp.ok: + return json.dumps({"success": False, "error": f"API returned {resp.status_code}"}) + + pair = resp.json() + return json.dumps({"success": True, "pair": pair}, indent=2) + except Exception as e: + return json.dumps({"success": False, "error": str(e)}) + + @create_action( + name="get_latest_base_tokens", + description="""Get the latest trending tokens on Base chain from DexScreener. +Useful for discovering new tokens and monitoring the Base ecosystem. + +No inputs required - defaults to Base chain.""", + schema=GetLatestTokensSchema, + ) + def get_latest_tokens(self, args: dict[str, Any]) -> str: + """Get latest tokens on a chain.""" + validated = GetLatestTokensSchema(**args) + try: + resp = requests.get( + f"https://api.dexscreener.com/token-profiles/latest/v1", + timeout=10, + ) + if not resp.ok: + return json.dumps({"success": False, "error": f"API returned {resp.status_code}"}) + + tokens = resp.json() + # Filter by chain + chain_tokens = [ + t for t in tokens + if t.get("chainId") == validated.chain_id + ][:20] + + return json.dumps({ + "success": True, + "chain": validated.chain_id, + "count": len(chain_tokens), + "tokens": chain_tokens, + }, indent=2) + except Exception as e: + return json.dumps({"success": False, "error": str(e)}) + + +def dexscreener() -> DexScreenerActionProvider: + """Create a DexScreenerActionProvider instance.""" + return DexScreenerActionProvider() diff --git a/python/coinbase-agentkit/coinbase_agentkit/action_providers/dexscreener/schemas.py b/python/coinbase-agentkit/coinbase_agentkit/action_providers/dexscreener/schemas.py new file mode 100644 index 000000000..853fcc3f4 --- /dev/null +++ b/python/coinbase-agentkit/coinbase_agentkit/action_providers/dexscreener/schemas.py @@ -0,0 +1,31 @@ +"""DexScreener action schemas.""" + +from pydantic import BaseModel, Field + + +class SearchTokenSchema(BaseModel): + """Schema for searching tokens on DexScreener.""" + + query: str = Field(description="Search query - token name, symbol, or contract address") + + +class GetTokenPairsSchema(BaseModel): + """Schema for getting trading pairs for a token.""" + + token_address: str = Field(description="The contract address of the token to look up") + + +class GetPairSchema(BaseModel): + """Schema for getting details of a specific trading pair.""" + + pair_address: str = Field(description="The address of the trading pair") + chain_id: str = Field(description="The chain ID (e.g., 'base', 'ethereum', 'solana')") + + +class GetLatestTokensSchema(BaseModel): + """Schema for getting the latest tokens on a chain.""" + + chain_id: str = Field( + default="base", + description="The chain ID to search (e.g., 'base', 'ethereum', 'solana')" + )