AlphaVantage-MCP

Instructions

Implementation Reference

• src/alpha_vantage_mcp/server.py:605-635 (handler)

  Main execution logic for the get-earnings-calendar tool. Extracts arguments, calls make_alpha_request with EARNINGS_CALENDAR function, formats using format_earnings_calendar, and returns formatted text content.
  elif name == "get-earnings-calendar": symbol = arguments.get("symbol") horizon = arguments.get("horizon", "12month") limit = arguments.get("limit", 100) sort_by = arguments.get("sort_by", "reportDate") sort_order = arguments.get("sort_order", "desc") async with httpx.AsyncClient() as client: params = {"horizon": horizon} if symbol: params["symbol"] = symbol.upper() earnings_data = await make_alpha_request( client, "EARNINGS_CALENDAR", None, params ) if isinstance(earnings_data, str): return [types.TextContent(type="text", text=f"Error: {earnings_data}")] formatted_earnings = format_earnings_calendar(earnings_data, limit, sort_by, sort_order) earnings_text = f"Earnings calendar" if symbol: earnings_text += f" for {symbol.upper()}" if horizon: earnings_text += f" ({horizon})" earnings_text += f":\n\n{formatted_earnings}" return [types.TextContent(type="text", text=earnings_text)]
• src/alpha_vantage_mcp/server.py:254-291 (registration)

  Tool registration in list_tools() handler, defining name, description, and JSON schema for input validation including optional symbol, horizon, limit, sort_by, and sort_order.
  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": [], }, ),
• src/alpha_vantage_mcp/tools.py:18-91 (helper)

  Utility function to make HTTP requests to Alpha Vantage API. Specially handles EARNINGS_CALENDAR by parsing CSV response into list of dicts.
  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)}"
• src/alpha_vantage_mcp/tools.py:332-420 (helper)

  Helper function that takes the raw earnings data (list of dicts from CSV), applies sorting by specified field/date/numeric handling, limits output, formats into readable multi-line string with company details, dates, and estimates.
  def format_earnings_calendar(earnings_data: List[Dict[str, str]], limit: int = 100, sort_by: str = "reportDate", sort_order: str = "desc") -> str: """Format earnings calendar data into a concise string with sorting. Args: earnings_data: List of earnings records from the Alpha Vantage EARNINGS_CALENDAR endpoint (CSV format) limit: Number of earnings entries to display (default: 100) sort_by: Field to sort by (default: reportDate) sort_order: Sort order asc or desc (default: desc) Returns: A formatted string containing the sorted earnings calendar information """ try: if not isinstance(earnings_data, list): return f"Unexpected data format: {type(earnings_data)}" if not earnings_data: return "No earnings calendar data available" # Sort the earnings data def get_sort_key(earning): value = earning.get(sort_by, "") # Special handling for dates to ensure proper chronological sorting if sort_by in ["reportDate", "fiscalDateEnding"]: try: # Convert date string to datetime for proper sorting if value: return datetime.strptime(value, "%Y-%m-%d") else: return datetime.min # Put empty dates at the beginning except ValueError: return datetime.min # Special handling for numeric fields like estimate elif sort_by == "estimate": try: if value and value.strip(): return float(value) else: return 0.0 except ValueError: return 0.0 # For text fields (symbol, name), return as-is for alphabetical sorting else: return str(value).upper() sorted_earnings = sorted( earnings_data, key=get_sort_key, reverse=(sort_order == "desc") ) formatted = [f"Upcoming Earnings Calendar (Sorted by {sort_by} {sort_order}):\n\n"] # Display limited number of entries display_earnings = sorted_earnings[:limit] if limit > 0 else sorted_earnings for earning in display_earnings: symbol = earning.get('symbol', 'N/A') name = earning.get('name', 'N/A') report_date = earning.get('reportDate', 'N/A') fiscal_date = earning.get('fiscalDateEnding', 'N/A') estimate = earning.get('estimate', 'N/A') currency = earning.get('currency', 'N/A') formatted.append(f"Company: {symbol} - {name}\n") formatted.append(f"Report Date: {report_date}\n") formatted.append(f"Fiscal Date End: {fiscal_date}\n") # Format estimate nicely if estimate and estimate != 'N/A' and estimate.strip(): try: est_float = float(estimate) formatted.append(f"Estimate: ${est_float:.2f} {currency}\n") except ValueError: formatted.append(f"Estimate: {estimate} {currency}\n") else: formatted.append(f"Estimate: Not available\n") formatted.append("---\n") if limit > 0 and len(earnings_data) > limit: formatted.append(f"\n... and {len(earnings_data) - limit} more earnings reports") return "".join(formatted) except Exception as e: return f"Error formatting earnings calendar data: {str(e)}"