Feature cookie api

.env.example

@@ -11,6 +11,9 @@ SOLANA_RPC_URL=your_solana_rpc_url
 # Alchemy Configuration
 ALCHEMY_API_KEY=
 
+# Cookie.fun Configuration
+COOKIE_FUN_API_KEY=
+
 # Telegram Configuration
 TELEGRAM_BOT_TOKEN=your_bot_token
 TELEGRAM_SERVER_IP=0.0.0.0

alphaswarm/services/cookiefun/__init__.py

@@ -0,0 +1 @@
+from .cookiefun_client import CookieFunClient, AgentMetrics, Contract, Tweet, PagedAgentsResponse, Interval

alphaswarm/services/cookiefun/cookiefun_client.py

@@ -0,0 +1,248 @@
+import logging
+import os
+from enum import Enum
+from typing import Dict, List, Optional
+
+import requests
+from alphaswarm.config import Config
+from alphaswarm.services.api_exception import ApiException
+from pydantic.dataclasses import dataclass
+
+# Set up logging
+logger = logging.getLogger(__name__)
+
+
+class Interval(str, Enum):
+    THREE_DAYS = "_3Days"
+    SEVEN_DAYS = "_7Days"
+
+
+@dataclass
+class Contract:
+    chain: int
+    contractAddress: str
+
+
+@dataclass
+class Tweet:
+    tweetUrl: str
+    tweetAuthorProfileImageUrl: str
+    tweetAuthorDisplayName: str
+    smartEngagementPoints: int
+    impressionsCount: int
+
+
+@dataclass
+class AgentMetrics:
+    agentName: str
+    contracts: List[Contract]
+    twitterUsernames: List[str]
+    mindshare: float
+    mindshareDeltaPercent: float
+    marketCap: float
+    marketCapDeltaPercent: float
+    price: float
+    priceDeltaPercent: float
+    liquidity: float
+    volume24Hours: float
+    volume24HoursDeltaPercent: float
+    holdersCount: int
+    holdersCountDeltaPercent: float
+    averageImpressionsCount: float
+    averageImpressionsCountDeltaPercent: float
+    averageEngagementsCount: float
+    averageEngagementsCountDeltaPercent: float
+    followersCount: int
+    smartFollowersCount: int
+    topTweets: List[Tweet]
+
+
+@dataclass
+class PagedAgentsResponse:
+    """Response from the paged agents endpoint"""
+
+    data: List[AgentMetrics]
+    currentPage: int
+    totalPages: int
+    totalCount: int
+
+
+class CookieFunClient:
+    """Client for interacting with the Cookie.fun API"""
+
+    BASE_URL = "https://api.cookie.fun/v2/agents"
+
+    def __init__(
+        self, base_url: str = BASE_URL, api_key: Optional[str] = None, config: Optional[Config] = None, **kwargs
+    ):
+        """Initialize the Cookie.fun API client
+
+        Args:
+            base_url: Base URL for the API
+            api_key: API key for authentication
+            config: Config instance for token lookups
+
+        Raises:
+            ValueError: If COOKIE_FUN_API_KEY environment variable is not set
+        """
+        self.base_url = base_url
+        self.api_key = api_key or os.getenv("COOKIE_FUN_API_KEY")
+        if not self.api_key:
+            raise ValueError("COOKIE_FUN_API_KEY environment variable not set")
+
+        self.headers = {"x-api-key": self.api_key}
+        self.config = config or Config()
+        logger.debug("CookieFun client initialized")
+
+    def _get_token_address(self, symbol: str) -> tuple[Optional[str], Optional[str]]:
+        """Get token address and chain from symbol using config
+
+        Args:
+            symbol: Token symbol to look up
+
+        Returns:
+            tuple: (token_address, chain) if found, (None, None) otherwise
+        """
+        try:
+            # Get all supported chains from config
+            supported_chains = self.config.get_supported_networks()
+
+            # Search through each chain for the token
+            for chain in supported_chains:
+                chain_config = self.config.get_chain_config(chain)
+                token_info = chain_config.get_token_info_or_none(symbol)
+                if token_info:
+                    logger.debug(f"Found token {symbol} on chain {chain}")
+                    return token_info.address, chain
+
+            logger.warning(f"Token {symbol} not found in any chain config")
+            return None, None
+
+        except Exception:
+            logger.exception(f"Failed to find token address for {symbol}")
+            return None, None
+
+    def _make_request(self, endpoint: str, params: Dict = None) -> Dict:
+        """Make API request to Cookie.fun
+
+        Args:
+            endpoint: API endpoint path
+            params: Query parameters
+
+        Returns:
+            Dict: API response data
+
+        Raises:
+            ApiException: If API request fails
+            Exception: For other errors
+        """
+        url = f"{self.BASE_URL}{endpoint}"
+
+        try:
+            response = requests.get(url, headers=self.headers, params=params)
+
+            if response.status_code >= 400:
+                raise ApiException(response)
+
+            return response.json()
+
+        except Exception:
+            logger.exception("Error fetching data from Cookie.fun")
+            raise
+
+    def _parse_agent_response(self, response_data: dict) -> AgentMetrics:
+        """Parse API response into AgentMetrics object
+
+        Args:
+            response_data: Raw API response dictionary
+
+        Returns:
+            AgentMetrics: Parsed metrics object
+        """
+        logger.debug(f"Parsing agent response: {response_data}")
+
+        data = response_data["ok"]
+        return AgentMetrics(**data)
+
+    def get_agent_by_twitter(self, username: str, interval: Interval) -> AgentMetrics:
+        """Get agent metrics by Twitter username
+
+        Args:
+            username: Twitter username of the agent
+            interval: Time interval for metrics
+
+        Returns:
+            AgentMetrics: Agent metrics data
+
+        Raises:
+            ApiException: If API request fails
+        """
+        logger.info(f"Fetching metrics for Twitter username: {username}")
+
+        response = self._make_request(f"/twitterUsername/{username}", params={"interval": interval})
+        return self._parse_agent_response(response)
+
+    def get_agent_by_contract(self, address_or_symbol: str, interval: Interval, chain: str = None) -> AgentMetrics:
+        """Get agent metrics by contract address or symbol
+
+        Args:
+            address_or_symbol: Contract address or token symbol
+            interval: Time interval for metrics
+            chain: Optional chain override (not needed for symbols as they are unique per chain)
+
+        Returns:
+            AgentMetrics: Agent metrics data
+
+        Raises:
+            ApiException: If API request fails
+            ValueError: If symbol not found in any chain
+        """
+        # If input looks like an address, use it directly with provided chain
+        if address_or_symbol.startswith("0x") or address_or_symbol.startswith("1"):
+            address = address_or_symbol
+        else:
+            # Try to look up symbol
+            address, detected_chain = self._get_token_address(address_or_symbol)
+            if not address:
+                raise ValueError(f"Could not find address for token {address_or_symbol} in any chain")
+
+            # Use detected chain unless explicitly overridden
+            chain = chain or detected_chain
+            logger.info(f"Resolved symbol {address_or_symbol} to address {address} on chain {chain}")
+
+        logger.info(f"Fetching metrics for contract address: {address}")
+
+        response = self._make_request(f"/contractAddress/{address}", params={"interval": interval})
+        return self._parse_agent_response(response)
+
+    def get_agents_paged(self, interval: Interval, page: int, page_size: int) -> PagedAgentsResponse:
+        """Get paged list of AI agents ordered by mindshare
+
+        Args:
+            interval: Time interval for metrics
+            page: Page number (starts at 1)
+            page_size: Number of agents per page (between 1 and 25)
+
+        Returns:
+            PagedAgentsResponse: Paged list of agent metrics
+
+        Raises:
+            ValueError: If page_size is not between 1 and 25
+            ApiException: If API request fails
+        """
+        if not 1 <= page_size <= 25:
+            raise ValueError("page_size must be between 1 and 25")
+
+        logger.info(f"Fetching agents page {page} with size {page_size}")
+
+        response = self._make_request(
+            "/agentsPaged", params={"interval": interval, "page": page, "pageSize": page_size}
+        )
+
+        data = response["ok"]
+        return PagedAgentsResponse(
+            data=[AgentMetrics(**agent) for agent in data["data"]],
+            currentPage=data["currentPage"],
+            totalPages=data["totalPages"],
+            totalCount=data["totalCount"],
+        )

