Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| Finnhub MCP Server - Standard MCP Protocol (JSON-RPC over SSE) | |
| This is a standalone MCP server that implements the standard Model Context Protocol. | |
| It provides three tools for accessing Finnhub financial data: | |
| - get_quote: Real-time stock quotes | |
| - get_market_news: Latest market news by category | |
| - get_company_news: Company-specific news | |
| Usage: | |
| python mcp_server.py | |
| Environment Variables: | |
| FINNHUB_API_KEY: Your Finnhub API key (required) | |
| """ | |
| import os | |
| import requests | |
| from typing import Optional, Dict, Any | |
| from datetime import datetime, timedelta | |
| from mcp.server.fastmcp import FastMCP | |
| # Initialize FastMCP server with standard MCP protocol | |
| mcp = FastMCP("finnhub-market-info") | |
| def get_api_key() -> Optional[str]: | |
| """Get the Finnhub API key from environment""" | |
| return 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 FINNHUB_API_KEY environment variable."} | |
| 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: Company stock symbol | |
| - from_date: Start date of news range | |
| - to_date: End date of news range | |
| - total_articles: Total number of articles returned | |
| - articles: List of news articles (max 10), each with: | |
| * headline: News headline | |
| * summary: Brief summary | |
| * source: News source | |
| * url: Link to full article | |
| * published: Publication timestamp | |
| * related: Related stock symbols (if any) | |
| """ | |
| # Set default date range 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 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({ | |
| "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, | |
| "related": article.get('related', '') | |
| }) | |
| return { | |
| "symbol": symbol.upper(), | |
| "from_date": from_date, | |
| "to_date": to_date, | |
| "total_articles": len(articles), | |
| "articles": articles | |
| } | |
| if __name__ == "__main__": | |
| # Check for API key | |
| api_key = get_api_key() | |
| if not api_key: | |
| print("❌ Error: FINNHUB_API_KEY environment variable is not set") | |
| print("Please set your Finnhub API key:") | |
| print(" export FINNHUB_API_KEY='your_api_key_here'") | |
| exit(1) | |
| print("✅ API Key loaded from environment variable") | |
| print("▶️ Starting Finnhub MCP Server (JSON-RPC over SSE)...") | |
| print("📡 Server name: finnhub-market-info") | |
| print("🔧 Protocol: Model Context Protocol (MCP)") | |
| print("🌐 Transport: SSE (Server-Sent Events)") | |
| print("") | |
| print("Available tools:") | |
| print(" • get_quote(symbol)") | |
| print(" • get_market_news(category, min_id)") | |
| print(" • get_company_news(symbol, from_date, to_date)") | |
| print("") | |
| # Run the MCP server with SSE transport | |
| mcp.run(transport="sse") | |