Spaces:
Sleeping
Sleeping
| import os | |
| import requests | |
| import gradio as gr | |
| from typing import Optional, List, Dict, Any | |
| from datetime import datetime, timedelta | |
| from mcp.server.fastmcp import FastMCP | |
| # Initialize FastMCP server with standard MCP protocol (JSON-RPC over SSE) | |
| mcp = FastMCP("finnhub-market-info") | |
| # Global variable to store API key | |
| API_KEY = None | |
| def set_api_key(key: str): | |
| """Set the Finnhub API key""" | |
| global API_KEY | |
| API_KEY = key | |
| def get_api_key() -> Optional[str]: | |
| """Get the Finnhub API key""" | |
| return API_KEY or os.getenv("FINNHUB_API_KEY") | |
| def make_finnhub_request(endpoint: str, params: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| Make a request to Finnhub API | |
| Args: | |
| endpoint: API endpoint path | |
| params: Query parameters | |
| Returns: | |
| API response as dictionary | |
| """ | |
| api_key = get_api_key() | |
| if not api_key: | |
| return {"error": "API key not configured. Please set your Finnhub API key."} | |
| params["token"] = api_key | |
| base_url = "https://finnhub.io/api/v1" | |
| url = f"{base_url}/{endpoint}" | |
| try: | |
| response = requests.get(url, params=params, timeout=10) | |
| response.raise_for_status() | |
| return response.json() | |
| except requests.exceptions.RequestException as e: | |
| return {"error": f"API request failed: {str(e)}"} | |
| def get_quote(symbol: str) -> dict: | |
| """ | |
| Get real-time quote data for US stocks. Use this tool when you need current stock price | |
| information and market performance metrics for any US-listed stock. | |
| When to use: | |
| - User asks "What's the current price of [stock]?" | |
| - Need real-time stock quote data | |
| - User mentions "stock price", "current value", "how is [stock] trading?" | |
| - Want to check latest market price and daily changes | |
| Examples: | |
| - "What's Apple's stock price?" → get_quote(symbol="AAPL") | |
| - "How is Tesla trading today?" → get_quote(symbol="TSLA") | |
| - "Show me Microsoft's current quote" → get_quote(symbol="MSFT") | |
| Args: | |
| symbol: Stock ticker symbol (e.g., 'AAPL', 'MSFT', 'TSLA', 'GOOGL') | |
| Returns: | |
| dict: Real-time quote data containing: | |
| - symbol: Stock ticker symbol | |
| - current_price: Current trading price (c) | |
| - change: Price change in dollars (d) | |
| - percent_change: Price change in percentage (dp) | |
| - high: Today's high price (h) | |
| - low: Today's low price (l) | |
| - open: Opening price (o) | |
| - previous_close: Previous trading day's closing price (pc) | |
| - timestamp: Quote timestamp | |
| """ | |
| result = make_finnhub_request("quote", {"symbol": symbol.upper()}) | |
| if "error" in result: | |
| return { | |
| "error": result["error"], | |
| "symbol": symbol.upper() | |
| } | |
| # Return structured data | |
| return { | |
| "symbol": symbol.upper(), | |
| "current_price": result.get('c'), | |
| "change": result.get('d'), | |
| "percent_change": result.get('dp'), | |
| "high": result.get('h'), | |
| "low": result.get('l'), | |
| "open": result.get('o'), | |
| "previous_close": result.get('pc'), | |
| "timestamp": datetime.fromtimestamp(result.get('t', 0)).strftime('%Y-%m-%d %H:%M:%S') if result.get('t') else None | |
| } | |
| def get_market_news(category: str = "general", min_id: int = 0) -> dict: | |
| """ | |
| Get latest market news across different categories. Use this tool when you need current market | |
| news, trends, and developments in general markets, forex, cryptocurrency, or mergers. | |
| When to use: | |
| - User asks "What's the latest market news?" | |
| - Need current financial news and market updates | |
| - User mentions "news", "market trends", "what's happening in the market?" | |
| - Want to get news for specific categories (forex, crypto, M&A) | |
| Categories explained: | |
| - general: General market news, stocks, economy, major companies | |
| - forex: Foreign exchange and currency market news | |
| - crypto: Cryptocurrency and blockchain news | |
| - merger: Mergers & acquisitions, corporate deals | |
| Examples: | |
| - "What's the latest market news?" → get_market_news(category="general") | |
| - "Show me crypto news" → get_market_news(category="crypto") | |
| - "Any forex updates?" → get_market_news(category="forex") | |
| - "Recent merger news" → get_market_news(category="merger") | |
| Args: | |
| category: News category - "general", "forex", "crypto", or "merger" (default: "general") | |
| min_id: Minimum news ID for pagination (default: 0, use 0 to get latest news) | |
| Returns: | |
| dict: Market news data containing: | |
| - category: News category requested | |
| - total_articles: Total number of articles returned | |
| - articles: List of news articles (max 10), each with: | |
| * id: Article ID | |
| * headline: News headline | |
| * summary: Brief summary of the article | |
| * source: News source | |
| * url: Link to full article | |
| * published: Publication timestamp | |
| * image: Article image URL (if available) | |
| """ | |
| params = {"category": category} | |
| if min_id > 0: | |
| params["minId"] = min_id | |
| result = make_finnhub_request("news", params) | |
| if isinstance(result, dict) and "error" in result: | |
| return { | |
| "error": result["error"], | |
| "category": category | |
| } | |
| if not result or len(result) == 0: | |
| return { | |
| "category": category, | |
| "total_articles": 0, | |
| "articles": [], | |
| "message": "No news articles found for this category" | |
| } | |
| # Format the news articles | |
| articles = [] | |
| for article in result[:10]: # Limit to 10 articles | |
| articles.append({ | |
| "id": article.get('id'), | |
| "headline": article.get('headline', 'No headline'), | |
| "summary": article.get('summary', ''), | |
| "source": article.get('source', 'Unknown'), | |
| "url": article.get('url'), | |
| "published": datetime.fromtimestamp(article.get('datetime', 0)).strftime('%Y-%m-%d %H:%M:%S') if article.get('datetime') else None, | |
| "image": article.get('image') | |
| }) | |
| return { | |
| "category": category, | |
| "total_articles": len(articles), | |
| "articles": articles | |
| } | |
| def get_company_news(symbol: str, from_date: Optional[str] = None, to_date: Optional[str] = None) -> dict: | |
| """ | |
| Get latest news for a specific company by stock symbol. This endpoint provides company-specific | |
| news, press releases, and announcements. Only available for North American companies. | |
| When to use: | |
| - User asks about a specific company's news (e.g., "Apple news", "Tesla updates") | |
| - Need company-specific announcements or press releases | |
| - User mentions "[company name] news", "recent [company] developments" | |
| - Want to filter news by date range | |
| Date range tips: | |
| - Default: Last 7 days if no dates specified | |
| - Can go back up to several years | |
| - Use YYYY-MM-DD format (e.g., "2024-01-01") | |
| Examples: | |
| - "What's the latest Apple news?" → get_company_news(symbol="AAPL") | |
| - "Tesla news from last month" → get_company_news(symbol="TSLA", from_date="2024-10-01", to_date="2024-10-31") | |
| - "Microsoft announcements this week" → get_company_news(symbol="MSFT") | |
| - "Show me Amazon news from January 2024" → get_company_news(symbol="AMZN", from_date="2024-01-01", to_date="2024-01-31") | |
| Args: | |
| symbol: Company stock ticker symbol (e.g., 'AAPL', 'MSFT', 'TSLA', 'GOOGL') | |
| Must be a North American (US/Canada) listed company | |
| from_date: Start date in YYYY-MM-DD format (default: 7 days ago) | |
| to_date: End date in YYYY-MM-DD format (default: today) | |
| Returns: | |
| dict: Company news data containing: | |
| - symbol: Stock ticker symbol | |
| - from_date: Start date of news search | |
| - to_date: End date of news search | |
| - total_articles: Number of articles found | |
| - articles: List of news articles (max 10), each with: | |
| * id: Article ID | |
| * headline: News headline | |
| * summary: Article summary | |
| * source: News source | |
| * url: Link to full article | |
| * published: Publication date and time | |
| * image: Article image URL (if available) | |
| * related_symbols: Other stock symbols mentioned | |
| """ | |
| # Set default dates if not provided | |
| if not to_date: | |
| to_date = datetime.now().strftime('%Y-%m-%d') | |
| if not from_date: | |
| from_date = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d') | |
| params = { | |
| "symbol": symbol.upper(), | |
| "from": from_date, | |
| "to": to_date | |
| } | |
| result = make_finnhub_request("company-news", params) | |
| if isinstance(result, dict) and "error" in result: | |
| return { | |
| "error": result["error"], | |
| "symbol": symbol.upper(), | |
| "from_date": from_date, | |
| "to_date": to_date | |
| } | |
| if not result or len(result) == 0: | |
| # Provide helpful suggestions based on the symbol | |
| suggestion = "Try expanding the date range or check if the symbol is correct." | |
| if symbol.upper() in ["BABA", "BIDU", "JD", "PDD", "NIO"]: | |
| suggestion = "Note: Chinese ADRs may have limited news coverage. Try US companies like AAPL, MSFT, TSLA, or GOOGL for better results." | |
| return { | |
| "symbol": symbol.upper(), | |
| "from_date": from_date, | |
| "to_date": to_date, | |
| "total_articles": 0, | |
| "articles": [], | |
| "message": f"No news articles found for {symbol.upper()} between {from_date} and {to_date}.", | |
| "suggestion": suggestion, | |
| "note": "Company news is only available for North American companies. Some companies may have limited news coverage during certain periods." | |
| } | |
| # Format the news articles | |
| articles = [] | |
| for article in result[:10]: # Limit to 10 articles | |
| articles.append({ | |
| "id": article.get('id'), | |
| "headline": article.get('headline', 'No headline'), | |
| "summary": article.get('summary', ''), | |
| "source": article.get('source', 'Unknown'), | |
| "url": article.get('url'), | |
| "published": datetime.fromtimestamp(article.get('datetime', 0)).strftime('%Y-%m-%d %H:%M:%S') if article.get('datetime') else None, | |
| "image": article.get('image'), | |
| "related_symbols": article.get('related', []) | |
| }) | |
| return { | |
| "symbol": symbol.upper(), | |
| "from_date": from_date, | |
| "to_date": to_date, | |
| "total_articles": len(articles), | |
| "articles": articles | |
| } | |
| # Gradio Interface Functions | |
| def configure_api_key(api_key: str) -> str: | |
| """Configure the Finnhub API key""" | |
| if not api_key or api_key.strip() == "": | |
| return "❌ Please provide a valid API key" | |
| set_api_key(api_key.strip()) | |
| return "✅ API Key configured successfully! You can now use the MCP tools." | |
| def test_quote_tool(symbol: str) -> str: | |
| """Test the Quote tool""" | |
| if not symbol or symbol.strip() == "": | |
| return "❌ Please provide a stock symbol" | |
| result = get_quote(symbol.strip()) | |
| if "error" in result: | |
| return f"❌ Error: {result['error']}" | |
| # Format for display | |
| output = f"""📊 Real-time Quote for {result['symbol']} | |
| Current Price: ${result.get('current_price', 'N/A')} | |
| Change: ${result.get('change', 'N/A')} | |
| Percent Change: {result.get('percent_change', 'N/A')}% | |
| High: ${result.get('high', 'N/A')} | |
| Low: ${result.get('low', 'N/A')} | |
| Open: ${result.get('open', 'N/A')} | |
| Previous Close: ${result.get('previous_close', 'N/A')} | |
| Timestamp: {result.get('timestamp', 'N/A')} | |
| """ | |
| return output.strip() | |
| def test_market_news_tool(category: str) -> str: | |
| """Test the Market News tool""" | |
| result = get_market_news(category) | |
| if "error" in result: | |
| return f"❌ Error: {result['error']}" | |
| if result.get('total_articles', 0) == 0: | |
| return result.get('message', 'No news articles found') | |
| # Format for display | |
| output = f"📰 Latest Market News ({result['category']})\n" | |
| output += f"Total Articles: {result['total_articles']}\n\n" | |
| for idx, article in enumerate(result['articles'], 1): | |
| output += f"{idx}. {article['headline']}\n" | |
| output += f" Source: {article['source']}\n" | |
| if article.get('summary'): | |
| summary = article['summary'][:200] + "..." if len(article['summary']) > 200 else article['summary'] | |
| output += f" Summary: {summary}\n" | |
| output += f" URL: {article.get('url', 'N/A')}\n" | |
| output += f" Published: {article.get('published', 'N/A')}\n\n" | |
| return output.strip() | |
| def test_company_news_tool(symbol: str, from_date: str, to_date: str) -> str: | |
| """Test the Company News tool""" | |
| if not symbol or symbol.strip() == "": | |
| return "❌ Please provide a stock symbol" | |
| # Use None if dates are empty | |
| from_d = from_date.strip() if from_date and from_date.strip() else None | |
| to_d = to_date.strip() if to_date and to_date.strip() else None | |
| result = get_company_news(symbol.strip(), from_d, to_d) | |
| if "error" in result: | |
| return f"❌ Error: {result['error']}" | |
| if result.get('total_articles', 0) == 0: | |
| # Show detailed message with suggestions | |
| output = f"⚠️ {result.get('message', 'No news articles found')}\n\n" | |
| if 'suggestion' in result: | |
| output += f"💡 {result['suggestion']}\n\n" | |
| if 'note' in result: | |
| output += f"📝 {result['note']}" | |
| return output | |
| # Format for display | |
| output = f"📰 Company News for {result['symbol']}\n" | |
| output += f"Period: {result['from_date']} to {result['to_date']}\n" | |
| output += f"Total Articles: {result['total_articles']}\n\n" | |
| for idx, article in enumerate(result['articles'], 1): | |
| output += f"{idx}. {article['headline']}\n" | |
| output += f" Source: {article['source']}\n" | |
| if article.get('summary'): | |
| summary = article['summary'][:200] + "..." if len(article['summary']) > 200 else article['summary'] | |
| output += f" Summary: {summary}\n" | |
| output += f" URL: {article.get('url', 'N/A')}\n" | |
| output += f" Published: {article.get('published', 'N/A')}\n\n" | |
| return output.strip() | |
| def check_server_health() -> str: | |
| """Check server health status and API connectivity""" | |
| status_parts = [] | |
| # Check API key configuration | |
| api_key = get_api_key() | |
| if not api_key: | |
| status_parts.append("❌ API Key: Not configured") | |
| status_parts.append("\n⚠️ Please configure your Finnhub API key in the Configuration tab or set FINNHUB_API_KEY environment variable.") | |
| return "\n".join(status_parts) | |
| else: | |
| status_parts.append("✅ API Key: Configured") | |
| # Test Finnhub API connectivity | |
| status_parts.append("\n🔍 Testing Finnhub API connectivity...") | |
| try: | |
| test_result = make_finnhub_request("quote", {"symbol": "AAPL"}) | |
| if "error" in test_result: | |
| status_parts.append(f"❌ Finnhub API: {test_result['error']}") | |
| status_parts.append("\n⚠️ Possible issues:") | |
| status_parts.append(" • Invalid API key") | |
| status_parts.append(" • Network connectivity problem") | |
| status_parts.append(" • Finnhub API service is down") | |
| status_parts.append(" • Rate limit exceeded") | |
| else: | |
| status_parts.append(f"✅ Finnhub API: Online and reachable") | |
| status_parts.append(f" Test query result: AAPL @ ${test_result.get('c', 'N/A')}") | |
| except Exception as e: | |
| status_parts.append(f"❌ Finnhub API: Connection failed") | |
| status_parts.append(f" Error: {str(e)}") | |
| return "\n".join(status_parts) | |
| # Load HTML content from file | |
| def load_html_content(): | |
| """Load the HTML welcome page""" | |
| try: | |
| # 优先使用精简版 | |
| if os.path.exists('index_simple.html'): | |
| with open('index_simple.html', 'r', encoding='utf-8') as f: | |
| return f.read() | |
| # 后备：Gradio 优化版 | |
| elif os.path.exists('index_gradio.html'): | |
| with open('index_gradio.html', 'r', encoding='utf-8') as f: | |
| return f.read() | |
| # 后备：完整版 | |
| elif os.path.exists('index.html'): | |
| with open('index.html', 'r', encoding='utf-8') as f: | |
| return f.read() | |
| else: | |
| return "<h1>📈 Finnhub MCP Server</h1><p>欢迎使用</p>" | |
| except Exception as e: | |
| print(f"Error loading HTML: {e}") | |
| return "<h1>📈 Finnhub MCP Server</h1><p>欢迎使用</p>" | |
| # Create Gradio Interface | |
| with gr.Blocks(title="Finnhub Market Info MCP Server") as demo: | |
| # Welcome tab with HTML content | |
| with gr.Tab("🏠 Home"): | |
| gr.HTML(load_html_content()) | |
| # Health Check tab | |
| with gr.Tab("🩺 Health Check"): | |
| gr.Markdown("### Server Health Status") | |
| gr.Markdown("Check if the MCP server and Finnhub API are working properly.") | |
| health_check_btn = gr.Button("Check Health", variant="primary", size="lg") | |
| health_status_output = gr.Textbox( | |
| label="Health Status", | |
| lines=15, | |
| interactive=False, | |
| placeholder="Click 'Check Health' to test the server and API connectivity..." | |
| ) | |
| health_check_btn.click( | |
| fn=check_server_health, | |
| inputs=[], | |
| outputs=[health_status_output] | |
| ) | |
| gr.Markdown(""" | |
| ### What this checks: | |
| 1. **API Key Configuration** - Verifies if Finnhub API key is set | |
| 2. **Finnhub API Connectivity** - Tests connection to Finnhub servers | |
| 3. **API Response** - Validates that the API returns data correctly | |
| ### Troubleshooting: | |
| If health check fails: | |
| - ❌ **API Key Not Configured**: Go to API Key Config tab and enter your key | |
| - ❌ **Invalid API Key**: Check if your key is correct at [finnhub.io](https://finnhub.io/dashboard) | |
| - ❌ **Network Error**: Check your internet connection | |
| - ❌ **Rate Limit**: Wait a minute before trying again (free tier: 60 calls/min) | |
| """) | |
| with gr.Tab("📊 Quote Tool"): | |
| gr.Markdown("### Test Real-time Stock Quote") | |
| gr.Markdown("Get real-time quote data for US stocks. Example symbols: AAPL, MSFT, TSLA, GOOGL") | |
| with gr.Row(): | |
| quote_symbol = gr.Textbox( | |
| label="Stock Symbol", | |
| placeholder="AAPL", | |
| value="AAPL" | |
| ) | |
| quote_btn = gr.Button("Get Quote", variant="primary") | |
| quote_output = gr.Textbox(label="Quote Data", lines=12, interactive=False) | |
| quote_btn.click( | |
| fn=test_quote_tool, | |
| inputs=[quote_symbol], | |
| outputs=[quote_output] | |
| ) | |
| with gr.Tab("📰 Market News Tool"): | |
| gr.Markdown("### Test Market News") | |
| gr.Markdown("Get latest market news by category") | |
| with gr.Row(): | |
| news_category = gr.Dropdown( | |
| choices=["general", "forex", "crypto", "merger"], | |
| label="News Category", | |
| value="general" | |
| ) | |
| market_news_btn = gr.Button("Get Market News", variant="primary") | |
| market_news_output = gr.Textbox(label="Market News", lines=15, interactive=False) | |
| market_news_btn.click( | |
| fn=test_market_news_tool, | |
| inputs=[news_category], | |
| outputs=[market_news_output] | |
| ) | |
| with gr.Tab("🏢 Company News Tool"): | |
| gr.Markdown("### Test Company News") | |
| gr.Markdown("Get latest company news by symbol. Only available for North American companies.") | |
| with gr.Row(): | |
| company_symbol = gr.Textbox( | |
| label="Stock Symbol", | |
| placeholder="AAPL", | |
| value="AAPL" | |
| ) | |
| with gr.Row(): | |
| company_from_date = gr.Textbox( | |
| label="From Date (YYYY-MM-DD)", | |
| placeholder="Leave empty for 7 days ago", | |
| value="" | |
| ) | |
| company_to_date = gr.Textbox( | |
| label="To Date (YYYY-MM-DD)", | |
| placeholder="Leave empty for today", | |
| value="" | |
| ) | |
| company_news_btn = gr.Button("Get Company News", variant="primary") | |
| company_news_output = gr.Textbox(label="Company News", lines=15, interactive=False) | |
| company_news_btn.click( | |
| fn=test_company_news_tool, | |
| inputs=[company_symbol, company_from_date, company_to_date], | |
| outputs=[company_news_output] | |
| ) | |
| with gr.Tab("ℹ️ MCP Connection"): | |
| gr.Markdown(""" | |
| ### Connect via MCP Client | |
| This server implements the **standard MCP protocol (JSON-RPC over SSE)**. | |
| #### 📥 Step 1: Clone Repository | |
| ```bash | |
| git clone https://huggingface.co/spaces/JC321/MarketandStockMCP | |
| cd MarketandStockMCP | |
| pip install -r requirements.txt | |
| ``` | |
| #### 🔑 Step 2: Get API Key | |
| Get your free Finnhub API key at [finnhub.io/register](https://finnhub.io/register) | |
| #### ⚙️ Step 3: Configure MCP Client | |
| **For Claude Desktop:** | |
| Edit config file: | |
| - macOS: `~/Library/Application Support/Claude/claude_desktop_config.json` | |
| - Windows: `%APPDATA%\Claude\claude_desktop_config.json` | |
| Add this configuration: | |
| ```json | |
| { | |
| "mcpServers": { | |
| "finnhub": { | |
| "command": "python", | |
| "args": ["C:\\path\\to\\MarketandStockMCP\\mcp_server.py"], | |
| "env": { | |
| "FINNHUB_API_KEY": "your_finnhub_api_key_here" | |
| } | |
| } | |
| } | |
| } | |
| ``` | |
| **For Cursor IDE:** | |
| Use the same JSON configuration in MCP settings. | |
| #### 🚀 Step 4: Test | |
| Restart your MCP client and try: | |
| - "What's the current price of Apple stock?" | |
| - "Show me the latest market news" | |
| - "Get Tesla company news from last week" | |
| #### 🛠️ Available Tools: | |
| - `get_quote(symbol: str)` - Real-time stock quote | |
| - `get_market_news(category: str, min_id: int)` - Market news by category | |
| - `get_company_news(symbol: str, from_date: str, to_date: str)` - Company-specific news | |
| #### 📖 Documentation: | |
| See [README.md](https://huggingface.co/spaces/JC321/MarketandStockMCP/blob/main/README.md) for detailed instructions. | |
| """) | |
| # API Key Config tab (moved to last) | |
| with gr.Tab("🔑 API Key Config"): | |
| gr.Markdown("### Configure Finnhub API Key for Web Testing") | |
| gr.Markdown(""" | |
| 💡 **Note:** This is only for testing tools in the web UI above. | |
| For MCP client use, configure the API key in your MCP client settings (see "MCP Connection" tab). | |
| """) | |
| with gr.Row(): | |
| api_key_input = gr.Textbox( | |
| label="Finnhub API Key", | |
| placeholder="Enter your Finnhub API key here...", | |
| type="password" | |
| ) | |
| config_btn = gr.Button("Configure", variant="primary") | |
| config_output = gr.Textbox(label="Status", interactive=False) | |
| config_btn.click( | |
| fn=configure_api_key, | |
| inputs=[api_key_input], | |
| outputs=[config_output] | |
| ) | |
| if __name__ == "__main__": | |
| import sys | |
| # Check for API key in environment variable | |
| env_api_key = os.getenv("FINNHUB_API_KEY") | |
| if env_api_key: | |
| set_api_key(env_api_key) | |
| print("✅ API Key loaded from environment variable") | |
| # Set port from environment (HF Space sets PORT=7860) | |
| port = int(os.getenv("PORT", "7860")) | |
| host = os.getenv("HOST", "0.0.0.0") | |
| # Check if running as MCP server or Gradio UI | |
| if "--mcp" in sys.argv or os.getenv("RUN_MCP_SERVER") == "true": | |
| # Run as MCP server (standalone mode) | |
| print("▶️ Starting MCP Server (standalone mode)...") | |
| mcp.run(transport="sse") | |
| else: | |
| # Launch Gradio interface | |
| print("▶️ Starting Gradio Interface...") | |
| print("🌐 Gradio UI available at: /") | |
| print("📡 MCP tools available through Gradio tabs") | |
| demo.launch( | |
| server_name=host, | |
| server_port=port, | |
| share=False | |
| ) | |