AlphaVantage-MCP

Instructions

Implementation Reference

• src/alpha_vantage_mcp/server.py:579-603 (handler)

  The handler logic for executing the 'get-crypto-monthly' tool. It validates input, makes an API request to Alpha Vantage's DIGITAL_CURRENCY_MONTHLY endpoint, formats the response using format_crypto_time_series, and returns the formatted monthly cryptocurrency time series data.
  elif name == "get-crypto-monthly": 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_MONTHLY", 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, "monthly") data_text = f"Monthly cryptocurrency time series for {symbol} in {market}:\n\n{formatted_data}" return [types.TextContent(type="text", text=data_text)]
• src/alpha_vantage_mcp/server.py:235-253 (registration)

  Registration of the 'get-crypto-monthly' tool in the list_tools() function, including its name, description, and input schema definition.
  types.Tool( name="get-crypto-monthly", description="Get monthly 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"], }, ),
• src/alpha_vantage_mcp/server.py:238-252 (schema)

  JSON schema for input validation of the 'get-crypto-monthly' tool, requiring 'symbol' and optional 'market' parameters.
  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"], },
• src/alpha_vantage_mcp/tools.py:265-330 (helper)

  Helper function that formats the raw API response for cryptocurrency time series data, specifically selecting the 'Time Series (Digital Currency Monthly)' key for monthly data and formatting the most recent 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 used by the handler to make the HTTP request to Alpha Vantage API, specifically called with function='DIGITAL_CURRENCY_MONTHLY', handling errors, rate limits, and parsing 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)}"