EasyReportDataMCP/mcp_server_fastmcp.py

"""

MCP Server for SEC EDGAR Financial Data - FastMCP Implementation

Uses Anthropic official FastMCP SDK for cleaner, more maintainable code

"""

from mcp.server.fastmcp import FastMCP
from edgar_client import EdgarDataClient
from financial_analyzer import FinancialAnalyzer

# Initialize EDGAR clients
edgar_client = EdgarDataClient(
    user_agent="Juntao Peng Financial Report Metrics App (jtyxabc@gmail.com)"
)

financial_analyzer = FinancialAnalyzer(
    user_agent="Juntao Peng Financial Report Metrics App (jtyxabc@gmail.com)"
)

# Create FastMCP server with pure JSON response and stateless HTTP
mcp = FastMCP("sec-financial-data", json_response=True, stateless_http=True)


@mcp.tool()
def search_company(company_name: str) -> dict:
    """

    Search for a company by name in the SEC EDGAR database. Use this tool when the user mentions 

    a company name or asks about a company without providing its CIK code. This tool will find the 

    company's official information needed for other financial queries.

    

    When to use:

    - User mentions a company name (e.g., "Tesla", "Apple", "Microsoft")

    - Need to find a company's CIK code for other tool calls

    - User asks "tell me about [company name]"

    - Need to verify company ticker symbols

    

    Examples:

    - "Search for Tesla" → Returns Tesla's CIK, ticker (TSLA), and industry info

    - "Find Apple" → Returns Apple's CIK, ticker (AAPL), and classification

    - "What's Microsoft's CIK?" → Returns CIK code and full company details

    

    Args:

        company_name: Company name to search (e.g., "Microsoft", "Apple Inc", "Tesla Motors")

    

    Returns:

        dict: Company information containing:

              - cik: Company Central Index Key (unique identifier needed for other tools)

              - name: Official company name registered with SEC

              - tickers: Stock ticker symbol(s) (e.g., ["TSLA"], ["AAPL"])

              - sic: Standard Industrial Classification code

              - sic_description: Industry/sector description

    """
    result = edgar_client.search_company_by_name(company_name)
    if result:
        return result
    else:
        return {"error": f"No company found with name: {company_name}"}


@mcp.tool()
def get_company_info(cik: str) -> dict:
    """

    Get detailed company information using CIK code. Use this when you already have a company's 

    CIK code and need to retrieve or verify its official details.

    

    When to use:

    - Already have a CIK code from search_company

    - Need to verify company details

    - User provides a CIK code directly

    

    Args:

        cik: Company CIK code in 10-digit format (e.g., "0000789019" for Microsoft)

    

    Returns:

        dict: Detailed company information including name, tickers, SIC code, and industry description

    """
    result = edgar_client.get_company_info(cik)
    if result:
        return result
    else:
        return {"error": f"No company found with CIK: {cik}"}


@mcp.tool()
def get_company_filings(cik: str, form_types: list[str] | None = None) -> dict:
    """

    Get a list of SEC filings for a company. SEC filings are official documents companies must 

    submit, including annual reports (10-K), quarterly reports (10-Q), and foreign company 

    annual reports (20-F). Use this to see what reports are available or to get filing dates.

    

    When to use:

    - User asks "what reports has [company] filed?"

    - Need to see filing history or dates

    - Want to know what documents are available

    - Checking if specific report types exist

    

    Common form types:

    - 10-K: Annual report (comprehensive yearly financial statement)

    - 10-Q: Quarterly report (financial updates every 3 months)

    - 20-F: Annual report for foreign companies

    - 8-K: Current report (major events/changes)

    

    Args:

        cik: Company CIK code

        form_types: Optional list to filter by specific form types (e.g., ["10-K", "10-Q"])

                   If None, returns all filing types

    

    Returns:

        dict: Filing information containing:

              - total: Total number of filings found

              - returned: Number of filings in response (max 20)

              - filings: List of filing details with dates, form types, and document links

    """
    # Convert list to tuple for caching compatibility
    if form_types:
        form_types = tuple(form_types)
    result = edgar_client.get_company_filings(cik, form_types)
    if result:
        limited_result = result[:20]
        return {
            "total": len(result),
            "returned": len(limited_result),
            "filings": limited_result
        }
    else:
        return {"error": f"No filings found for CIK: {cik}"}


