""" title: Coingecko API Tool author: naoufel version: 1.0 license: MIT """ import json import os import requests from datetime import datetime from pydantic import BaseModel, Field class EventEmitter: def __init__(self, emitter=None): self.emitter = emitter async def emit( self, description="Unknown State", status="in_progress", done=False, step_number=None, source=None ): if self.emitter: message = { "type": "status", "data": { "status": status, "description": description, "done": done, } } if step_number: message["data"]["step"] = step_number await self.emitter(message) # Send source if provided if source: await self.emitter({ "type": "source", "data": source }) class Tools: class Valves(BaseModel): """ A class containing constants and configuration values for the CoinGecko API. """ # Base API URL COINGECKO_API_BASE: str = Field( default="https://pro-api.coingecko.com/api/v3" ) # Free API URL COINGECKO_FREE_API_BASE: str = Field( default="https://api.coingecko.com/api/v3" ) # API Key configuration API_KEY: str = Field( default=os.getenv("COINGECKO_API_KEY", "") ) # Update the parameter name to match CoinGecko's expected format API_KEY_PARAM: str = Field(default="x-cg-pro-api-key") # Common cryptocurrency IDs CRYPTO_IDS: dict = Field( default={ "bitcoin": "bitcoin", "ethereum": "ethereum", "binancecoin": "binancecoin", "solana": "solana", "cardano": "cardano", "ripple": "ripple", "polkadot": "polkadot", "dogecoin": "dogecoin", "avalanche-2": "avalanche-2", "matic-network": "matic-network", "inverse": "inverse-finance", "dola": "dola-usd", } ) # Time range options for historical data TIME_RANGES: dict = Field( default={ "1d": "1", "7d": "7", "14d": "14", "30d": "30", "90d": "90", "180d": "180", "365d": "365", "max": "max", } ) # API endpoints ENDPOINTS: dict = Field( default={ "simple_price": "/simple/price", "market_chart": "/coins/{id}/market_chart", "coin_info": "/coins/{id}", "trending": "/search/trending", "global": "/global", } ) # Default parameters DEFAULT_VS_CURRENCY: str = Field(default="usd") DEFAULT_TIME_RANGE: str = Field(default="1") DEFAULT_PRICE_PARAMS: dict = Field( default={ "include_market_cap": "true", "include_24hr_vol": "true", "include_24hr_change": "true", "include_last_updated_at": "true", } ) # Rate limiting MAX_REQUESTS_PER_MINUTE: int = Field(default=50) # Error messages ERROR_MESSAGES: dict = Field( default={ "invalid_coin_id": "Invalid cryptocurrency ID provided", "invalid_time_range": "Invalid time range specified", "rate_limit_exceeded": "API rate limit exceeded", "api_error": "Error accessing CoinGecko API", } ) def __init__(self): self.valves = self.Valves() pass async def get_crypto_data( self, coin_id: str, days: str = None, __event_emitter__=None, ) -> str: """ Fetches cryptocurrency data from CoinGecko API. This function retrieves price, market cap, and volume data for a specific cryptocurrency. Args: coin_id (str): The CoinGecko ID of the cryptocurrency (e.g., 'bitcoin', 'ethereum') days (str): Time range for historical data (e.g., '1', '14', '30', 'max') __event_emitter__ (callable, optional): A function to emit status events during execution Returns: str: JSON string containing the cryptocurrency data or an error message if the fetch fails """ # Check API key first if not self.valves.API_KEY: error_msg = "CoinGecko API key is not set in environment variable 'COINGECKO_API_KEY'" await emitter.emit(error_msg, done=True) return error_msg days = days or self.valves.DEFAULT_TIME_RANGE emitter = EventEmitter(__event_emitter__) await emitter.emit(f"Fetching data for {coin_id}...") if coin_id not in self.valves.CRYPTO_IDS.values(): await emitter.emit(self.valves.ERROR_MESSAGES["invalid_coin_id"], done=True) return self.valves.ERROR_MESSAGES["invalid_coin_id"] try: # Fetch current price and market data price_params = { "ids": coin_id, "vs_currencies": self.valves.DEFAULT_VS_CURRENCY, "x_cg_pro_api_key": self.valves.API_KEY, **self.valves.DEFAULT_PRICE_PARAMS, } price_url = f"{self.valves.COINGECKO_API_BASE}{self.valves.ENDPOINTS['simple_price']}" price_response = requests.get( price_url, params=price_params, ) price_response.raise_for_status() # Fetch historical market data history_url = ( f"{self.valves.COINGECKO_API_BASE}{self.valves.ENDPOINTS['market_chart']}".format( id=coin_id ) ) history_params = { "vs_currency": self.valves.DEFAULT_VS_CURRENCY, "days": days, "x_cg_pro_api_key": self.valves.API_KEY, } history_response = requests.get( history_url, params=history_params, ) history_response.raise_for_status() price_data = price_response.json() history_data = history_response.json() # Format the response formatted_data = { "current_data": { "price_usd": price_data[coin_id]["usd"], "market_cap_usd": price_data[coin_id]["usd_market_cap"], "volume_24h_usd": price_data[coin_id]["usd_24h_vol"], "price_change_24h": price_data[coin_id]["usd_24h_change"], }, "historical_data": { "prices": history_data["prices"], "market_caps": history_data["market_caps"], "total_volumes": history_data["total_volumes"], }, } await emitter.emit( "Cryptocurrency data retrieved successfully", done=True, source={ "document": [json.dumps({ "current_data": formatted_data["current_data"], "call": f"simple/price?ids={coin_id}&vs_currencies=usd", "historical_call": f"coins/{coin_id}/market_chart?days={days}" })], "metadata": [{ "source": "CoinGecko API", "date_accessed": datetime.now().isoformat(), "name": f"Price data for {coin_id}" }], "source": { "name": "CoinGecko", "url": f"https://www.coingecko.com/en/coins/{coin_id}", "type": "crypto_price_data" } } ) return json.dumps(formatted_data) except requests.RequestException as e: await emitter.emit(f"Error: {str(e)}", done=True) return f"Error fetching data for {coin_id}: {str(e)}" async def get_trending_coins(self, __event_emitter__=None) -> str: """ Fetches trending cryptocurrency data from CoinGecko API. """ emitter = EventEmitter(__event_emitter__) await emitter.emit("Fetching trending coins...") try: # Use the free API endpoint for trending coins trending_url = f"{self.valves.COINGECKO_FREE_API_BASE}{self.valves.ENDPOINTS['trending']}" trending_response = requests.get(trending_url) trending_response.raise_for_status() trending_data = trending_response.json() # Format the response formatted_data = { "trending_coins": [ { "name": coin["item"]["name"], "symbol": coin["item"]["symbol"], "market_cap_rank": coin["item"]["market_cap_rank"], "price_btc": coin["item"]["price_btc"] } for coin in trending_data["coins"] ] } await emitter.emit( "Trending cryptocurrency data retrieved successfully", done=True, source={ "document": [json.dumps({ "trending_coin": coin, "call": "search/trending", "timestamp": datetime.now().timestamp() }) for coin in formatted_data["trending_coins"]], "metadata": [{ "source": "CoinGecko API", "date_accessed": datetime.now().isoformat(), "name": "Trending Cryptocurrencies" }], "source": { "name": "CoinGecko Trending", "url": "https://www.coingecko.com/en/trending-cryptocurrencies", "type": "trending_crypto" } } ) return json.dumps(formatted_data, indent=2) except requests.RequestException as e: error_msg = f"Error fetching trending coins: {str(e)}" await emitter.emit(error_msg, done=True) return error_msg