app.py

"""
SEC Financial Data MCP Server with Gradio UI
Provides standard MCP service + Web interface for testing
"""
import os
import gradio as gr
import json
from mcp.server.fastmcp import FastMCP
from edgar_client import EdgarDataClient
from financial_analyzer import FinancialAnalyzer

# Initialize FastMCP server
mcp = FastMCP("sec-financial-data")

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

# Define MCP tools
@mcp.tool()
def search_company(company_name: str) -> dict:
    """Search for a company by name in SEC EDGAR database."""
    result = edgar_client.search_company_by_name(company_name)
    return result if result else {"error": f"No company found with name: {company_name}"}

@mcp.tool()
def get_company_info(cik: str) -> dict:
    """Get detailed company information."""
    result = edgar_client.get_company_info(cik)
    return result if result else {"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."""
    form_types_tuple = tuple(form_types) if form_types else None
    result = edgar_client.get_company_filings(cik, form_types_tuple)
    if result:
        limited_result = result[:20]
        return {
            "total": len(result),
            "returned": len(limited_result),
            "filings": limited_result
        }
    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."""
    result = edgar_client.get_financial_data_for_period(cik, period)
    return result if result and "period" in result else {"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."""
    if years < 1 or years > 10:
        return {"error": "Years parameter must be between 1 and 10"}
    
    metrics = financial_analyzer.extract_financial_metrics(cik, years)
    if metrics:
        formatted = financial_analyzer.format_financial_data(metrics)
        return {"periods": len(formatted), "data": formatted}
    return {"error": f"No financial metrics extracted for CIK: {cik}"}

@mcp.tool()
def get_latest_financial_data(cik: str) -> dict:
    """Get the most recent financial data available."""
    result = financial_analyzer.get_latest_financial_data(cik)
    return result if result and "period" in result else {"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."""
    result = financial_analyzer.search_company(company_input)
    return result if not result.get("error") else {"error": result["error"]}

# Gradio wrapper functions (添加调试和超时处理)
def gradio_search_company(company_name: str):
    """Gradio wrapper for search_company"""
    if not company_name or not company_name.strip():
        return json.dumps({"error": "Company name cannot be empty"}, indent=2)
    try:
        import sys
        print(f"[DEBUG] Searching company: {company_name.strip()}", file=sys.stderr)
        result = search_company(company_name.strip())
        print(f"[DEBUG] Search result type: {type(result)}", file=sys.stderr)
        print(f"[DEBUG] Search result: {result}", file=sys.stderr)
        # Ensure result is a dict
        if not isinstance(result, dict):
            result = {"error": f"Unexpected result type: {type(result)}"}
        return json.dumps(result, indent=2)
    except TimeoutError as e:
        return json.dumps({"error": f"Request timeout: {str(e)}"}, indent=2)
    except Exception as e:
        import traceback
        traceback.print_exc()
        return json.dumps({"error": f"Exception: {str(e)}", "type": str(type(e))}, indent=2)

def gradio_get_company_info(cik: str):
    """Gradio wrapper for get_company_info"""
    if not cik or not cik.strip():
        return json.dumps({"error": "CIK cannot be empty"}, indent=2)
    try:
        import sys
        print(f"[DEBUG] Getting company info for CIK: {cik.strip()}", file=sys.stderr)
        result = get_company_info(cik.strip())
        print(f"[DEBUG] Company info result: {result}", file=sys.stderr)
        if not isinstance(result, dict):
            result = {"error": f"Unexpected result type: {type(result)}"}
        return json.dumps(result, indent=2)
    except TimeoutError as e:
        return json.dumps({"error": f"Request timeout: {str(e)}"}, indent=2)
    except Exception as e:
        import traceback
        traceback.print_exc()
        return json.dumps({"error": f"Exception: {str(e)}", "type": str(type(e))}, indent=2)

def gradio_extract_metrics(cik: str, years: float):
    """Gradio wrapper for extract_financial_metrics"""
    if not cik or not cik.strip():
        return json.dumps({"error": "CIK cannot be empty"}, indent=2)
    try:
        import sys
        years_int = int(years)
        print(f"[DEBUG] Extracting metrics for CIK: {cik.strip()}, Years: {years_int}", file=sys.stderr)
        result = extract_financial_metrics(cik.strip(), years_int)
        print(f"[DEBUG] Extract metrics result: {result}", file=sys.stderr)
        if not isinstance(result, dict):
            result = {"error": f"Unexpected result type: {type(result)}"}
        return json.dumps(result, indent=2)
    except TimeoutError as e:
        return json.dumps({"error": f"Request timeout: {str(e)}"}, indent=2)
    except Exception as e:
        import traceback
        traceback.print_exc()
        return json.dumps({"error": f"Exception: {str(e)}", "type": str(type(e))}, indent=2)

def gradio_get_latest(cik: str):
    """Gradio wrapper for get_latest_financial_data"""
    if not cik or not cik.strip():
        return json.dumps({"error": "CIK cannot be empty"}, indent=2)
    try:
        import sys
        print(f"[DEBUG] Getting latest data for CIK: {cik.strip()}", file=sys.stderr)
        result = get_latest_financial_data(cik.strip())
        print(f"[DEBUG] Latest data result: {result}", file=sys.stderr)
        if not isinstance(result, dict):
            result = {"error": f"Unexpected result type: {type(result)}"}
        return json.dumps(result, indent=2)
    except TimeoutError as e:
        return json.dumps({"error": f"Request timeout: {str(e)}"}, indent=2)
    except Exception as e:
        import traceback
        traceback.print_exc()
        return json.dumps({"error": f"Exception: {str(e)}", "type": str(type(e))}, indent=2)

# Create Gradio interface
with gr.Blocks(title="SEC Financial Data MCP Server", theme=gr.themes.Soft()) as demo:
    gr.Markdown("""
    # 📊 SEC Financial Data MCP Server
    
    Access real-time SEC EDGAR financial data via Model Context Protocol
    
    **MCP Endpoint:** Use `mcp_server_fastmcp.py` for standard MCP client connections
    """)
    
    with gr.Tab("🔍 Search Company"):
        gr.Markdown("### Search for a company by name")
        with gr.Row():
            with gr.Column():
                company_input = gr.Textbox(label="Company Name", placeholder="Tesla", value="Tesla")
                search_btn = gr.Button("Search", variant="primary")
            with gr.Column():
                search_output = gr.Code(label="Result", language="json", lines=15)
        search_btn.click(gradio_search_company, inputs=company_input, outputs=search_output)
    
    with gr.Tab("ℹ️ Company Info"):
        gr.Markdown("### Get detailed company information")
        with gr.Row():
            with gr.Column():
                cik_input = gr.Textbox(label="Company CIK", placeholder="0001318605", value="0001318605")
                info_btn = gr.Button("Get Info", variant="primary")
            with gr.Column():
                info_output = gr.Code(label="Result", language="json", lines=15)
        info_btn.click(gradio_get_company_info, inputs=cik_input, outputs=info_output)
    
    with gr.Tab("📈 Financial Metrics"):
        gr.Markdown("### Extract multi-year financial metrics ⭐")
        with gr.Row():
            with gr.Column():
                metrics_cik = gr.Textbox(label="Company CIK", placeholder="0001318605", value="0001318605")
                metrics_years = gr.Slider(minimum=1, maximum=10, value=3, step=1, label="Years")
                metrics_btn = gr.Button("Extract Metrics", variant="primary")
            with gr.Column():
                metrics_output = gr.Code(label="Result", language="json", lines=20)
        metrics_btn.click(gradio_extract_metrics, inputs=[metrics_cik, metrics_years], outputs=metrics_output)
    
    with gr.Tab("🆕 Latest Data"):
        gr.Markdown("### Get latest financial data")
        with gr.Row():
            with gr.Column():
                latest_cik = gr.Textbox(label="Company CIK", placeholder="0001318605", value="0001318605")
                latest_btn = gr.Button("Get Latest", variant="primary")
            with gr.Column():
                latest_output = gr.Code(label="Result", language="json", lines=15)
        latest_btn.click(gradio_get_latest, inputs=latest_cik, outputs=latest_output)
    
    with gr.Tab("📖 Documentation"):
        gr.Markdown("""
        ## 🛠️ Available Tools (7)
        
        1. **search_company** - Search by company name
        2. **get_company_info** - Get company details by CIK
        3. **get_company_filings** - List SEC filings
        4. **get_financial_data** - Get specific period data
        5. **extract_financial_metrics** ⭐ - Multi-year trends
        6. **get_latest_financial_data** - Latest snapshot
        7. **advanced_search_company** - Flexible search
        
        ## 🔗 MCP Integration
        
        For MCP client integration, run:
        ```bash
        python mcp_server_fastmcp.py
        ```
        
        Then configure your MCP client (e.g., Claude Desktop):
        ```json
        {
          "mcpServers": {
            "sec-financial-data": {
              "command": "python",
              "args": ["path/to/mcp_server_fastmcp.py"]
            }
          }
        }
        ```
        
        ## 📊 Data Source
        
        - **SEC EDGAR API** - Official SEC data
        - **Financial Statements** - 10-K, 10-Q, 20-F forms
        - **XBRL Data** - Structured metrics
        """)

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860)