@mcp.tool()
def get_financial_data(cik: str, period: str) -> dict:
    """

    Get financial data for a specific time period (year or quarter). Use this when the user asks 

    about financials for a particular period, like "2024 results" or "Q3 2024 performance".

    

    When to use:

    - User specifies a particular year (e.g., "2024 financials")

    - User asks about a specific quarter (e.g., "Q3 2024 results")

    - Need data for a single, specific time period

    - Comparing specific periods (call multiple times)

    

    Period format:

    - Annual: "YYYY" (e.g., "2024" for fiscal year 2024)

    - Quarterly: "YYYYQX" (e.g., "2024Q3" for Q3 of 2024, "2023Q4" for Q4 of 2023)

    

    Args:

        cik: Company CIK code

        period: Time period in format "YYYY" for annual or "YYYYQX" for quarterly

                Examples: "2024", "2023", "2024Q3", "2023Q4"

    

    Returns:

        dict: Financial metrics for the specified period including:

              - period: Time identifier (e.g., "FY2024", "2024Q3")

              - total_revenue: Total sales/revenue for the period

              - net_income: Profit (or loss) after all expenses

              - earnings_per_share: Profit per share of stock (EPS)

              - operating_expenses: Costs of running business operations

              - operating_cash_flow: Cash generated from business operations

              - source_url: Link to the SEC filing document

              - source_form: Type of SEC form (10-K, 10-Q, or 20-F)

              - data_source: Data source standard (us-gaap or ifrs-full)

    """
    result = edgar_client.get_financial_data_for_period(cik, period)
    if result and "period" in result:
        return result
    else:
        return {"error": f"No financial data found for CIK: {cik}, Period: {period}"}


@mcp.tool()
def extract_financial_metrics(cik: str, years: int = 3) -> dict:
    """

    Extract comprehensive financial metrics spanning multiple years with both annual and quarterly 

    data. This is the MOST POWERFUL tool for financial analysis - it returns complete multi-year 

    trends including all quarters. Perfect for understanding company performance over time, 

    identifying growth patterns, and comprehensive financial analysis.

    

    When to use (RECOMMENDED for most financial analysis):

    - User asks about "trends over time"

    - Questions about "growth", "performance over years"

    - "Show me [company]'s financials" (without specifying a period)

    - Comparative analysis needs

    - "How has [company] been doing?"

    - ANY request for multiple periods of data

    

    What makes this tool special:

    - Returns BOTH annual (FY) and quarterly (Q1-Q4) data

    - Sorted newest to oldest (FY2024 → Q4 → Q3 → Q2 → Q1 → FY2023...)

    - Includes multiple years in one call (saves time)

    - Ideal for trend analysis and year-over-year comparisons

    

    Example use cases:

    - "Show Tesla's financial trends for 3 years" → Perfect use case

    - "How has Apple's revenue grown?" → Use this (default 3 years)

    - "Compare Microsoft's quarterly performance" → Returns all quarters

    - "What are Amazon's financial metrics?" → Comprehensive overview

    

    Args:

        cik: Company CIK code

        years: Number of recent years to extract (1-10, default: 3)

               - 3 years = ~15 data points (3 annual + ~12 quarterly)

               - 5 years = ~25 data points (5 annual + ~20 quarterly)

               - More years = more comprehensive trend analysis

    

    Returns:

        dict: Comprehensive financial dataset containing:

              - periods: Total number of time periods returned

              - data: List of financial records, each with:

                * period: Time identifier (e.g., "FY2024", "2024Q3", "2023Q1")

                * total_revenue: Company's total sales/revenue for that period

                * net_income: Profit after all expenses (can be negative for losses)

                * earnings_per_share: Profit per share of stock (EPS)

                * operating_expenses: Costs of running the business

                * operating_cash_flow: Actual cash generated from operations

                * source_url: Link to SEC filing document

                * source_form: SEC form type (10-K for annual, 10-Q for quarterly, 20-F for foreign companies)

                * data_source: Data standard used (us-gaap or ifrs-full)

                * _sequence: Internal ordering number

              

              Data is sorted newest first: FY2024 → 2024Q4 → 2024Q3 → 2024Q2 → 2024Q1 → FY2023...

              

              Note: The underlying data structure is optimized to reduce redundancy.

              Each period's metadata (form, dates, URLs) is stored efficiently.

    """
    if years < 1 or years > 10:
        return {"error": "Years parameter must be between 1 and 10"}
    
    # ✅ 直接调用 extract_financial_metrics,不做预检查
    # extract_financial_metrics 内部会处理所有情况(10-K, 20-F, 数据缺失等)
    metrics = financial_analyzer.extract_financial_metrics(cik, years)
    
    if metrics:
        formatted = financial_analyzer.format_financial_data(metrics)
        return {
            "periods": len(formatted),
            "data": formatted
        }
    else:
        # ✅ 如果没有数据,返回简洁错误信息
        return {
            "error": f"No financial metrics found for CIK: {cik}",
            "suggestion": "Please verify the CIK is correct or try get_latest_financial_data"
        }


