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 SEC EDGAR database.

    

    Args:

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

    

    Returns:

        dict: Company information including CIK, name, and ticker symbol

    """
    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 including name, tickers, SIC code, and industry description.

    

    Args:

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

    

    Returns:

        dict: Company information

    """
    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 list of company SEC filings (10-K, 10-Q, 20-F, etc.) with filing dates and document links.

    

    Args:

        cik: Company CIK code

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

    

    Returns:

        dict: Filings list with total count and limited results

    """
    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 period including revenue, net income, EPS, operating expenses, and cash flow.

    

    Args:

        cik: Company CIK code

        period: Period in format YYYY for annual or YYYYQX for quarterly (e.g., 2024, 2024Q3)

    

    Returns:

        dict: Financial data for the specified period

    """
    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 for multiple years including both annual and quarterly data.

    Returns data in chronological order (newest first): FY -> Q4 -> Q3 -> Q2 -> Q1.

    

    Args:

        cik: Company CIK code

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

    

    Returns:

        dict: Financial metrics with periods and data

    """
    if years < 1 or years > 10:
        return {"error": "Years parameter must be between 1 and 10"}
    
    # Check if company has filings (use tuple for caching)
    filings_10k = edgar_client.get_company_filings(cik, ('"10-K"',))
    filings_20f = edgar_client.get_company_filings(cik, ('"20-F"',))
    total_filings = len(filings_10k) + len(filings_20f)
    
    if total_filings == 0:
        return {
            "error": f"No annual filings found for CIK: {cik}",
            "suggestion": "Please check if the CIK is correct"
        }
    
    # Extract metrics
    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 debug info
        debug_info = {
            "error": f"No financial metrics extracted for CIK: {cik}",
            "debug": {
                "cik": cik,
                "years_requested": years,
                "filings_found": {
                    "10-K": len(filings_10k),
                    "20-F": len(filings_20f)
                },
                "latest_filings": []
            },
            "suggestion": "Try using get_latest_financial_data or get_financial_data with a specific period"
        }
        
        # Add latest filing dates
        all_filings = filings_10k + filings_20f
        for filing in all_filings[:5]:
            debug_info["debug"]["latest_filings"].append({
                "form": filing.get("form_type"),
                "date": filing.get("filing_date")
            })
        
        return debug_info


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

    Get the most recent financial data available for a company.

    

    Args:

        cik: Company CIK code

    

    Returns:

        dict: Latest financial data

    """
    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:
    """

    Advanced search supporting both company name and CIK code. Automatically detects input type.

    

    Args:

        company_input: Company name, ticker, or CIK code

    

    Returns:

        dict: Company information

    """
    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 from environment (HF Space sets PORT=7860)
    port = int(os.getenv("PORT", "7860"))
    host = os.getenv("HOST", "0.0.0.0")
    
    # 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 HTTP transport (stateless mode)
    mcp.run(transport="http")