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