#!/usr/bin/env python3 """ 正确的金十数据MCP API测试 """ import json import requests TOKEN = "sk-rxxce9SpTSYXWdYtyGPvEEQ22kizv8Nt_qPunm8ASOo" URL = "https://mcp.jin10.com/mcp" headers = { "Content-Type": "application/json", "Accept": "application/json, text/event-stream", "Authorization": f"Bearer {TOKEN}" } def test_mcp_flow(): """测试完整MCP流程""" print("🔍 测试金十数据MCP API...") # 1. 初始化 print("\n1. 初始化连接...") init_request = { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-11-25", "capabilities": {}, "clientInfo": { "name": "OpenClaw-情报大师", "version": "1.0.0" } } } try: response = requests.post(URL, headers=headers, json=init_request, timeout=10, stream=True) print(f" 状态码: {response.status_code}") if response.status_code == 200: print(" ✅ 初始化请求成功") # 读取SSE响应 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) print(f" 收到响应: {json.dumps(json_data, indent=2, ensure_ascii=False)}") if 'result' in json_data: print("\n 🎉 初始化成功！") server_info = json_data['result'].get('serverInfo', {}) print(f" 服务器: {server_info.get('name')} v{server_info.get('version')}") print(f" 标题: {server_info.get('title')}") # 2. 发送initialized通知 return send_initialized() except json.JSONDecodeError: print(f" 无法解析JSON: {data[:100]}") except Exception as e: print(f" 处理响应异常: {e}") print(" ❌ 未收到有效响应") return False else: print(f" ❌ 请求失败: {response.status_code}") print(f" 响应: {response.text[:200]}") return False except Exception as e: print(f" ❌ 请求异常: {e}") return False def send_initialized(): """发送initialized通知""" print("\n2. 发送initialized通知...") initialized_request = { "jsonrpc": "2.0", "method": "notifications/initialized", "params": {} } try: response = requests.post(URL, headers=headers, json=initialized_request, timeout=10) print(f" 状态码: {response.status_code}") if response.status_code == 200: print(" ✅ initialized通知已发送") # 3. 测试获取工具列表 return test_tools_list() else: print(f" ❌ 发送失败: {response.status_code}") print(f" 响应: {response.text[:200]}") return False except Exception as e: print(f" ❌ 请求异常: {e}") return False def test_tools_list(): """测试获取工具列表""" print("\n3. 获取工具列表...") tools_request = { "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} } try: response = requests.post(URL, headers=headers, json=tools_request, timeout=10, stream=True) print(f" 状态码: {response.status_code}") if response.status_code == 200: print(" ✅ 工具列表请求成功") 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 and 'tools' in json_data['result']: tools = json_data['result']['tools'] print(f"\n 🎯 获取到 {len(tools)} 个工具:") for i, tool in enumerate(tools[:5]): # 只显示前5个 print(f" {i+1}. {tool.get('name')}: {tool.get('description', '')[:80]}...") if len(tools) > 5: print(f" ... 还有 {len(tools)-5} 个工具") # 4. 测试调用工具 return test_get_quote() except json.JSONDecodeError: print(f" 无法解析JSON: {data[:100]}") except Exception as e: print(f" 处理响应异常: {e}") print(" ❌ 未收到工具列表") return False else: print(f" ❌ 请求失败: {response.status_code}") print(f" 响应: {response.text[:200]}") return False except Exception as e: print(f" ❌ 请求异常: {e}") return False def test_get_quote(): """测试调用get_quote工具""" print("\n4. 测试get_quote工具...") quote_request = { "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "get_quote", "arguments": { "code": "XAUUSD" } } } try: response = requests.post(URL, headers=headers, json=quote_request, timeout=10, stream=True) print(f" 状态码: {response.status_code}") if response.status_code == 200: print(" ✅ 报价请求成功") 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: result = json_data['result'] print("\n 📊 获取报价成功！") if 'structuredContent' in result: quote = result['structuredContent'] print(f" 品种: {quote.get('code')} ({quote.get('name')})") print(f" 时间: {quote.get('time')}") print(f" 价格: {quote.get('close')}") print(f" 涨跌: {quote.get('ups_price')} ({quote.get('ups_percent')}%)") print(f" 区间: {quote.get('low')} - {quote.get('high')}") print(f" 开盘: {quote.get('open')}, 成交量: {quote.get('volume')}") if 'content' in result: print(f"\n 摘要: {result['content'][:100]}...") return True elif 'error' in json_data: print(f" ❌ 错误: {json_data['error']}") return False except json.JSONDecodeError: print(f" 无法解析JSON: {data[:100]}") except Exception as e: print(f" 处理响应异常: {e}") print(" ❌ 未收到报价数据") return False else: print(f" ❌ 请求失败: {response.status_code}") print(f" 响应: {response.text[:200]}") return False except Exception as e: print(f" ❌ 请求异常: {e}") return False def main(): """主函数""" print("=" * 60) print("金十数据MCP API完整测试") print("=" * 60) print(f"Token: {TOKEN[:10]}...") print(f"API端点: {URL}") success = test_mcp_flow() print("\n" + "=" * 60) if success: print("🎉 测试成功！MCP API工作正常") print("\n可用功能:") print("• 实时行情查询 (get_quote)") print("• K线数据获取 (get_kline)") print("• 财经快讯 (list_flash, search_flash)") print("• 深度资讯 (list_news, search_news)") print("• 财经日历 (list_calendar)") else: print("❌ 测试失败") print("\n可能原因:") print("1. Token已过期或无效") print("2. API端点有变化") print("3. 网络连接问题") print("4. 服务器暂时不可用") if __name__ == "__main__": main()