smart-start.py•7.36 kB
#!/usr/bin/env python3
"""
MCP FeedForward 智能启动器
自动检测环境并选择最佳启动方式
"""
import os
import sys
import subprocess
import platform
import socket
import argparse
def check_port(port):
"""检查端口是否可用"""
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
return s.connect_ex(('localhost', port)) != 0
except:
return False
def find_available_port(start_port=8000):
"""查找可用端口"""
for port in range(start_port, start_port + 100):
if check_port(port):
return port
raise Exception("未找到可用端口")
def detect_environment():
"""检测运行环境"""
is_docker = os.path.exists('/.dockerenv')
is_ssh = 'SSH_CLIENT' in os.environ or 'SSH_TTY' in os.environ
has_display = 'DISPLAY' in os.environ or platform.system() == 'Windows'
is_windows = platform.system() == 'Windows'
is_linux = platform.system() == 'Linux'
is_mac = platform.system() == 'Darwin'
# 检测是否在服务器环境
is_server = is_ssh or is_docker or (is_linux and not has_display)
return {
'docker': is_docker,
'ssh': is_ssh,
'has_display': has_display,
'windows': is_windows,
'linux': is_linux,
'mac': is_mac,
'server': is_server,
'recommend_web': is_server
}
def check_dependencies():
"""检查依赖是否安装"""
deps = {
'fastmcp': False,
'pillow': False,
'flask': False,
'tkinter': False
}
try:
import fastmcp
deps['fastmcp'] = True
except ImportError:
pass
try:
import PIL
deps['pillow'] = True
except ImportError:
pass
try:
import flask
deps['flask'] = True
except ImportError:
pass
try:
import tkinter
deps['tkinter'] = True
except ImportError:
pass
return deps
def install_dependencies(mode='auto'):
"""安装缺失的依赖"""
deps = check_dependencies()
missing = []
if not deps['fastmcp']:
missing.append('fastmcp>=0.0.4')
if not deps['pillow']:
missing.append('pillow>=8.0.0')
if mode == 'web' and not deps['flask']:
missing.extend(['flask>=2.0.0', 'werkzeug>=2.0.0'])
if missing:
print(f"📦 安装缺失依赖: {', '.join(missing)}")
try:
subprocess.check_call([sys.executable, '-m', 'pip', 'install'] + missing)
print("✅ 依赖安装完成")
return True
except subprocess.CalledProcessError:
print("❌ 依赖安装失败")
return False
return True
def main():
"""智能启动主函数"""
parser = argparse.ArgumentParser(description="MCP FeedForward 智能启动器")
parser.add_argument('--mode', choices=['auto', 'gui', 'web'], default='auto',
help='启动模式 (默认: auto)')
parser.add_argument('--port', type=int, default=8000,
help='Web模式端口 (默认: 8000)')
parser.add_argument('--host', default='0.0.0.0',
help='Web模式主机 (默认: 0.0.0.0)')
parser.add_argument('--install-deps', action='store_true',
help='自动安装缺失的依赖')
args = parser.parse_args()
print("🚀 MCP FeedForward 智能启动器")
print("=" * 60)
# 检测环境
env = detect_environment()
print(f"🔍 环境检测:")
print(f" 操作系统: {platform.system()} {platform.release()}")
print(f" Python: {sys.version.split()[0]}")
print(f" Docker: {'是' if env['docker'] else '否'}")
print(f" SSH: {'是' if env['ssh'] else '否'}")
print(f" 显示器: {'是' if env['has_display'] else '否'}")
print(f" 服务器环境: {'是' if env['server'] else '否'}")
# 检查依赖
deps = check_dependencies()
print(f"\n📦 依赖检查:")
print(f" FastMCP: {'✅' if deps['fastmcp'] else '❌'}")
print(f" Pillow: {'✅' if deps['pillow'] else '❌'}")
print(f" Flask: {'✅' if deps['flask'] else '❌'}")
print(f" Tkinter: {'✅' if deps['tkinter'] else '❌'}")
# 决定启动模式
if args.mode == 'auto':
if env['recommend_web']:
mode = 'web'
reason = '检测到服务器环境'
elif deps['tkinter'] and env['has_display']:
mode = 'gui'
reason = '检测到桌面环境'
else:
mode = 'web'
reason = 'GUI不可用，回退到Web模式'
else:
mode = args.mode
reason = '用户指定'
print(f"\n🎯 启动模式: {mode.upper()} ({reason})")
# 安装依赖
if args.install_deps or not deps['fastmcp'] or not deps['pillow']:
if not install_dependencies(mode):
return 1
# 检查模式特定依赖
if mode == 'web' and not deps['flask']:
print("❌ Web模式需要Flask")
print("💡 请运行: pip install flask werkzeug")
print(" 或使用: --install-deps 参数")
return 1
if mode == 'gui' and not deps['tkinter']:
print("❌ GUI模式需要Tkinter")
if env['linux']:
print("💡 请安装: sudo apt-get install python3-tk")
mode = 'web' # 回退到Web模式
print("🔄 回退到Web模式")
# 准备启动参数
if mode == 'web':
# 检查端口
if not check_port(args.port):
print(f"⚠️ 端口 {args.port} 被占用")
new_port = find_available_port(args.port + 1)
print(f"🔄 使用端口 {new_port}")
args.port = new_port
print(f"\n🌐 启动Web服务器...")
print(f"📱 本地访问: http://localhost:{args.port}")
print(f"🔗 远程访问: http://服务器IP:{args.port}")
if env['ssh']:
print(f"💡 SSH端口转发: ssh -L {args.port}:localhost:{args.port} user@server")
# 构建命令
cmd = [
sys.executable, '-m', 'mcp_feedforward.server',
'--web', '--host', args.host, '--port', str(args.port)
]
else:
print(f"\n🖥️ 启动GUI模式...")
cmd = [sys.executable, '-m', 'mcp_feedforward.server']
print("=" * 60)
# 设置工作目录
if os.path.exists('src/mcp_feedforward'):
# 开发模式
os.environ['PYTHONPATH'] = os.pathsep.join([
os.path.join(os.getcwd(), 'src'),
os.environ.get('PYTHONPATH', '')
])
# 启动服务
try:
print("⚡ 正在启动...")
result = subprocess.run(cmd)
return result.returncode
except KeyboardInterrupt:
print("\n🛑 用户停止服务")
return 0
except FileNotFoundError:
print("❌ 找不到 mcp_feedforward 模块")
print("💡 请确保项目已正确安装:")
print(" pip install -e .")
print(" 或")
print(" pip install mcp-feedforward")
return 1
except Exception as e:
print(f"❌ 启动失败: {e}")
return 1
if __name__ == "__main__":
sys.exit(main())