Enhanced MCP Stock Query System

""" MCP Stock Server - Enhanced Financial Data Provider with CSV Fallback This module implements a Model Context Protocol (MCP) server that provides stock market data functionality using the Yahoo Finance API with CSV fallback support. It offers tools for retrieving current stock prices and comparing multiple stock symbols, along with MCP resources and prompts for better protocol compliance. The server exposes: - Tools: get_stock_price, compare_stocks, get_market_summary - Resources: stock_data, market_info - Prompts: stock_analysis, comparison_prompt Fallback Strategy: - Primary: Yahoo Finance API (yfinance) - Fallback: Local CSV file (stocks_data.csv) Dependencies: - mcp.server.fastmcp: FastMCP framework for creating MCP servers - yfinance: Yahoo Finance API wrapper for stock data retrieval - pandas: For CSV data handling """ from mcp.server.fastmcp import FastMCP from mcp.types import Resource, TextResourceContents, Prompt, PromptMessage, Role import yfinance as yf import pandas as pd import os import logging from typing import Optional, Tuple, Dict, Any, List from datetime import datetime import json # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) mcp = FastMCP("Enhanced Stock Server") # Configuration CSV_FILE_PATH = "stocks_data.csv" SUPPORTED_SYMBOLS = [ "AAPL", "MSFT", "GOOGL", "AMZN", "META", "TSLA", "BRK-B", "JNJ", "V", "WMT", "JPM", "PG", "UNH", "NVDA", "HD", "DIS", "MA", "BAC", "VZ", "ADBE" ] class StockDataError(Exception): """Custom exception for stock data related errors.""" pass def validate_stock_symbol(symbol: str) -> str: """ Validate and normalize stock symbol. Args: symbol: Stock ticker symbol to validate Returns: Normalized symbol in uppercase Raises: StockDataError: If symbol is invalid """ if not symbol or not isinstance(symbol, str): raise StockDataError("Symbol must be a non-empty string") normalized_symbol = symbol.strip().upper() if len(normalized_symbol) > 10: raise StockDataError("Symbol too long (max 10 characters)") return normalized_symbol def get_price_from_csv(symbol: str) -> Optional[Tuple[float, str]]: """ Retrieve stock price and last updated date from local CSV file. Expected CSV format: symbol,price,last_updated AAPL,150.25,2024-01-15 MSFT,380.50,2024-01-15 Args: symbol: Stock ticker symbol Returns: Tuple of (price, last_updated) if found, None otherwise Raises: StockDataError: If CSV file is malformed """ try: if not os.path.exists(CSV_FILE_PATH): logger.warning(f"CSV file {CSV_FILE_PATH} not found") return None df = pd.read_csv(CSV_FILE_PATH) # Validate CSV structure required_columns = ['symbol', 'price', 'last_updated'] if not all(col in df.columns for col in required_columns): raise StockDataError(f"CSV file missing required columns: {required_columns}") # Convert symbol column to uppercase for case-insensitive matching df['symbol'] = df['symbol'].str.upper() symbol = symbol.upper() # Find the stock in the CSV stock_row = df[df['symbol'] == symbol] if not stock_row.empty: price = float(stock_row['price'].iloc[0]) last_updated = stock_row['last_updated'].iloc[0] return price, last_updated else: return None except pd.errors.EmptyDataError: raise StockDataError("CSV file is empty") except pd.errors.ParserError as e: raise StockDataError(f"CSV file is malformed: {e}") except Exception as e: logger.error(f"Error reading CSV file: {e}") raise StockDataError(f"Failed to read CSV file: {e}") def get_stock_price_with_fallback(symbol: str) -> Tuple[Optional[float], str, Optional[str]]: """ Get stock price with fallback mechanism. Args: symbol: Stock ticker symbol Returns: Tuple of (price, source, last_updated) where: - price: Stock price or None if not found - source: 'yfinance', 'csv', or 'none' - last_updated: Date string for CSV data, None for yfinance Raises: StockDataError: If symbol validation fails """ symbol = validate_stock_symbol(symbol) # Try yfinance first try: logger.info(f"Attempting to fetch {symbol} from Yahoo Finance") ticker = yf.Ticker(symbol) # Get today's data (may be empty if market is closed) data = ticker.history(period="1d") if not data.empty: price = float(data['Close'].iloc[-1]) logger.info(f"Successfully fetched {symbol} from Yahoo Finance: ${price:.2f}") return price, 'yfinance', None else: # Try using regular market price from ticker info info = ticker.info price = info.get("regularMarketPrice") if price is not None: price = float(price) logger.info(f"Successfully fetched {symbol} from Yahoo Finance info: ${price:.2f}") return price, 'yfinance', None except Exception as e: logger.warning(f"Yahoo Finance error for {symbol}: {e}") # Fallback to CSV try: logger.info(f"Falling back to CSV for {symbol}") csv_result = get_price_from_csv(symbol) if csv_result is not None: price, last_updated = csv_result logger.info(f"Successfully fetched {symbol} from CSV: ${price:.2f}") return price, 'csv', last_updated except StockDataError as e: logger.error(f"CSV fallback failed for {symbol}: {e}") logger.warning(f"Could not retrieve price for {symbol} from any source") return None, 'none', None @mcp.resource("stock_data") async def get_stock_data_resource() -> Resource: """ MCP Resource: Provides access to available stock data. Returns: Resource containing information about available stocks and data sources """ try: # Get available symbols from CSV available_symbols = [] if os.path.exists(CSV_FILE_PATH): df = pd.read_csv(CSV_FILE_PATH) available_symbols = df['symbol'].str.upper().tolist() resource_data = { "description": "Stock market data with Yahoo Finance and CSV fallback", "available_symbols": available_symbols, "data_sources": ["Yahoo Finance API", "Local CSV file"], "csv_file": CSV_FILE_PATH, "last_updated": datetime.now().isoformat() } return Resource( uri="stock://data", name="Stock Data Information", mimeType="application/json", contents=[TextResourceContents( type="text", text=json.dumps(resource_data, indent=2) )] ) except Exception as e: logger.error(f"Error creating stock data resource: {e}") raise StockDataError(f"Failed to create stock data resource: {e}") @mcp.resource("market_info") async def get_market_info_resource() -> Resource: """ MCP Resource: Provides general market information and server capabilities. Returns: Resource containing market information and server capabilities """ market_info = { "server_name": "Enhanced Stock Server", "capabilities": [ "Real-time stock prices via Yahoo Finance", "Fallback to local CSV data", "Stock price comparisons", "Market summary information" ], "supported_operations": [ "get_stock_price", "compare_stocks", "get_market_summary" ], "data_freshness": "Real-time for Yahoo Finance, static for CSV fallback" } return Resource( uri="stock://market-info", name="Market Information", mimeType="application/json", contents=[TextResourceContents( type="text", text=json.dumps(market_info, indent=2) )] ) @mcp.prompt("stock_analysis") async def stock_analysis_prompt() -> Prompt: """ MCP Prompt: Template for stock analysis queries. Returns: Prompt template for analyzing stock performance """ return Prompt( name="stock_analysis", description="Analyze stock performance and provide insights", messages=[ PromptMessage( role=Role.user, content={ "type": "text", "text": "Analyze the stock {symbol} and provide insights on:\n" "1. Current price and recent performance\n" "2. Key factors affecting the stock\n" "3. Potential risks and opportunities\n" "Please use the available stock tools to get current data." } ) ] ) @mcp.prompt("comparison_prompt") async def comparison_prompt() -> Prompt: """ MCP Prompt: Template for comparing multiple stocks. Returns: Prompt template for stock comparison analysis """ return Prompt( name="comparison_prompt", description="Compare multiple stocks and provide investment insights", messages=[ PromptMessage( role=Role.user, content={ "type": "text", "text": "Compare stocks {symbol1} and {symbol2} by analyzing:\n" "1. Current prices and price difference\n" "2. Which stock might be a better investment\n" "3. Risk factors for each stock\n" "Use the compare_stocks tool to get current data." } ) ] ) @mcp.tool() def get_stock_price(symbol: str) -> str: """ Retrieve the current stock price for the given ticker symbol. First tries Yahoo Finance API, then falls back to local CSV file. Args: symbol: Stock ticker symbol (e.g., 'AAPL', 'MSFT') Returns: Current stock price information with data source Raises: StockDataError: If symbol is invalid or data cannot be retrieved """ try: price, source, last_updated = get_stock_price_with_fallback(symbol) if price is not None: if source == 'yfinance': return f"The current price of {symbol} is ${price:.2f} (from Yahoo Finance)" elif source == 'csv': return f"The current price of {symbol} is ${price:.2f} (from local data, last updated: {last_updated})" return f"Could not retrieve price for {symbol} from either Yahoo Finance or local data. "\ f"Please ensure the symbol is correct and that local data file '{CSV_FILE_PATH}' "\ f"exists with the required format." except StockDataError as e: logger.error(f"Stock data error for {symbol}: {e}") return f"Error retrieving stock price for {symbol}: {e}" except Exception as e: logger.error(f"Unexpected error for {symbol}: {e}") return f"Unexpected error retrieving stock price for {symbol}. Please try again." @mcp.tool() def compare_stocks(symbol1: str, symbol2: str) -> str: """ Compare the current stock prices of two ticker symbols. First tries Yahoo Finance API, then falls back to local CSV file for each symbol. Args: symbol1: First stock ticker symbol symbol2: Second stock ticker symbol Returns: Comparison of the two stock prices with data sources Raises: StockDataError: If symbols are invalid or data cannot be retrieved """ try: # Get prices for both symbols price1, source1, updated1 = get_stock_price_with_fallback(symbol1) price2, source2, updated2 = get_stock_price_with_fallback(symbol2) if price1 is None: return f"Could not retrieve price for {symbol1} from either Yahoo Finance or local data." if price2 is None: return f"Could not retrieve price for {symbol2} from either Yahoo Finance or local data." # Create source information source1_text = " (YF)" if source1 == 'yfinance' else f" (local, {updated1})" source2_text = " (YF)" if source2 == 'yfinance' else f" (local, {updated2})" # Calculate difference difference = abs(price1 - price2) percentage_diff = (difference / min(price1, price2)) * 100 if price1 > price2: result = f"{symbol1} (${price1:.2f}{source1_text}) is higher than {symbol2} (${price2:.2f}{source2_text}) by ${difference:.2f} ({percentage_diff:.1f}%)." elif price1 < price2: result = f"{symbol1} (${price1:.2f}{source1_text}) is lower than {symbol2} (${price2:.2f}{source2_text}) by ${difference:.2f} ({percentage_diff:.1f}%)." else: result = f"Both {symbol1} and {symbol2} have the same price (${price1:.2f})." return result except StockDataError as e: logger.error(f"Stock data error comparing {symbol1} and {symbol2}: {e}") return f"Error comparing stocks: {e}" except Exception as e: logger.error(f"Unexpected error comparing {symbol1} and {symbol2}: {e}") return f"Unexpected error comparing stocks. Please try again." @mcp.tool() def get_market_summary() -> str: """ Get a summary of available stocks and their current status. Returns: Summary of market data availability and key stocks """ try: summary_data = [] available_count = 0 # Sample a few key stocks for summary key_symbols = ["AAPL", "MSFT", "GOOGL", "AMZN", "TSLA"] for symbol in key_symbols: try: price, source, updated = get_stock_price_with_fallback(symbol) if price is not None: source_info = "Live" if source == 'yfinance' else f"Cached ({updated})" summary_data.append(f"{symbol}: ${price:.2f} ({source_info})") available_count += 1 except Exception as e: logger.warning(f"Error getting summary for {symbol}: {e}") continue if summary_data: result = f"Market Summary ({available_count}/{len(key_symbols)} stocks available):\n" result += "\n".join(summary_data) result += f"\n\nData sources: Yahoo Finance (primary), Local CSV (fallback)" return result else: return "Market summary unavailable - no stock data could be retrieved." except Exception as e: logger.error(f"Error generating market summary: {e}") return f"Error generating market summary: {e}" if __name__ == "__main__": """ Entry point for the Enhanced MCP Stock Server. When this script is run directly, it starts the FastMCP server which will: 1. Register available tools (get_stock_price, compare_stocks, get_market_summary) 2. Register available resources (stock_data, market_info) 3. Register available prompts (stock_analysis, comparison_prompt) 4. Listen for MCP client connections 5. Handle tool execution requests from clients 6. Provide stock data through Yahoo Finance with CSV fallback The server runs indefinitely until manually stopped (Ctrl+C) or terminated. Enhanced Features: - Comprehensive error handling with custom exceptions - Detailed logging for debugging and monitoring - Input validation and sanitization - MCP resources for data discovery - MCP prompts for common use cases - Type hints for better code quality - Percentage calculations in comparisons CSV File Format: The CSV file should be named 'stocks_data.csv' and have the following format: symbol,price,last_updated AAPL,150.25,2024-01-15 MSFT,380.50,2024-01-15 GOOGL,140.75,2024-01-15 Server Details: - Server Name: "Enhanced Stock Server" - Available Tools: get_stock_price, compare_stocks, get_market_summary - Available Resources: stock_data, market_info - Available Prompts: stock_analysis, comparison_prompt - Protocol: Model Context Protocol (MCP) - Primary Data Source: Yahoo Finance via yfinance library - Fallback Data Source: Local CSV file (stocks_data.csv) """ logger.info("Starting Enhanced MCP Stock Server...") mcp.run()