#!/usr/bin/env python3 """ 优化金十数据MCP连接 解决初始化时序问题 """ import json import requests import time import threading class Jin10Optimized: """优化的金十数据客户端""" def __init__(self, token: str): self.token = token self.base_url = "https://mcp.jin10.com/mcp" self.ready = False self.session_id = None def initialize_with_retry(self, max_retries=3): """带重试的初始化""" for attempt in range(max_retries): print(f"初始化尝试 {attempt + 1}/{max_retries}...") if self._initialize_single(): print("✅ MCP初始化成功") return True time.sleep(2) # 等待后重试 print("❌ MCP初始化失败，使用模拟模式") return False def _initialize_single(self): """单次初始化尝试""" try: # 创建独立会话 session = requests.Session() session.headers.update({ "Content-Type": "application/json", "Accept": "application/json, text/event-stream", "Authorization": f"Bearer {self.token}" }) # 1. 发送initialize init_request = { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-11-25", "capabilities": {}, "clientInfo": { "name": "情报大师-优化版", "version": "1.0.0" } } } response = session.post(self.base_url, json=init_request, timeout=30, stream=True) if response.status_code != 200: print(f"初始化请求失败: {response.status_code}") return False # 2. 读取initialize响应 init_response = None for line in response.iter_lines(): if line: line_str = line.decode('utf-8') if line_str.startswith('data: '): data = line_str[6:] try: json_data = json.loads(data) if json_data.get('id') == 1 and 'result' in json_data: init_response = json_data self.session_id = response.headers.get('mcp-session-id') break except: continue if not init_response: print("未收到初始化响应") return False # 3. 等待一段时间让服务器处理 time.sleep(1) # 4. 发送initialized通知（使用新请求） initialized_request = { "jsonrpc": "2.0", "method": "notifications/initialized", "params": {} } # 使用相同的会话和session_id if self.session_id: session.headers['mcp-session-id'] = self.session_id init_notify_response = session.post(self.base_url, json=initialized_request, timeout=10) if init_notify_response.status_code not in [200, 202]: print(f"initialized通知失败: {init_notify_response.status_code}") return False # 5. 等待服务器准备 time.sleep(2) # 6. 测试工具列表 tools_request = { "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} } tools_response = session.post(self.base_url, json=tools_request, timeout=30, stream=True) if tools_response.status_code != 200: print(f"工具列表请求失败: {tools_response.status_code}") return False # 检查是否有工具 has_tools = False for line in tools_response.iter_lines(): if line: line_str = line.decode('utf-8') if line_str.startswith('data: '): data = line_str[6:] try: json_data = json.loads(data) if 'result' in json_data and 'tools' in json_data['result']: tools = json_data['result']['tools'] print(f"✅ 获取到 {len(tools)} 个工具") for tool in tools[:3]: # 显示前3个 print(f" • {tool.get('name')}") has_tools = True break except: continue if has_tools: self.ready = True self.session = session return True else: print("未获取到工具列表") return False except Exception as e: print(f"初始化异常: {e}") return False def get_quote_simple(self, code: str): """简化版报价获取""" if not self.ready: print(f"⚠️ MCP未就绪，使用模拟数据: {code}") return self._mock_quote(code) print(f"获取报价: {code}") quote_request = { "jsonrpc": "2.0", "id": 100, "method": "tools/call", "params": { "name": "get_quote", "arguments": { "code": code } } } try: # 使用独立的请求，避免会话问题 headers = { "Content-Type": "application/json", "Accept": "application/json, text/event-stream", "Authorization": f"Bearer {self.token}" } if self.session_id: headers['mcp-session-id'] = self.session_id response = requests.post(self.base_url, headers=headers, json=quote_request, timeout=30, stream=True) if response.status_code == 200: for line in response.iter_lines(): if line: line_str = line.decode('utf-8') if line_str.startswith('data: '): data = line_str[6:] try: json_data = json.loads(data) if 'result' in json_data: print(f"✅ 获取到 {code} 数据") return json_data['result'].get('structuredContent', {}) elif 'error' in json_data: print(f"❌ {code} 错误: {json_data['error'].get('message')}") break except: continue print(f"⚠️ {code}: 未收到有效数据") else: print(f"❌ {code}: 请求失败 {response.status_code}") except Exception as e: print(f"❌ {code}: 异常 {e}") # 失败时返回模拟数据 return self._mock_quote(code) def _mock_quote(self, code: str): """模拟报价数据""" import random from datetime import datetime mock_prices = { "XAUUSD": 2350.50, "USOIL": 85.30, "USDJPY": 154.20, "EURUSD": 1.0850 } mock_names = { "XAUUSD": "现货黄金", "USOIL": "WTI原油", "USDJPY": "美元/日元", "EURUSD": "欧元/美元" } base_price = mock_prices.get(code, 100) base_name = mock_names.get(code, code) # 添加一些随机波动 price = base_price + random.uniform(-2, 2) change = random.uniform(-0.5, 0.5) return { "code": code, "name": base_name, "close": round(price, 2), "ups_percent": round(change, 2), "time": datetime.now().isoformat(), "source": "模拟数据" } def test_optimized(): """测试优化版本""" print("="*60) print("测试优化版金十数据连接") print("="*60) client = Jin10Optimized("sk-rxxce9SpTSYXWdYtyGPvEEQ22kizv8Nt_qPunm8ASOo") # 初始化 if client.initialize_with_retry(): print("\n✅ MCP连接就绪") # 测试获取报价 test_codes = ["XAUUSD", "USOIL", "USDJPY"] for code in test_codes: quote = client.get_quote_simple(code) if quote: print(f"{code}: {quote.get('name')} = {quote.get('close')} ({quote.get('ups_percent'):+.2f}%)") time.sleep(1) # 避免请求过快 else: print("\n⚠️ 使用模拟数据模式") # 测试模拟数据 test_codes = ["XAUUSD", "USOIL"] for code in test_codes: quote = client._mock_quote(code) print(f"{code}: {quote.get('name')} = {quote.get('close')} ({quote.get('ups_percent'):+.2f}%)") print("\n" + "="*60) print("测试完成") print("="*60) if __name__ == "__main__": test_optimized()