mcp_server.py

#!/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)}"}


@mcp.tool()
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
    }


@mcp.tool()
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
    }


@mcp.tool()
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")