AlphaVantage-MCP

Instructions

Implementation Reference

• src/alpha_vantage_mcp/server.py:122-196 (registration)

  Tool registration including name, description, and detailed input schema for get-historical-options
  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"], }, ),
• src/alpha_vantage_mcp/server.py:477-525 (handler)

  Handler logic in call_tool that processes arguments, calls Alpha Vantage API for HISTORICAL_OPTIONS, and formats response using format_historical_options
  elif name == "get-historical-options": symbol = arguments.get("symbol") date = arguments.get("date") expiry_date = arguments.get("expiry_date") min_strike = arguments.get("min_strike") max_strike = arguments.get("max_strike") contract_id = arguments.get("contract_id") contract_type = arguments.get("contract_type") limit = arguments.get("limit", 10) sort_by = arguments.get("sort_by", "strike") sort_order = arguments.get("sort_order", "asc") if not symbol: return [types.TextContent(type="text", text="Missing symbol parameter")] symbol = symbol.upper() async with httpx.AsyncClient() as client: params = {} if date: params["date"] = date options_data = await make_alpha_request( client, "HISTORICAL_OPTIONS", symbol, params ) if isinstance(options_data, str): return [types.TextContent(type="text", text=f"Error: {options_data}")] formatted_options = format_historical_options( options_data, limit, sort_by, sort_order, expiry_date, min_strike, max_strike, contract_id, contract_type ) options_text = f"Historical options data for {symbol}" if date: options_text += f" on {date}" options_text += f":\n\n{formatted_options}" return [types.TextContent(type="text", text=options_text)]
• src/alpha_vantage_mcp/tools.py:507-674 (helper)

  Core helper function that formats the historical options data with filtering (by expiry, strike range, contract ID, type), sorting, and limiting, producing the final text output
  def format_historical_options( options_data: Dict[str, Any], limit: int = 10, sort_by: str = "strike", sort_order: str = "asc", expiry_date: Optional[str] = None, min_strike: Optional[float] = None, max_strike: Optional[float] = None, contract_id: Optional[str] = None, contract_type: Optional[str] = None ) -> str: """Format historical options chain data into a concise string with advanced filtering and sorting. Args: options_data: The response data from the Alpha Vantage HISTORICAL_OPTIONS endpoint limit: Number of contracts to return after filtering (-1 for all) sort_by: Field to sort by sort_order: Sort order (asc or desc) expiry_date: Optional expiration date filter (YYYY-MM-DD) min_strike: Optional minimum strike price filter max_strike: Optional maximum strike price filter contract_id: Optional specific contract ID filter contract_type: Optional contract type filter (call/put/C/P) Returns: A formatted string containing the filtered and sorted historical options information """ try: if "Error Message" in options_data: return f"Error: {options_data['Error Message']}" options_chain = options_data.get("data", []) if not options_chain: return "No options data available in the response" # Apply filters filtered_chain = [] for contract in options_chain: # Contract ID filter (exact match) if contract_id and contract.get('contractID', '') != contract_id: continue # Expiry date filter if expiry_date: contract_expiry = contract.get('expiration', '') if contract_expiry != expiry_date: continue # Strike price filters if min_strike is not None or max_strike is not None: try: strike_str = str(contract.get('strike', '0')).replace('$', '').strip() if strike_str: strike_price = float(strike_str) if min_strike is not None and strike_price < min_strike: continue if max_strike is not None and strike_price > max_strike: continue except (ValueError, TypeError): continue # Contract type filter if contract_type: contract_type_norm = contract_type.upper() # Handle both full names and single letters if contract_type_norm in ['CALL', 'C']: expected_types = ['call', 'C', 'CALL'] elif contract_type_norm in ['PUT', 'P']: expected_types = ['put', 'P', 'PUT'] else: expected_types = [contract_type] actual_type = contract.get('type', '') if actual_type not in expected_types: continue filtered_chain.append(contract) if not filtered_chain: filters_applied = [] if contract_id: filters_applied.append(f"contract_id={contract_id}") if expiry_date: filters_applied.append(f"expiry={expiry_date}") if min_strike is not None: filters_applied.append(f"min_strike={min_strike}") if max_strike is not None: filters_applied.append(f"max_strike={max_strike}") if contract_type: filters_applied.append(f"type={contract_type}") filter_text = ", ".join(filters_applied) if filters_applied else "applied" return f"No options contracts found matching the specified filters: {filter_text}" formatted = [ f"Historical Options Data (Filtered):\n", f"Status: {options_data.get('message', 'N/A')}\n", ] # Add filter summary filters_applied = [] if contract_id: filters_applied.append(f"Contract ID: {contract_id}") if expiry_date: filters_applied.append(f"Expiry: {expiry_date}") if min_strike is not None or max_strike is not None: strike_range = [] if min_strike is not None: strike_range.append(f"min ${min_strike}") if max_strike is not None: strike_range.append(f"max ${max_strike}") filters_applied.append(f"Strike: {' - '.join(strike_range)}") if contract_type: filters_applied.append(f"Type: {contract_type}") if filters_applied: formatted.append(f"Filters: {', '.join(filters_applied)}\n") formatted.append(f"Found {len(filtered_chain)} contracts, sorted by: {sort_by} ({sort_order})\n\n") # Convert string values to float for numeric sorting def get_sort_key(contract): value = contract.get(sort_by, 0) try: # Remove $ and % signs if present if isinstance(value, str): value = value.replace('$', '').replace('%', '') return float(value) except (ValueError, TypeError): return value # Sort the filtered chain sorted_chain = sorted( filtered_chain, key=get_sort_key, reverse=(sort_order == "desc") ) # If limit is -1, show all contracts display_contracts = sorted_chain if limit == -1 else sorted_chain[:limit] for contract in display_contracts: formatted.append(f"Contract Details:\n") formatted.append(f"Contract ID: {contract.get('contractID', 'N/A')}\n") formatted.append(f"Expiration: {contract.get('expiration', 'N/A')}\n") formatted.append(f"Strike: ${contract.get('strike', 'N/A')}\n") formatted.append(f"Type: {contract.get('type', '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"IV: {contract.get('implied_volatility', 'N/A')}\n") formatted.append(f"Delta: {contract.get('delta', 'N/A')}\n") formatted.append(f"Gamma: {contract.get('gamma', 'N/A')}\n") formatted.append(f"Theta: {contract.get('theta', 'N/A')}\n") formatted.append(f"Vega: {contract.get('vega', 'N/A')}\n") formatted.append(f"Rho: {contract.get('rho', 'N/A')}\n") formatted.append("---\n") if limit != -1 and len(sorted_chain) > limit: formatted.append(f"\n... and {len(sorted_chain) - limit} more contracts") return "".join(formatted) except Exception as e: return f"Error formatting options data: {str(e)}"