AlphaVantage-MCP

Instructions

Implementation Reference

• src/alpha_vantage_mcp/server.py:553-577 (handler)

  Main handler logic for the 'get-crypto-weekly' tool. Extracts parameters, makes API call to Alpha Vantage DIGITAL_CURRENCY_WEEKLY endpoint, handles errors, formats output using helper function, and returns formatted text content.
  elif name == "get-crypto-weekly": 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_WEEKLY", 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, "weekly") data_text = f"Weekly 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:216-234 (schema)

  Input schema definition for the 'get-crypto-weekly' tool, specifying required 'symbol' parameter and optional 'market' with default 'USD'.
  types.Tool( name="get-crypto-weekly", description="Get weekly 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:32-361 (registration)

  Tool registration via the @server.list_tools() decorator. The 'get-crypto-weekly' tool is included in the returned list of available tools.
  @server.list_tools() async def handle_list_tools() -> list[types.Tool]: """ List available tools. Each tool specifies its arguments using JSON Schema validation. """ return [ types.Tool( name="get-stock-quote", description="Get current stock quote information", inputSchema={ "type": "object", "properties": { "symbol": { "type": "string", "description": "Stock symbol (e.g., AAPL, MSFT)", }, }, "required": ["symbol"], }, ), types.Tool( name="get-company-info", description="Get detailed company information", inputSchema={ "type": "object", "properties": { "symbol": { "type": "string", "description": "Stock symbol (e.g., AAPL, MSFT)", }, }, "required": ["symbol"], }, ), types.Tool( name="get-crypto-exchange-rate", description="Get current cryptocurrency exchange rate", inputSchema={ "type": "object", "properties": { "crypto_symbol": { "type": "string", "description": "Cryptocurrency symbol (e.g., BTC, ETH)", }, "market": { "type": "string", "description": "Market currency (e.g., USD, EUR)", "default": "USD" } }, "required": ["crypto_symbol"], }, ), types.Tool( name="get-time-series", description="Get daily time series data for a stock with optional date filtering", inputSchema={ "type": "object", "properties": { "symbol": { "type": "string", "description": "Stock symbol (e.g., AAPL, MSFT)", }, "outputsize": { "type": "string", "description": "compact (latest 100 data points) or full (up to 20 years of data). When start_date or end_date is specified, defaults to 'full'", "enum": ["compact", "full"], "default": "compact" }, "start_date": { "type": "string", "description": "Optional: Start date in YYYY-MM-DD format for filtering results", "pattern": "^20[0-9]{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12][0-9]|3[01])$" }, "end_date": { "type": "string", "description": "Optional: End date in YYYY-MM-DD format for filtering results", "pattern": "^20[0-9]{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12][0-9]|3[01])$" }, "limit": { "type": "integer", "description": "Optional: Number of data points to return when no date filtering is applied (default: 5)", "default": 5, "minimum": 1 } }, "required": ["symbol"], }, ), types.Tool( name="get-historical-options", description="Get historical options chain data for a stock with advanced filtering and sorting capabilities", inputSchema={ "type": "object", "properties": { "symbol": { "type": "string", "description": "Stock symbol (e.g., AAPL, MSFT)", }, "date": { "type": "string", "description": "Optional: Trading date in YYYY-MM-DD format (defaults to previous trading day, must be after 2008-01-01)", "pattern": "^20[0-9]{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12][0-9]|3[01])$" }, "expiry_date": { "type": "string", "description": "Optional: Filter by expiration date in YYYY-MM-DD format", "pattern": "^20[0-9]{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12][0-9]|3[01])$" }, "min_strike": { "type": "number", "description": "Optional: Minimum strike price filter (e.g., 100.00)", "minimum": 0 }, "max_strike": { "type": "number", "description": "Optional: Maximum strike price filter (e.g., 200.00)", "minimum": 0 }, "contract_id": { "type": "string", "description": "Optional: Filter by specific contract ID (e.g., MSTR260116C00000500)" }, "contract_type": { "type": "string", "description": "Optional: Filter by contract type (call or put)", "enum": ["call", "put", "C", "P"] }, "limit": { "type": "integer", "description": "Optional: Number of contracts to return after filtering (default: 10, use -1 for all contracts)", "default": 10, "minimum": -1 }, "sort_by": { "type": "string", "description": "Optional: Field to sort by", "enum": [ "strike", "expiration", "volume", "open_interest", "implied_volatility", "delta", "gamma", "theta", "vega", "rho", "last", "bid", "ask" ], "default": "strike" }, "sort_order": { "type": "string", "description": "Optional: Sort order", "enum": ["asc", "desc"], "default": "asc" } }, "required": ["symbol"], }, ), 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"], }, ), types.Tool( name="get-crypto-weekly", description="Get weekly 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"], }, ), 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"], }, ), types.Tool( name="get-earnings-calendar", description="Get upcoming earnings calendar data for companies with sorting capabilities", inputSchema={ "type": "object", "properties": { "symbol": { "type": "string", "description": "Optional: Stock symbol to filter earnings for a specific company (e.g., AAPL, MSFT, IBM)" }, "horizon": { "type": "string", "description": "Optional: Time horizon for earnings data (3month, 6month, or 12month)", "enum": ["3month", "6month", "12month"], "default": "12month" }, "limit": { "type": "integer", "description": "Optional: Number of earnings entries to return (default: 100)", "default": 100, "minimum": 1 }, "sort_by": { "type": "string", "description": "Optional: Field to sort by", "enum": ["reportDate", "symbol", "name", "fiscalDateEnding", "estimate"], "default": "reportDate" }, "sort_order": { "type": "string", "description": "Optional: Sort order", "enum": ["asc", "desc"], "default": "desc" } }, "required": [], }, ), types.Tool( name="get-historical-earnings", description="Get historical earnings data for a specific company", inputSchema={ "type": "object", "properties": { "symbol": { "type": "string", "description": "Stock symbol for the company (e.g., AAPL, MSFT, IBM)" }, "limit_annual": { "type": "integer", "description": "Optional: Number of annual earnings to return (default: 5)", "default": 5, "minimum": 1 }, "limit_quarterly": { "type": "integer", "description": "Optional: Number of quarterly earnings to return (default: 8)", "default": 8, "minimum": 1 } }, "required": ["symbol"], }, ), types.Tool( name="get-realtime-options", description="Get realtime options chain data for a stock with optional Greeks and filtering", inputSchema={ "type": "object", "properties": { "symbol": { "type": "string", "description": "Stock symbol (e.g., AAPL, MSFT)", }, "require_greeks": { "type": "boolean", "description": "Optional: Enable Greeks and implied volatility calculation (default: false)", "default": False }, "contract": { "type": "string", "description": "Optional: Specific options contract ID to retrieve" }, "datatype": { "type": "string", "description": "Optional: Response format (json or csv, default: json)", "enum": ["json", "csv"], "default": "json" } }, "required": ["symbol"], }, ), types.Tool( name="get-etf-profile", description="Get comprehensive ETF profile information including holdings, sector allocation, and key metrics", inputSchema={ "type": "object", "properties": { "symbol": { "type": "string", "description": "ETF symbol (e.g., QQQ, SPY, VTI)", } }, "required": ["symbol"], }, ) ]
• src/alpha_vantage_mcp/tools.py:265-330 (helper)

  Helper function to format the raw API response for cryptocurrency time series data. For weekly data, it uses key 'Time Series (Digital Currency Weekly)' and formats metadata and the most 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)

  General helper for making HTTP requests to Alpha Vantage API, used by the handler to fetch DIGITAL_CURRENCY_WEEKLY data with symbol and market parameters.
  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)}"