alphaswarm/tools/cookie/cookie_metrics.py

@@ -0,0 +1,92 @@
+import logging
+from typing import Optional
+
+from alphaswarm.services.cookiefun.cookiefun_client import AgentMetrics, CookieFunClient, Interval, PagedAgentsResponse
+from smolagents import Tool
+
+# Set up logging
+logger = logging.getLogger(__name__)
+
+
+class CookieMetricsByTwitter(Tool):
+    name = "CookieMetricsByTwitter"
+    description = "Retrieve AI agent metrics such as mindshare, market cap, price, liquidity, volume, holders, average impressions, average engagements, followers, and top tweets by Twitter username from Cookie.fun"
+    inputs = {
+        "username": {
+            "type": "string",
+            "description": "Twitter username of the agent",
+        },
+        "interval": {
+            "type": "string",
+            "description": "Time interval for metrics (_3Days or _7Days)",
+            "enum": ["_3Days", "_7Days"],
+        },
+    }
+    output_type = "object"
+
+    def __init__(self, client: Optional[CookieFunClient] = None):
+        super().__init__()
+        self.client = client or CookieFunClient()
+
+    def forward(self, username: str, interval: str) -> AgentMetrics:
+        return self.client.get_agent_by_twitter(username, Interval(interval))
+
+
+class CookieMetricsByContract(Tool):
+    name = "CookieMetricsByContract"
+    description = "Retrieve AI agent metrics such as mindshare, market cap, price, liquidity, volume, holders, average impressions, average engagements, followers, and top tweets by contract address or token symbol from Cookie.fun"
+    inputs = {
+        "address_or_symbol": {
+            "type": "string",
+            "description": "Contract address of the agent token or token symbol (e.g. 'COOKIE' or '0xc0041ef357b183448b235a8ea73ce4e4ec8c265f'). For addresses, chain must be specified.",
+        },
+        "chain": {
+            "type": "string",
+            "description": "Chain to look up token on (required only when using contract address)",
+            "nullable": True,
+        },
+        "interval": {
+            "type": "string",
+            "description": "Time interval for metrics (_3Days or _7Days)",
+            "enum": ["_3Days", "_7Days"],
+        },
+    }
+    output_type = "object"
+
+    def __init__(self, client: Optional[CookieFunClient] = None):
+        super().__init__()
+        self.client = client or CookieFunClient()
+
+    def forward(self, address_or_symbol: str, interval: str, chain: Optional[str] = None) -> AgentMetrics:
+        return self.client.get_agent_by_contract(address_or_symbol, Interval(interval), chain)
+
+
+class CookieMetricsPaged(Tool):
+    name = "CookieMetricsPaged"
+    description = "Retrieve paged list of AI agents ordered by mindshare from Cookie.fun. Important for getting a list of trending AI agents. page_size is the number of agents per page. If asked for example for Top 10 agents, page_size should be 10."
+    inputs = {
+        "interval": {
+            "type": "string",
+            "description": "Time interval for metrics (_3Days or _7Days)",
+            "enum": ["_3Days", "_7Days"],
+        },
+        "page": {
+            "type": "integer",
+            "description": "Page number (starts at 1)",
+            "minimum": 1,
+        },
+        "page_size": {
+            "type": "integer",
+            "description": "Number of agents per page",
+            "minimum": 1,
+            "maximum": 25,
+        },
+    }
+    output_type = "object"
+
+    def __init__(self, client: Optional[CookieFunClient] = None):
+        super().__init__()
+        self.client = client or CookieFunClient()
+
+    def forward(self, interval: str, page: int, page_size: int) -> PagedAgentsResponse:
+        return self.client.get_agents_paged(Interval(interval), page, page_size)

