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)}"