Upload app.py

app.py

@@ -41,6 +41,47 @@ def normalize_cik(cik):
     # 填充前导 0 至 10 位
     return cik_str.zfill(10) if cik_str else None
 
+def parse_mcp_response(response_data):
+    """
+    解析 MCP 协议响应数据
+    支持格式:
+    1. {"result": {"content": [{"type": "text", "text": "..."}]}}
+    2. {"content": [{"type": "text", "text": "..."}]}
+    3. 直接的 JSON 数据
+    """
+    if not isinstance(response_data, dict):
+        return response_data
+    
+    # 格式 1: {"result": {"content": [...]}}
+    if "result" in response_data and "content" in response_data["result"]:
+        content = response_data["result"]["content"]
+        if content and len(content) > 0:
+            text_content = content[0].get("text", "{}")
+            # 移除 emoji 前缀（如 "🔍 Search Result:\n\n"）
+            if ":\n\n" in text_content:
+                text_content = text_content.split(":\n\n", 1)[1]
+            try:
+                return json.loads(text_content)
+            except json.JSONDecodeError:
+                return text_content
+        return {}
+    
+    # 格式 2: {"content": [...]}
+    elif "content" in response_data:
+        content = response_data.get("content", [])
+        if content and len(content) > 0:
+            text_content = content[0].get("text", "{}")
+            if ":\n\n" in text_content:
+                text_content = text_content.split(":\n\n", 1)[1]
+            try:
+                return json.loads(text_content)
+            except json.JSONDecodeError:
+                return text_content
+        return {}
+    
+    # 格式 3: 直接返回
+    return response_data
+
 # MCP 工具定义
 def create_mcp_tools():
     """创建 MCP 工具列表"""
@@ -121,16 +162,8 @@ def query_financial_data(company_name, query_type):
         
         try:
             result = search_resp.json()
-            # MCP 协议返回格式: {"content": [{"type": "text", "text": "..."}]}
-            if isinstance(result, dict) and "content" in result:
-                content = result.get("content", [])
-                if content and len(content) > 0:
-                    text_content = content[0].get("text", "{}")
-                    company = json.loads(text_content) if isinstance(text_content, str) else text_content
-                else:
-                    company = {}
-            else:
-                company = result
+            # 使用统一的 MCP 响应解析函数
+            company = parse_mcp_response(result)
         except (ValueError, KeyError, json.JSONDecodeError) as e:
             return f"❌ JSON Parse Error: {str(e)}\n\nResponse: {search_resp.text[:500]}"
         
@@ -172,17 +205,9 @@ def query_financial_data(company_name, query_type):
             
             try:
                 data_result = data_resp.json()
-                # MCP 协议返回格式
-                if isinstance(data_result, dict) and "content" in data_result:
-                    content = data_result.get("content", [])
-                    if content and len(content) > 0:
-                        text_content = content[0].get("text", "{}")
-                        data = json.loads(text_content) if isinstance(text_content, str) else text_content
-                    else:
-                        data = {}
-                else:
-                    data = data_result
-            except (ValueError, KeyError) as e:
+                # 使用统一的 MCP 响应解析函数
+                data = parse_mcp_response(data_result)
+            except (ValueError, KeyError, json.JSONDecodeError) as e:
                 return result + f"❌ JSON Parse Error: {str(e)}\n\n{data_resp.text[:500]}"
             
             if isinstance(data, dict) and data.get("error"):
@@ -226,17 +251,9 @@ def query_financial_data(company_name, query_type):
             
             try:
                 metrics_result = metrics_resp.json()
-                # MCP 协议返回格式
-                if isinstance(metrics_result, dict) and "content" in metrics_result:
-                    content = metrics_result.get("content", [])
-                    if content and len(content) > 0:
-                        text_content = content[0].get("text", "{}")
-                        metrics = json.loads(text_content) if isinstance(text_content, str) else text_content
-                    else:
-                        metrics = {}
-                else:
-                    metrics = metrics_result
-            except (ValueError, KeyError) as e:
+                # 使用统一的 MCP 响应解析函数
+                metrics = parse_mcp_response(metrics_result)
+            except (ValueError, KeyError, json.JSONDecodeError) as e:
                 return result + f"❌ JSON Parse Error: {str(e)}\n\n{metrics_resp.text[:500]}"
             
             if isinstance(metrics, dict) and metrics.get("error"):
@@ -316,17 +333,9 @@ def query_financial_data(company_name, query_type):
             
             try:
                 metrics_result = metrics_resp.json()
-                # MCP 协议返回格���
-                if isinstance(metrics_result, dict) and "content" in metrics_result:
-                    content = metrics_result.get("content", [])
-                    if content and len(content) > 0:
-                        text_content = content[0].get("text", "{}")
-                        metrics = json.loads(text_content) if isinstance(text_content, str) else text_content
-                    else:
-                        metrics = {}
-                else:
-                    metrics = metrics_result
-            except (ValueError, KeyError) as e:
+                # 使用统一的 MCP 响应解析函数
+                metrics = parse_mcp_response(metrics_result)
+            except (ValueError, KeyError, json.JSONDecodeError) as e:
                 return result + f"❌ JSON Parse Error: {str(e)}\n\n{metrics_resp.text[:500]}"
             
             if isinstance(metrics, dict) and metrics.get("error"):
@@ -406,17 +415,9 @@ def query_financial_data(company_name, query_type):
             
             try:
                 filings_result = filings_resp.json()
-                # MCP 协议返回格式
-                if isinstance(filings_result, dict) and "content" in filings_result:
-                    content = filings_result.get("content", [])
-                    if content and len(content) > 0:
-                        text_content = content[0].get("text", "{}")
-                        filings_data = json.loads(text_content) if isinstance(text_content, str) else text_content
-                    else:
-                        filings_data = {}
-                else:
-                    filings_data = filings_result
-            except (ValueError, KeyError) as e:
+                # 使用统一的 MCP 响应解析函数
+                filings_data = parse_mcp_response(filings_result)
+            except (ValueError, KeyError, json.JSONDecodeError) as e:
                 return result + f"❌ JSON Parse Error: {str(e)}\n\n{filings_resp.text[:500]}"
             
             if isinstance(filings_data, dict) and filings_data.get("error"):