#!/usr/bin/env python3 import json import sys import os import subprocess import logging from datetime import datetime # 设置调试日志 logging.basicConfig( level=logging.DEBUG, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[ logging.FileHandler('/tmp/mcp_server_debug.log'), logging.StreamHandler(sys.stderr) ] ) logger = logging.getLogger(__name__) # 设置环境变量确保安静模式 os.environ['MCP_QUIET'] = '1' # 添加python目录到路径 sys.path.insert(0, os.path.join(os.path.dirname(__file__))) logger.info("=== MCP Server Debug Mode Started ===") logger.info(f"Python path: {sys.path}") logger.info(f"Working directory: {os.getcwd()}") logger.info(f"Script path: {__file__}") try: logger.info("尝试导入SSH Manager...") from enhanced_ssh_manager import EnhancedSSHManager logger.info("SSH Manager导入成功") except ImportError as e: logger.error(f"无法导入SSH Manager: {e}") error_response = {"error": f"无法导入SSH Manager: {e}"} print(json.dumps(error_response, ensure_ascii=False), flush=True) sys.exit(1) class MCPServerDebug: def __init__(self): logger.info("初始化MCP服务器...") try: self.ssh_manager = EnhancedSSHManager() logger.info("SSH Manager初始化成功") except Exception as e: logger.error(f"初始化SSH Manager失败: {e}") error_response = {"error": f"初始化SSH Manager失败: {e}"} print(json.dumps(error_response, ensure_ascii=False), flush=True) sys.exit(1) def log_request(self, request): """记录请求详情""" logger.info(f"收到请求: {request.get('method', 'unknown')} (ID: {request.get('id', 'none')})") logger.debug(f"完整请求: {json.dumps(request, ensure_ascii=False)}") def log_response(self, response): """记录响应详情""" logger.info(f"发送响应: ID {response.get('id', 'none')}") if 'error' in response: logger.error(f"响应包含错误: {response['error']}") else: logger.debug(f"完整响应: {json.dumps(response, ensure_ascii=False)}") def handle_request(self, request): self.log_request(request) try: method = request.get('method') request_id = request.get('id') if method == 'initialize': logger.info("处理initialize请求") # 使用客户端请求的协议版本 client_protocol_version = request.get('params', {}).get('protocolVersion', '2024-11-05') response = { "jsonrpc": "2.0", "id": request_id, "result": { "protocolVersion": client_protocol_version, # 使用客户端版本 "capabilities": { "tools": {} }, "serverInfo": { "name": "remote-terminal-mcp", "version": "1.0.0-debug" } } } logger.info(f"initialize响应已准备，协议版本: {client_protocol_version}") elif method == 'notifications/initialized': logger.info("处理notifications/initialized通知") # 这是一个通知，不需要响应，只记录日志 return None # 不返回响应 elif method == 'tools/list': logger.info("处理tools/list请求") response = { "jsonrpc": "2.0", "id": request_id, "result": { "tools": [ { "name": "list_servers", "description": "列出所有可用的服务器", "inputSchema": { "type": "object", "properties": {}, "additionalProperties": False } }, { "name": "connect_server", "description": "连接到指定服务器", "inputSchema": { "type": "object", "properties": { "server_name": { "type": "string", "description": "服务器名称" } }, "required": ["server_name"], "additionalProperties": False } }, { "name": "execute_command", "description": "在指定服务器上执行命令", "inputSchema": { "type": "object", "properties": { "server_name": { "type": "string", "description": "服务器名称" }, "command": { "type": "string", "description": "要执行的命令" } }, "required": ["server_name", "command"], "additionalProperties": False } }, { "name": "get_server_status", "description": "获取服务器状态", "inputSchema": { "type": "object", "properties": { "server_name": { "type": "string", "description": "服务器名称" } }, "required": ["server_name"], "additionalProperties": False } }, { "name": "run_local_command", "description": "在本地执行命令", "inputSchema": { "type": "object", "properties": { "command": { "type": "string", "description": "要执行的本地命令" } }, "required": ["command"], "additionalProperties": False } } ] } } logger.info(f"tools/list响应已准备，包含{len(response['result']['tools'])}个工具") elif method == 'tools/call': params = request.get('params', {}) tool_name = params.get('name') arguments = params.get('arguments', {}) logger.info(f"处理tools/call请求: {tool_name}") logger.debug(f"工具参数: {arguments}") if tool_name == 'list_servers': logger.info("执行list_servers工具") try: servers = self.ssh_manager.list_servers() logger.info(f"获取到{len(servers)}个服务器") # 只返回服务器名称和基本信息，避免大数据量 simple_servers = [] for server in servers: simple_servers.append({ 'name': server.get('name', ''), 'description': server.get('description', ''), 'connected': server.get('connected', False), 'type': server.get('type', '') }) response = { "jsonrpc": "2.0", "id": request_id, "result": { "content": [ { "type": "text", "text": json.dumps(simple_servers, ensure_ascii=False, indent=2) } ] } } logger.info("list_servers工具执行成功") except Exception as e: logger.error(f"list_servers工具执行失败: {e}") response = { "jsonrpc": "2.0", "id": request_id, "error": {"code": -1, "message": f"list_servers失败: {e}"} } elif tool_name == 'connect_server': server_name = arguments.get('server_name') logger.info(f"执行connect_server工具: {server_name}") if not server_name: response = { "jsonrpc": "2.0", "id": request_id, "error": {"code": -1, "message": "缺少server_name参数"} } else: try: success, message = self.ssh_manager.smart_connect(server_name) result = f"成功: {success}, 消息: {message}" response = { "jsonrpc": "2.0", "id": request_id, "result": { "content": [ { "type": "text", "text": f"连接结果: {result}" } ] } } logger.info(f"connect_server工具执行成功: {server_name}") except Exception as e: logger.error(f"connect_server工具执行失败: {e}") response = { "jsonrpc": "2.0", "id": request_id, "error": {"code": -1, "message": f"连接失败: {e}"} } elif tool_name == 'execute_command': server_name = arguments.get('server_name') command = arguments.get('command') logger.info(f"执行execute_command工具: {server_name} -> {command}") if not server_name or not command: response = { "jsonrpc": "2.0", "id": request_id, "error": {"code": -1, "message": "缺少server_name或command参数"} } else: try: result = self.ssh_manager.execute_command(server_name, command) response = { "jsonrpc": "2.0", "id": request_id, "result": { "content": [ { "type": "text", "text": f"命令执行结果:\n{result}" } ] } } logger.info(f"execute_command工具执行成功") except Exception as e: logger.error(f"execute_command工具执行失败: {e}") response = { "jsonrpc": "2.0", "id": request_id, "error": {"code": -1, "message": f"命令执行失败: {e}"} } elif tool_name == 'get_server_status': server_name = arguments.get('server_name') logger.info(f"执行get_server_status工具: {server_name}") if not server_name: response = { "jsonrpc": "2.0", "id": request_id, "error": {"code": -1, "message": "缺少server_name参数"} } else: try: status = self.ssh_manager.get_connection_status(server_name) response = { "jsonrpc": "2.0", "id": request_id, "result": { "content": [ { "type": "text", "text": json.dumps(status, ensure_ascii=False, indent=2) } ] } } logger.info(f"get_server_status工具执行成功") except Exception as e: logger.error(f"get_server_status工具执行失败: {e}") response = { "jsonrpc": "2.0", "id": request_id, "error": {"code": -1, "message": f"获取状态失败: {e}"} } elif tool_name == 'run_local_command': command = arguments.get('command') logger.info(f"执行run_local_command工具: {command}") if not command: response = { "jsonrpc": "2.0", "id": request_id, "error": {"code": -1, "message": "缺少command参数"} } else: try: result = subprocess.run(command, shell=True, capture_output=True, text=True) output = f"退出代码: {result.returncode}\n标准输出:\n{result.stdout}\n标准错误:\n{result.stderr}" response = { "jsonrpc": "2.0", "id": request_id, "result": { "content": [ { "type": "text", "text": output } ] } } logger.info(f"run_local_command工具执行成功") except Exception as e: logger.error(f"run_local_command工具执行失败: {e}") response = { "jsonrpc": "2.0", "id": request_id, "error": {"code": -1, "message": f"本地命令执行失败: {e}"} } else: logger.warning(f"未知工具: {tool_name}") response = { "jsonrpc": "2.0", "id": request_id, "error": {"code": -1, "message": f"未知工具: {tool_name}"} } else: logger.warning(f"未知方法: {method}") response = { "jsonrpc": "2.0", "id": request_id, "error": {"code": -1, "message": f"未知方法: {method}"} } self.log_response(response) return response except Exception as e: logger.error(f"处理请求时出错: {e}", exc_info=True) response = { "jsonrpc": "2.0", "id": request.get('id', None), "error": {"code": -1, "message": f"处理请求时出错: {e}"} } self.log_response(response) return response def main(): logger.info("启动MCP服务器主循环...") server = MCPServerDebug() try: logger.info("开始监听stdin...") for line_number, line in enumerate(sys.stdin, 1): logger.debug(f"收到第{line_number}行输入: {repr(line)}") line = line.strip() if not line: logger.debug("跳过空行") continue try: request = json.loads(line) logger.debug(f"JSON解析成功: {request}") response = server.handle_request(request) if response is not None: # 只有非通知才发送响应 response_json = json.dumps(response, ensure_ascii=False) print(response_json, flush=True) logger.debug(f"响应已发送: {len(response_json)} 字符") else: logger.debug("通知处理完成，无需响应") except json.JSONDecodeError as e: logger.error(f"JSON解析错误 (第{line_number}行): {e}") logger.error(f"原始数据: {repr(line)}") error_response = { "jsonrpc": "2.0", "id": None, "error": {"code": -1, "message": f"JSON解析错误: {e}"} } print(json.dumps(error_response, ensure_ascii=False), flush=True) except KeyboardInterrupt: logger.info("收到中断信号，正常退出") except Exception as e: logger.error(f"主循环异常: {e}", exc_info=True) error_response = { "jsonrpc": "2.0", "id": None, "error": {"code": -1, "message": f"服务器错误: {e}"} } print(json.dumps(error_response, ensure_ascii=False), flush=True) finally: logger.info("=== MCP Server Debug Mode Ended ===") if __name__ == "__main__": main()