XTQuantAI

MIT License
19
  • Apple
#!/usr/bin/env python # -*- coding: utf-8 -*- """ xtquantai 启动脚本 提供两种启动方式: 1. 直接启动 xtquantai 服务器 2. 通过 MCP Inspector 启动 xtquantai 服务器 """ import os import sys import subprocess import argparse import importlib import asyncio import shutil def ensure_path(): """确保当前目录在 Python 路径中""" current_dir = os.path.dirname(os.path.abspath(__file__)) if current_dir not in sys.path: sys.path.insert(0, current_dir) def check_node_installed(): """检查 Node.js 是否已安装""" try: # 检查 node 命令是否可用 subprocess.run(["node", "--version"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True) return True except (subprocess.CalledProcessError, FileNotFoundError): return False def check_npx_installed(): """检查 npx 命令是否可用""" try: # 检查 npx 命令是否可用 subprocess.run(["npx", "--version"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True) return True except (subprocess.CalledProcessError, FileNotFoundError): return False def start_direct(): """直接启动 xtquantai 服务器""" ensure_path() try: from xtquantai import main print("正在直接启动 xtquantai 服务器...") main() except ImportError as e: print(f"导入 xtquantai 失败: {e}") print("尝试启动备选服务器...") try: # 尝试启动 server_direct.py if os.path.exists("server_direct.py"): print("找到 server_direct.py,启动独立服务器...") subprocess.run([sys.executable, "server_direct.py"], check=True) else: print("未找到 server_direct.py,无法启动服务器") print("请确保已安装 xtquantai 包或当前目录包含 xtquantai 源代码") sys.exit(1) except Exception as e: print(f"启动备选服务器失败: {e}") sys.exit(1) def start_with_inspector(python_path=None, venv_path=None): """通过 MCP Inspector 启动 xtquantai 服务器""" # 检查 Node.js 和 npx 是否已安装 if not check_node_installed(): print("未检测到 Node.js,无法使用 MCP Inspector 模式") print("正在切换到直接模式...") start_direct() return if not check_npx_installed(): print("未检测到 npx 命令,无法使用 MCP Inspector 模式") print("正在切换到直接模式...") start_direct() return cmd = ["npx", "@modelcontextprotocol/inspector"] if venv_path: cmd.extend(["uv", "run", "--venv", venv_path, "xtquantai"]) elif python_path: cmd.extend(["uv", "run", "--python", python_path, "xtquantai"]) else: cmd.extend(["uv", "run", "xtquantai"]) print(f"执行命令: {' '.join(cmd)}") try: subprocess.run(cmd, check=True) except subprocess.CalledProcessError as e: print(f"启动 MCP Inspector 失败: {e}") print("尝试切换到直接模式...") start_direct() except FileNotFoundError: print("找不到 npx 命令,请确保已安装 Node.js 和 npm") print("尝试切换到直接模式...") start_direct() def main(): """主函数,解析命令行参数并启动服务器""" parser = argparse.ArgumentParser(description="启动 xtquantai 服务器") parser.add_argument("--mode", choices=["direct", "inspector", "auto"], default="auto", help="启动模式: direct=直接启动, inspector=通过MCP Inspector启动, auto=自动选择") parser.add_argument("--python", help="Python 解释器路径") parser.add_argument("--venv", help="虚拟环境路径") parser.add_argument("--port", type=int, default=8000, help="直接模式下的服务器端口") args = parser.parse_args() if args.mode == "direct": start_direct() elif args.mode == "inspector": start_with_inspector(args.python, args.venv) else: # auto 模式 # 检查是否可以使用 MCP Inspector if check_node_installed() and check_npx_installed(): print("检测到 Node.js 和 npx,使用 MCP Inspector 模式") start_with_inspector(args.python, args.venv) else: print("未检测到 Node.js 或 npx,使用直接模式") start_direct() if __name__ == "__main__": main()