examples/terminal.py

@@ -6,6 +6,7 @@
 from alphaswarm.agent.clients import TerminalClient
 from alphaswarm.config import Config
 from alphaswarm.tools.alchemy import AlchemyPriceHistoryByAddress, AlchemyPriceHistoryBySymbol
+from alphaswarm.tools.cookie.cookie_metrics import CookieMetricsByContract, CookieMetricsByTwitter, CookieMetricsPaged
 from alphaswarm.tools.exchanges import GetTokenPriceTool
 from alphaswarm.tools.price_tool import PriceTool
 from smolagents import Tool
@@ -21,6 +22,9 @@ async def main():
         GetTokenPriceTool(config),
         AlchemyPriceHistoryByAddress(),
         AlchemyPriceHistoryBySymbol(),
+        CookieMetricsByTwitter(),
+        CookieMetricsByContract(),
+        CookieMetricsPaged(),
     ]
     agent = AlphaSwarmAgent(tools=tools, model_id="gpt-4o")
     manager = AlphaSwarmAgentManager(agent)

tests/integration/conftest.py

@@ -1,10 +1,11 @@
 import time
-
 from _pytest.fixtures import fixture
+import pytest
 
 from alphaswarm.config import Config
 from alphaswarm.services.alchemy import AlchemyClient
 from tests.unit.conftest import default_config
+from alphaswarm.services.cookiefun import CookieFunClient
 
 __all__ = ["default_config"]
 
@@ -14,3 +15,10 @@ def alchemy_client(default_config: Config) -> AlchemyClient:
     # this helps with rate limit
     time.sleep(1)
     return AlchemyClient()
+
+
+@pytest.fixture
+def cookiefun_client(default_config: Config) -> CookieFunClient:
+    """Create CookieFun client for testing"""
+    # Set test API key if not present
+    return CookieFunClient()