OPS MCP Server

Instructions

Implementation Reference

• server_monitor_sse/tools/docker_tools.py:10-61 (handler)

  Primary handler function for the list_docker_containers MCP tool. Executes 'docker ps' via SSH, checks Docker installation, parses output using ServerInspector.parse_docker_containers, and returns structured InspectionResult.
  @handle_exceptions def list_docker_containers( hostname: str, username: str, password: str = "", port: int = 22, show_all: bool = False, # 是否显示所有容器，包括已停止的 timeout: int = 30 ) -> dict: """列出Docker容器及其信息""" result = InspectionResult() try: with SSHManager(hostname, username, password, port, timeout) as ssh: # 检查Docker是否安装 stdin, stdout, stderr = ssh.exec_command("command -v docker", timeout=timeout) if not stdout.read().strip(): result.status = "error" result.error = "Docker未安装在目标服务器上" return result.dict() # 构建命令 cmd = "docker ps" if show_all: cmd += " -a" # 执行命令 stdin, stdout, stderr = ssh.exec_command(cmd, timeout=timeout) container_output = stdout.read().decode('utf-8') error_output = stderr.read().decode('utf-8') if error_output: result.status = "error" result.error = f"获取容器列表失败: {error_output}" return result.dict() # 解析容器信息 containers = ServerInspector.parse_docker_containers(container_output) # 设置结果 result.status = "success" result.data = {"containers": containers} result.raw_outputs = {"container_list": container_output} container_count = len(containers) result.summary = f"找到 {container_count} 个{'所有' if show_all else '运行中的'}容器" except Exception as e: result.status = "error" result.error = f"获取容器列表失败: {str(e)}" return result.dict()
• server_monitor_sse/tools/utils.py:101-108 (schema)

  Tool schema definition used by list_available_tools() for MCP tool discovery, specifying parameters and description for list_docker_containers.
  {"name": "list_docker_containers", "description": "列出Docker容器及其信息", "parameters": [ {"name": "hostname", "type": "str", "default": None}, {"name": "username", "type": "str", "default": None}, {"name": "password", "type": "str", "default": ""}, {"name": "port", "type": "int", "default": 22}, {"name": "show_all", "type": "bool", "default": False}, {"name": "timeout", "type": "int", "default": 30} ]},
• server_monitor_sse/server.py:219-232 (registration)

  MCP @app.call_tool() dispatcher that handles tool calls for 'list_docker_containers' by invoking the handler with arguments.
  elif name == "list_docker_containers": required_args = ["hostname", "username"] for arg in required_args: if arg not in arguments: raise ValueError(f"Missing required argument '{arg}'") result = list_docker_containers( hostname=arguments["hostname"], username=arguments["username"], password=arguments.get("password", ""), port=arguments.get("port", 22), show_all=arguments.get("show_all", False), timeout=arguments.get("timeout", 30) )
• server_monitor/tools/docker_tools.py:8-93 (handler)

  Alternative/detailed handler implementation for list_docker_containers with custom Docker command formatting for container list and stats integration.
  @handle_exceptions def list_docker_containers( hostname: str, username: str, password: str = "", port: int = 22, show_all: bool = False, # 是否显示所有容器，包括已停止的 timeout: int = 30 ) -> dict: """列出Docker容器及其信息""" result = InspectionResult() try: with SSHManager(hostname, username, password, port, timeout) as ssh: # 检查Docker是否安装 stdin, stdout, stderr = ssh.exec_command("command -v docker") if not stdout.read().strip(): result.status = "error" result.error = "Docker未安装在目标服务器上" return result.dict() # 列出容器 cmd = "docker ps --format '{{.ID}}|{{.Names}}|{{.Image}}|{{.Status}}|{{.CreatedAt}}|{{.Ports}}'" if show_all: cmd += " -a" stdin, stdout, stderr = ssh.exec_command(cmd) container_output = stdout.read().decode('utf-8') # 获取容器资源使用情况 stdin, stdout, stderr = ssh.exec_command("docker stats --no-stream --format '{{.ID}}|{{.CPUPerc}}|{{.MemPerc}}'") stats_output = stdout.read().decode('utf-8') # 处理结果 containers = [] stats_map = {} # 解析资源使用情况 for line in stats_output.strip().split('\n'): if line: parts = line.split('|') if len(parts) >= 3: container_id = parts[0] cpu_perc = parts[1].replace('%', '') if parts[1] else "0" mem_perc = parts[2].replace('%', '') if parts[2] else "0" try: stats_map[container_id] = { 'cpu_usage': float(cpu_perc), 'memory_usage': float(mem_perc) } except ValueError: stats_map[container_id] = { 'cpu_usage': 0.0, 'memory_usage': 0.0 } # 解析容器列表 for line in container_output.strip().split('\n'): if line: parts = line.split('|') if len(parts) >= 6: container_id = parts[0] container_info = ContainerInfo( container_id=container_id, name=parts[1], image=parts[2], status=parts[3], created=parts[4], ports=parts[5], cpu_usage=stats_map.get(container_id, {}).get('cpu_usage'), memory_usage=stats_map.get(container_id, {}).get('memory_usage') ) containers.append(container_info) # 设置结果 result.status = "success" result.data = {"containers": containers} result.raw_outputs = {"container_list": container_output, "stats": stats_output} result.summary = f"发现 {len(containers)} 个容器" except Exception as e: result.status = "error" result.error = f"获取Docker容器信息失败: {str(e)}" return result.dict()
• server_monitor/main.py:42-61 (registration)

  FastMCP tool registration mapping 'list_docker_containers' to its handler function.
  tools_dict = { 'get_memory_info': get_memory_info, 'remote_server_inspection': remote_server_inspection, 'get_system_load': get_system_load, 'monitor_processes': monitor_processes, 'check_service_status': check_service_status, 'get_os_details': get_os_details, 'check_ssh_risk_logins': check_ssh_risk_logins, 'check_firewall_config': check_firewall_config, 'security_vulnerability_scan': security_vulnerability_scan, 'backup_critical_files': backup_critical_files, 'inspect_network': inspect_network, 'analyze_logs': analyze_logs, 'list_docker_containers': list_docker_containers, 'list_docker_images': list_docker_images, 'list_docker_volumes': list_docker_volumes, 'get_container_logs': get_container_logs, 'monitor_container_stats': monitor_container_stats, 'check_docker_health': check_docker_health }