Upload 2 files

mcp_server_fastmcp.py

@@ -0,0 +1,220 @@
+"""
+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")