Upload 6 files

README.md

@@ -3,10 +3,11 @@ title: SEC Financial Data MCP Server
 emoji: 📊
 colorFrom: blue
 colorTo: green
-sdk: docker
-app_port: 7860
+sdk: gradio
+sdk_version: 4.44.0
 app_file: app.py
 pinned: false
+license: mit
 ---
 
 # SEC Financial Data MCP Server (FastMCP)

app.py

@@ -1,29 +1,16 @@
 """
-Simple FastAPI web interface for MCP Server usage guide
-Integrated with MCP server directly
+SEC Financial Data MCP Server with Gradio UI
+Provides standard MCP service + Web interface for testing
 """
-from fastapi import FastAPI, Request
-from fastapi.responses import HTMLResponse, JSONResponse, Response, FileResponse
-from fastapi.middleware.cors import CORSMiddleware
-from pathlib import Path
-import uvicorn
+import os
+import gradio as gr
 import json
-
-# Import MCP server components
 from mcp.server.fastmcp import FastMCP
 from edgar_client import EdgarDataClient
 from financial_analyzer import FinancialAnalyzer
 
-app = FastAPI(title="SEC Financial Data MCP Server")
-
-# Add CORS middleware
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
+# Initialize FastMCP server
+mcp = FastMCP("sec-financial-data")
 
 # Initialize EDGAR clients
 edgar_client = EdgarDataClient(
@@ -34,39 +21,23 @@ financial_analyzer = FinancialAnalyzer(
     user_agent="Juntao Peng Financial Report Metrics App (jtyxabc@gmail.com)"
 )
 
-# Create FastMCP server with pure JSON response
-mcp = FastMCP("sec-financial-data", json_response=True)
-
-
+# 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)
-    if result:
-        return result
-    else:
-        return {"error": f"No company found with 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)
-    if result:
-        return result
-    else:
-        return {"error": f"No company found with CIK: {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."""
-    # Convert list to tuple for caching
-    if form_types:
-        form_types_tuple = tuple(form_types)
-    else:
-        form_types_tuple = None
-    
+    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]
@@ -75,19 +46,13 @@ def get_company_filings(cik: str, form_types: list[str] | None = None) -> dict:
             "returned": len(limited_result),
             "filings": limited_result
         }
-    else:
-        return {"error": f"No filings found for CIK: {cik}"}
-
+    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)
-    if result and "period" in result:
-        return result
-    else:
-        return {"error": f"No financial data found for CIK: {cik}, Period: {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:
@@ -95,186 +60,145 @@ def extract_financial_metrics(cik: str, years: int = 3) -> dict:
     if years < 1 or years > 10:
         return {"error": "Years parameter must be between 1 and 10"}
     
-    try:
-        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 helpful debug info
-            filings_10k = edgar_client.get_company_filings(cik, ('10-K',))
-            filings_20f = edgar_client.get_company_filings(cik, ('20-F',))
-            
-            return {
-                "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": [
-                        {"form": f.get("form_type"), "date": f.get("filing_date")}
-                        for f in (filings_10k + filings_20f)[:5]
-                    ]
-                },
-                "suggestion": "Try using get_latest_financial_data or get_financial_data with a specific period"
-            }
-    except Exception as e:
-        return {
-            "error": f"Exception extracting metrics: {str(e)}",
-            "cik": cik,
-            "years": years
-        }
-
+    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)
-    if result and "period" in result:
-        return result
-    else:
-        return {"error": f"No latest financial data found for CIK: {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)
-    if result.get("error"):
-        return {"error": result["error"]}
-    return result
+    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"""
+    try:
+        result = search_company(company_name)
+        return json.dumps(result, indent=2)
+    except Exception as e:
+        return json.dumps({"error": str(e)}, indent=2)
 
-# Mount MCP server HTTP endpoint (support both /mcp and /sse for backward compatibility)
-@app.post("/mcp")
-@app.get("/mcp")
-@app.post("/sse")
-@app.get("/sse")
-async def mcp_http_endpoint(request: Request):
-    """MCP HTTP endpoint - handle both GET and POST (accessible via /mcp or /sse)"""
+def gradio_get_company_info(cik: str):
+    """Gradio wrapper for get_company_info"""
     try:
-        # For POST requests, get JSON body
-        if request.method == "POST":
-            body = await request.json()
-            
-            # Handle JSON-RPC request
-            if body.get("method") == "tools/list":
-                # List all tools
-                tools = []
-                for tool_name, tool_func in mcp._tool_manager._tools.items():
-                    tools.append({
-                        "name": tool_name,
-                        "description": tool_func.__doc__ or "",
-                        "inputSchema": {
-                            "type": "object",
-                            "properties": {},
-                            "required": []
-                        }
-                    })
-                
-                return JSONResponse({
-                    "jsonrpc": "2.0",
-                    "id": body.get("id"),
-                    "result": {"tools": tools}
-                })
-            
-            elif body.get("method") == "tools/call":
-                # Call a specific tool
-                tool_name = body.get("params", {}).get("name")
-                arguments = body.get("params", {}).get("arguments", {})
-                
-                # Directly call the tool functions we defined
-                tool_map = {
-                    "search_company": search_company,
-                    "get_company_info": get_company_info,
-                    "get_company_filings": get_company_filings,
-                    "get_financial_data": get_financial_data,
-                    "extract_financial_metrics": extract_financial_metrics,
-                    "get_latest_financial_data": get_latest_financial_data,
-                    "advanced_search_company": advanced_search_company
-                }
-                
-                if tool_name in tool_map:
-                    tool_func = tool_map[tool_name]
-                    result = tool_func(**arguments)
-                    
-                    return JSONResponse({
-                        "jsonrpc": "2.0",
-                        "id": body.get("id"),
-                        "result": {
-                            "content": [{
-                                "type": "text",
-                                "text": json.dumps(result)
-                            }]
-                        }
-                    })
-                else:
-                    return JSONResponse({
-                        "jsonrpc": "2.0",
-                        "id": body.get("id"),
-                        "error": {
-                            "code": -32601,
-                            "message": f"Tool not found: {tool_name}"
-                        }
-                    })
-            
-            else:
-                return JSONResponse({
-                    "jsonrpc": "2.0",
-                    "id": body.get("id"),
-                    "error": {
-                        "code": -32601,
-                        "message": f"Method not found: {body.get('method')}"
-                    }
-                })
-        
-        else:
-            # GET request - return info
-            return JSONResponse({
-                "service": "SEC Financial Data MCP Server",
-                "protocol": "MCP 2024-11-05",
-                "transport": "HTTP",
-                "status": "online"
-            })
-    
+        result = get_company_info(cik)
+        return json.dumps(result, indent=2)
     except Exception as e:
-        return JSONResponse(
-            content={"error": str(e), "type": "server_error"},
-            status_code=500
-        )
+        return json.dumps({"error": str(e)}, indent=2)
 
-# Get the template directory path
-TEMPLATES_DIR = Path(__file__).parent / "templates"
+def gradio_extract_metrics(cik: str, years: int):
+    """Gradio wrapper for extract_financial_metrics"""
+    try:
+        result = extract_financial_metrics(cik, years)
+        return json.dumps(result, indent=2)
+    except Exception as e:
+        return json.dumps({"error": str(e)}, indent=2)
 
-@app.get("/", response_class=HTMLResponse)
-async def root():
-    """Serve the main index page (ignores query parameters like ?logs=container)"""
-    index_path = TEMPLATES_DIR / "index.html"
-    if index_path.exists():
-        return FileResponse(index_path, media_type="text/html")
-    # Return error with debug info
-    return HTMLResponse(
-        f"<h1>Error: Template not found</h1>"
-        f"<p>Looking for: {index_path}</p>"
-        f"<p>Exists: {index_path.exists()}</p>"
-        f"<p>Current dir: {Path(__file__).parent}</p>",
-        status_code=404
-    )
+def gradio_get_latest(cik: str):
+    """Gradio wrapper for get_latest_financial_data"""
+    try:
+        result = get_latest_financial_data(cik)
+        return json.dumps(result, indent=2)
+    except Exception as e:
+        return json.dumps({"error": str(e)}, indent=2)
 
-@app.get("/test", response_class=HTMLResponse)
-async def test_page():
-    """Serve the test page"""
-    test_path = TEMPLATES_DIR / "test.html"
-    if test_path.exists():
-        return FileResponse(test_path)
-    return HTMLResponse("<h1>Error: Test page not found</h1>", status_code=404)
+# 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__":
-    uvicorn.run(app, host="0.0.0.0", port=7860)
+    demo.launch(server_name="0.0.0.0", server_port=7860)

requirements.txt

@@ -1,6 +1,15 @@
-mcp[cli]==1.2.0
+# MCP Server dependencies
+mcp>=1.0.0
+fastmcp>=0.1.0
+
+# Web UI
+gradio>=4.44.0,<5.0.0
+
+# Web framework (for MCP endpoints)
 fastapi==0.109.0
 uvicorn[standard]>=0.30
 pydantic>=2.10.1
+
+# SEC EDGAR data access
 sec-edgar-api==1.1.0
 requests==2.31.0