@mcp.tool()
def get_latest_financial_data(cik: str) -> dict:
    """

    Get the most recent financial snapshot for a company - returns only the latest annual report 

    data. Use this for quick checks of current financial status or when the user asks about 

    "latest" or "most recent" results without needing historical data.

    

    When to use:

    - User asks "what are [company]'s latest financials?"

    - "How is [company] doing currently?"

    - "Show me [company]'s most recent results"

    - Quick status check without historical context

    - Need just the newest data point (faster than extract_financial_metrics)

    

    What this returns:

    - Only the most recent ANNUAL (fiscal year) data

    - Does NOT include quarterly breakdowns

    - Fastest way to get current snapshot

    

    Examples:

    - "What's Tesla's latest revenue?" → Returns most recent annual revenue

    - "How much did Apple earn recently?" → Returns latest annual net income

    - "Show me Microsoft's current financials" → Returns latest fiscal year data

    

    Args:

        cik: Company CIK code

    

    Returns:

        dict: Latest financial data from most recent fiscal year including:

              - period: The fiscal year (e.g., "FY2024")

              - total_revenue: Most recent annual revenue

              - net_income: Most recent annual profit

              - earnings_per_share: Latest annual EPS

              - operating_expenses: Latest annual operating costs

              - operating_cash_flow: Latest annual cash from operations

              - source_url: Link to the SEC filing

              - source_form: SEC form type (usually 10-K or 20-F)

    """
    result = financial_analyzer.get_latest_financial_data(cik)
    if result and "period" in result:
        return result
    else:
        return {
            "error": f"No latest financial data found for CIK: {cik}"
        }


@mcp.tool()
def advanced_search_company(company_input: str) -> dict:
    """

    Flexible smart search that accepts ANY type of company identifier - company name, stock ticker 

    symbol (like TSLA, AAPL, MSFT), or CIK code. The tool automatically detects what type of 

    identifier you provide. Use this when you're uncertain what type of identifier the user gave.

    

    When to use:

    - User provides just a ticker symbol (e.g., "TSLA", "AAPL")

    - Unclear if user gave name, ticker, or CIK

    - Want most flexible search option

    - User input could be any identifier type

    

    What it accepts:

    - Company names: "Tesla", "Apple Inc", "Microsoft Corporation"

    - Ticker symbols: "TSLA", "AAPL", "MSFT", "GOOGL"

    - CIK codes: "0001318605", "0000320193"

    

    Examples:

    - Input: "TSLA" → Recognizes as ticker, returns Tesla info

    - Input: "Tesla" → Searches by name, returns Tesla info

    - Input: "0001318605" → Recognizes as CIK, returns Tesla info

    - Input: "AAPL" → Returns Apple information

    

    Args:

        company_input: Any company identifier - name ("Tesla"), ticker ("TSLA"), or CIK ("0001318605")

    

    Returns:

        dict: Complete company information including:

              - cik: Company's Central Index Key

              - name: Official registered company name

              - tickers: Stock ticker symbol(s)

              - sic: Standard Industrial Classification code

              - sic_description: Industry/sector description

    """
    result = financial_analyzer.search_company(company_input)
    if result.get("error"):
        return {"error": result["error"]}
    return result


