EasyFinancialAgent/easy_financial_mcp.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 EasyReportDataMCP.edgar_client import EdgarDataClient
from EasyReportDataMCP.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"}
    
    # ✅ 直接调用 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 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")