File size: 5,845 Bytes
716f1cd 88cda27 716f1cd 88cda27 716f1cd 1643511 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 |
"""
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")
|