AlphaVantage-MCP

Instructions

Implementation Reference

• src/alpha_vantage_mcp/server.py:374-395 (handler)

  Handler logic for executing the get-stock-quote tool: validates symbol input, fetches real-time quote data from Alpha Vantage GLOBAL_QUOTE endpoint using make_alpha_request helper, formats the response with format_quote, and returns formatted text content.

  if name == "get-stock-quote":
      symbol = arguments.get("symbol")
      if not symbol:
          return [types.TextContent(type="text", text="Missing symbol parameter")]
  
      symbol = symbol.upper()
  
      async with httpx.AsyncClient() as client:
          quote_data = await make_alpha_request(
              client,
              "GLOBAL_QUOTE",
              symbol
          )
  
          if isinstance(quote_data, str):
              return [types.TextContent(type="text", text=f"Error: {quote_data}")]
  
          formatted_quote = format_quote(quote_data)
          quote_text = f"Stock quote for {symbol}:\n\n{formatted_quote}"
  
          return [types.TextContent(type="text", text=quote_text)]

• src/alpha_vantage_mcp/server.py:39-52 (registration)

  Registration of the get-stock-quote tool in the MCP server.list_tools() handler, including name, description, and JSON schema for input validation (requires 'symbol' string).

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

• src/alpha_vantage_mcp/server.py:42-51 (schema)

  JSON Schema definition for get-stock-quote tool input: object with required 'symbol' property (string).

  inputSchema={
      "type": "object",
      "properties": {
          "symbol": {
              "type": "string",
              "description": "Stock symbol (e.g., AAPL, MSFT)",
          },
      },
      "required": ["symbol"],
  },

• src/alpha_vantage_mcp/tools.py:93-118 (helper)

  Helper function to format the raw Global Quote API response into a human-readable string displaying price, change (absolute and percent), volume, high, and low.

  def format_quote(quote_data: Dict[str, Any]) -> str:
      """Format quote data into a concise string.
      
      Args:
          quote_data: The response data from the Alpha Vantage Global Quote endpoint
          
      Returns:
          A formatted string containing the quote information
      """
      try:
          global_quote = quote_data.get("Global Quote", {})
          if not global_quote:
              return "No quote data available in the response"
  
          return (
              f"Price: ${global_quote.get('05. price', 'N/A')}\n"
              f"Change: ${global_quote.get('09. change', 'N/A')} "
              f"({global_quote.get('10. change percent', 'N/A')})\n"
              f"Volume: {global_quote.get('06. volume', 'N/A')}\n"
              f"High: ${global_quote.get('03. high', 'N/A')}\n"
              f"Low: ${global_quote.get('04. low', 'N/A')}\n"
              "---"
          )
      except Exception as e:
          return f"Error formatting quote data: {str(e)}"

• src/alpha_vantage_mcp/tools.py:18-91 (helper)

  Core helper function that makes authenticated HTTP requests to the Alpha Vantage API. Used by get-stock-quote handler with function='GLOBAL_QUOTE' and the symbol. Handles JSON/CSV parsing, rate limits, timeouts, and API errors gracefully.

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