# For production deployment
if __name__ == "__main__":
    import os
    
    # ✅ Set port to 7861 for EasyReportDataMCP (MarketandStockMCP uses 7870)
    port = int(os.getenv("PORT", "7861"))
    host = os.getenv("HOST", "0.0.0.0")
    
    print("▶️ Starting EasyReportDataMCP Server...")
    print(f"📡 MCP server will listen on {host}:{port}")
    print("✅ Available tools: advanced_search_company, get_latest_financial_data, extract_financial_metrics")
    print(f"🔗 MCP endpoint: http://{host}:{port}/mcp")
    
    # ✅ Monkeypatch uvicorn.Config to use our port
    import uvicorn
    original_config_init = uvicorn.Config.__init__
    
    def patched_init(self, *args, **kwargs):
        kwargs['host'] = host
        kwargs['port'] = port
        return original_config_init(self, *args, **kwargs)
    
    uvicorn.Config.__init__ = patched_init
    
    # ✅ Run FastMCP with SSE transport + custom /mcp endpoint
    # Add stateless HTTP endpoint for direct tool calls
    from starlette.applications import Starlette
    from starlette.routing import Mount, Route
    from starlette.responses import JSONResponse
    from mcp.server.sse import SseServerTransport
    import anyio
    
    async def handle_mcp_post(request):
        """Handle direct JSON-RPC POST requests to /mcp endpoint"""
        try:
            json_data = await request.json()
            method = json_data.get("method")
            params = json_data.get("params", {})
            
            if method == "tools/call":
                tool_name = params.get("name")
                arguments = params.get("arguments", {})
                
                # Call the tool directly
                if tool_name == "search_company":
                    result = search_company(**arguments)
                elif tool_name == "get_company_info":
                    result = get_company_info(**arguments)
                elif tool_name == "get_company_filings":
                    result = get_company_filings(**arguments)
                elif tool_name == "get_financial_data":
                    result = get_financial_data(**arguments)
                elif tool_name == "extract_financial_metrics":
                    result = extract_financial_metrics(**arguments)
                elif tool_name == "get_latest_financial_data":
                    result = get_latest_financial_data(**arguments)
                elif tool_name == "advanced_search_company":
                    result = advanced_search_company(**arguments)
                else:
                    return JSONResponse({"jsonrpc": "2.0", "id": json_data.get("id"), "error": {"code": -32601, "message": f"Unknown tool: {tool_name}"}}, status_code=200)
                
                # Return MCP-formatted response with JSON serialization
                import json as json_module
                return JSONResponse({
                    "jsonrpc": "2.0",
                    "id": json_data.get("id"),
                    "result": {
                        "content": [{"type": "text", "text": json_module.dumps(result, ensure_ascii=False)}]
                    }
                }, status_code=200)
            elif method == "tools/list":
                return JSONResponse({
                    "jsonrpc": "2.0",
                    "id": json_data.get("id"),
                    "result": {"tools": []}
                }, status_code=200)
            else:
                return JSONResponse({"jsonrpc": "2.0", "id": json_data.get("id"), "error": {"code": -32601, "message": f"Unknown method: {method}"}}, status_code=200)
        except Exception as e:
            import traceback
            traceback.print_exc()
            return JSONResponse({"jsonrpc": "2.0", "id": 1, "error": {"code": -32603, "message": str(e)}}, status_code=500)
    
    async def run_custom_sse():
        sse = SseServerTransport("/messages/")
        
        async def handle_sse(request):
            async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
                await mcp._mcp_server.run(
                    streams[0],
                    streams[1],
                    mcp._mcp_server.create_initialization_options(),
                )
        
        starlette_app = Starlette(
            debug=False,
            routes=[
                Route("/sse", endpoint=handle_sse),
                Mount("/messages/", app=sse.handle_post_message),
                Route("/mcp", endpoint=handle_mcp_post, methods=["POST"]),  # Custom stateless endpoint
            ],
        )
        
        config = uvicorn.Config(
            starlette_app,
            host=host,
            port=port,
            log_level="info",
        )
        server = uvicorn.Server(config)
        await server.serve()
    
    anyio.run(run_custom_sse)