AlphaVantage-MCP

Instructions

Implementation Reference

• src/alpha_vantage_mcp/server.py:662-695 (handler)

  Handler function block that executes the get-realtime-options tool: validates inputs, calls Alpha Vantage REALTIME_OPTIONS API via make_alpha_request, formats response with format_realtime_options, and returns formatted text.
  elif name == "get-realtime-options": symbol = arguments.get("symbol") require_greeks = arguments.get("require_greeks", False) contract = arguments.get("contract") datatype = arguments.get("datatype", "json") if not symbol: return [types.TextContent(type="text", text="Missing symbol parameter")] symbol = symbol.upper() async with httpx.AsyncClient() as client: params = {} if require_greeks: params["require_greeks"] = "true" if contract: params["contract"] = contract if datatype: params["datatype"] = datatype options_data = await make_alpha_request( client, "REALTIME_OPTIONS", symbol, params ) if isinstance(options_data, str): return [types.TextContent(type="text", text=f"Error: {options_data}")] formatted_options = format_realtime_options(options_data) options_text = f"Realtime options data for {symbol}:\n\n{formatted_options}" return [types.TextContent(type="text", text=options_text)]
• src/alpha_vantage_mcp/server.py:318-346 (schema)

  Input JSON schema for the get-realtime-options tool, defining parameters: symbol (required), require_greeks (boolean), contract (string), datatype (json/csv). Part of tool registration in handle_list_tools().
  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"], }, ),
• src/alpha_vantage_mcp/tools.py:677-802 (helper)

  Key helper function that processes and formats the raw realtime options data from Alpha Vantage API. Handles premium/demo data detection, groups by expiration and strike, includes Greeks if available, and generates readable output.
  def format_realtime_options(options_data: Dict[str, Any]) -> str: """Format realtime options data into a concise string. Args: options_data: The response data from the Alpha Vantage REALTIME_OPTIONS endpoint Returns: A formatted string containing the realtime options information """ try: if "Error Message" in options_data: return f"Error: {options_data['Error Message']}" # Get the options contracts list options_list = options_data.get("data", []) if not options_list: return "No realtime options data available in the response" # Detect placeholder/demo data patterns def is_placeholder_data(contracts_list): """Detect if the response contains placeholder/demo data""" for contract in contracts_list: symbol = contract.get("symbol", "") contract_id = contract.get("contractID", "") expiration = contract.get("expiration", "") # Check for common placeholder patterns if (symbol == "XXYYZZ" or "XXYYZZ" in contract_id or expiration == "2099-99-99" or "999999" in contract_id): return True return False if is_placeholder_data(options_list): # Check if we're using demo API key api_key_status = "demo API key" if API_KEY == "demo" else f"API key ending in ...{API_KEY[-4:]}" if API_KEY and len(API_KEY) > 4 else "no API key set" return ( "❌ PREMIUM FEATURE REQUIRED ❌\n\n" "The realtime options data you requested requires a premium Alpha Vantage subscription.\n" "The API returned placeholder/demo data instead of real market data.\n\n" f"Current API key status: {api_key_status}\n\n" "Possible causes:\n" "1. Using demo API key instead of your actual API key\n" "2. API key is valid but account doesn't have premium access\n" "3. Alpha Vantage returns demo data for free accounts on this endpoint\n\n" "Solutions:\n" "1. Ensure your actual API key is set in ALPHA_VANTAGE_API_KEY environment variable\n" "2. Upgrade to Alpha Vantage Premium (600 or 1200 requests/minute plan)\n" "3. Use 'get-historical-options' tool for historical data (available with free accounts)\n\n" "Learn more: https://www.alphavantage.co/premium/\n\n" "Note: Historical options data may meet your analysis needs and works with free accounts." ) # Group contracts by expiration date and then by strike price contracts_by_expiry = {} for contract in options_list: expiry = contract.get("expiration", "Unknown") strike = contract.get("strike", "0.00") contract_type = contract.get("type", "unknown") if expiry not in contracts_by_expiry: contracts_by_expiry[expiry] = {} if strike not in contracts_by_expiry[expiry]: contracts_by_expiry[expiry][strike] = {} contracts_by_expiry[expiry][strike][contract_type] = contract # Extract symbol from first contract symbol = options_list[0].get("symbol", "Unknown") if options_list else "Unknown" formatted = [ f"Realtime Options Data for {symbol}\n", f"Found {len(options_list)} contracts\n\n" ] # Sort by expiration dates sorted_expiries = sorted(contracts_by_expiry.keys()) for expiry in sorted_expiries: formatted.append(f"=== Expiration: {expiry} ===\n") # Sort strikes numerically strikes = contracts_by_expiry[expiry] sorted_strikes = sorted(strikes.keys(), key=lambda x: float(x) if str(x).replace('.', '').isdigit() else 0) for strike in sorted_strikes: contract_types = strikes[strike] for contract_type in ["call", "put"]: if contract_type in contract_types: contract = contract_types[contract_type] formatted.append(f"\nStrike: ${strike} ({contract_type.upper()})\n") formatted.append(f"Contract ID: {contract.get('contractID', 'N/A')}\n") formatted.append(f"Last: ${contract.get('last', 'N/A')}\n") formatted.append(f"Mark: ${contract.get('mark', 'N/A')}\n") formatted.append(f"Bid: ${contract.get('bid', 'N/A')} (Size: {contract.get('bid_size', 'N/A')})\n") formatted.append(f"Ask: ${contract.get('ask', 'N/A')} (Size: {contract.get('ask_size', 'N/A')})\n") formatted.append(f"Volume: {contract.get('volume', 'N/A')}\n") formatted.append(f"Open Interest: {contract.get('open_interest', 'N/A')}\n") formatted.append(f"Date: {contract.get('date', 'N/A')}\n") # Include Greeks if available if 'implied_volatility' in contract: formatted.append(f"IV: {contract.get('implied_volatility', 'N/A')}\n") if 'delta' in contract: formatted.append(f"Delta: {contract.get('delta', 'N/A')}\n") if 'gamma' in contract: formatted.append(f"Gamma: {contract.get('gamma', 'N/A')}\n") if 'theta' in contract: formatted.append(f"Theta: {contract.get('theta', 'N/A')}\n") if 'vega' in contract: formatted.append(f"Vega: {contract.get('vega', 'N/A')}\n") if 'rho' in contract: formatted.append(f"Rho: {contract.get('rho', 'N/A')}\n") formatted.append("---\n") formatted.append("\n") return "".join(formatted) except Exception as e: return f"Error formatting realtime options data: {str(e)}"