get-crypto-daily
Retrieve daily cryptocurrency time series data for analysis, providing historical price information to inform investment decisions.
Instructions
Get daily time series data for a cryptocurrency
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| symbol | Yes | Cryptocurrency symbol (e.g., BTC, ETH) | |
| market | No | Market currency (e.g., USD, EUR) | USD |
Implementation Reference
- src/alpha_vantage_mcp/server.py:527-551 (handler)Handler for the 'get-crypto-daily' tool: extracts symbol and market parameters, calls Alpha Vantage DIGITAL_CURRENCY_DAILY API via make_alpha_request, formats response with format_crypto_time_series, and returns formatted daily crypto time series data.
elif name == "get-crypto-daily": symbol = arguments.get("symbol") market = arguments.get("market", "USD") if not symbol: return [types.TextContent(type="text", text="Missing symbol parameter")] symbol = symbol.upper() market = market.upper() async with httpx.AsyncClient() as client: crypto_data = await make_alpha_request( client, "DIGITAL_CURRENCY_DAILY", symbol, {"market": market} ) if isinstance(crypto_data, str): return [types.TextContent(type="text", text=f"Error: {crypto_data}")] formatted_data = format_crypto_time_series(crypto_data, "daily") data_text = f"Daily cryptocurrency time series for {symbol} in {market}:\n\n{formatted_data}" return [types.TextContent(type="text", text=data_text)] - Input schema definition for 'get-crypto-daily' tool registered in handle_list_tools(): requires 'symbol' (crypto symbol), optional 'market' defaulting to 'USD'.
types.Tool( name="get-crypto-daily", description="Get daily time series data for a cryptocurrency", inputSchema={ "type": "object", "properties": { "symbol": { "type": "string", "description": "Cryptocurrency symbol (e.g., BTC, ETH)", }, "market": { "type": "string", "description": "Market currency (e.g., USD, EUR)", "default": "USD" } }, "required": ["symbol"], }, ), - Helper function format_crypto_time_series used by get-crypto-daily handler to parse and format the Alpha Vantage 'DIGITAL_CURRENCY_DAILY' API response into readable text, showing metadata and recent 5 data points.
def format_crypto_time_series(time_series_data: Dict[str, Any], series_type: str) -> str: """Format cryptocurrency time series data into a concise string. Args: time_series_data: The response data from Alpha Vantage Digital Currency endpoints series_type: Type of time series (daily, weekly, monthly) Returns: A formatted string containing the cryptocurrency time series information """ try: # Determine the time series key based on series_type time_series_key = "" if series_type == "daily": time_series_key = "Time Series (Digital Currency Daily)" elif series_type == "weekly": time_series_key = "Time Series (Digital Currency Weekly)" elif series_type == "monthly": time_series_key = "Time Series (Digital Currency Monthly)" else: return f"Unknown series type: {series_type}" # Get the time series data time_series = time_series_data.get(time_series_key, {}) if not time_series: all_keys = ", ".join(time_series_data.keys()) return f"No cryptocurrency time series data found with key: '{time_series_key}'.\nAvailable keys: {all_keys}" # Get metadata metadata = time_series_data.get("Meta Data", {}) crypto_symbol = metadata.get("2. Digital Currency Code", "Unknown") crypto_name = metadata.get("3. Digital Currency Name", "Unknown") market = metadata.get("4. Market Code", "Unknown") market_name = metadata.get("5. Market Name", "Unknown") last_refreshed = metadata.get("6. Last Refreshed", "Unknown") time_zone = metadata.get("7. Time Zone", "Unknown") # Format the header formatted_data = [ f"{series_type.capitalize()} Time Series for {crypto_name} ({crypto_symbol})", f"Market: {market_name} ({market})", f"Last Refreshed: {last_refreshed} {time_zone}", "" ] # Format the most recent 5 data points for date, values in list(time_series.items())[:5]: # Get price information - based on the API response, we now know the correct field names open_price = values.get("1. open", "N/A") high_price = values.get("2. high", "N/A") low_price = values.get("3. low", "N/A") close_price = values.get("4. close", "N/A") volume = values.get("5. volume", "N/A") formatted_data.append(f"Date: {date}") formatted_data.append(f"Open: {open_price} {market}") formatted_data.append(f"High: {high_price} {market}") formatted_data.append(f"Low: {low_price} {market}") formatted_data.append(f"Close: {close_price} {market}") formatted_data.append(f"Volume: {volume}") formatted_data.append("---") return "\n".join(formatted_data) except Exception as e: return f"Error formatting cryptocurrency time series data: {str(e)}" - src/alpha_vantage_mcp/tools.py:18-91 (helper)Utility function make_alpha_request used by the handler to perform the HTTP GET request to Alpha Vantage API for 'DIGITAL_CURRENCY_DAILY' function with symbol and market params, handles errors, rate limits, and parses JSON/CSV responses.
async def make_alpha_request(client: httpx.AsyncClient, function: str, symbol: Optional[str], additional_params: Optional[Dict[str, Any]] = None) -> Dict[str, Any] | str: """Make a request to the Alpha Vantage API with proper error handling. Args: client: An httpx AsyncClient instance function: The Alpha Vantage API function to call symbol: The stock/crypto symbol (can be None for some endpoints) additional_params: Additional parameters to include in the request Returns: Either a dictionary containing the API response, or a string with an error message """ params = { "function": function, "apikey": API_KEY } if symbol: params["symbol"] = symbol if additional_params: params.update(additional_params) try: response = await client.get( ALPHA_VANTAGE_BASE, params=params, timeout=30.0 ) # Check for specific error responses if response.status_code == 429: return f"Rate limit exceeded. Error details: {response.text}" elif response.status_code == 403: return f"API key invalid or expired. Error details: {response.text}" response.raise_for_status() # Check if response is empty if not response.text.strip(): return "Empty response received from Alpha Vantage API" # Special handling for EARNINGS_CALENDAR which returns CSV by default if function == "EARNINGS_CALENDAR": try: # Parse CSV response csv_reader = csv.DictReader(io.StringIO(response.text)) earnings_list = list(csv_reader) return earnings_list except Exception as e: return f"Error parsing CSV response: {str(e)}" # For other functions, expect JSON try: data = response.json() except ValueError as e: return f"Invalid JSON response from Alpha Vantage API: {response.text[:200]}" # Check for Alpha Vantage specific error messages if "Error Message" in data: return f"Alpha Vantage API error: {data['Error Message']}" if "Note" in data and "API call frequency" in data["Note"]: return f"Rate limit warning: {data['Note']}" return data except httpx.TimeoutException: return "Request timed out after 30 seconds. The Alpha Vantage API may be experiencing delays." except httpx.ConnectError: return "Failed to connect to Alpha Vantage API. Please check your internet connection." except httpx.HTTPStatusError as e: return f"HTTP error occurred: {str(e)} - Response: {e.response.text}" except Exception as e: return f"Unexpected error occurred: {str(e)}"