list_docker_images
Retrieve a list of Docker images from remote servers to monitor container deployments and manage system resources.
Instructions
列出Docker镜像
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| hostname | Yes | ||
| username | Yes | ||
| password | No | ||
| port | No | ||
| timeout | No |
Implementation Reference
- Handler function implementing list_docker_images tool in server_monitor. Executes 'docker images' with custom format via SSH, parses output into ImageInfo objects, and returns structured InspectionResult.
@handle_exceptions def list_docker_images( hostname: str, username: str, password: str = "", port: int = 22, 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 images --format '{{.ID}}|{{.Repository}}|{{.Tag}}|{{.CreatedAt}}|{{.Size}}'" stdin, stdout, stderr = ssh.exec_command(cmd) image_output = stdout.read().decode('utf-8') # 处理结果 images = [] # 解析镜像列表 for line in image_output.strip().split('\n'): if line: parts = line.split('|') if len(parts) >= 5: image_info = ImageInfo( image_id=parts[0], repository=parts[1], tag=parts[2], created=parts[3], size=parts[4] ) images.append(image_info) # 设置结果 result.status = "success" result.data = {"images": images} result.raw_outputs = {"image_list": image_output} result.summary = f"发现 {len(images)} 个Docker镜像" except Exception as e: result.status = "error" result.error = f"获取Docker镜像信息失败: {str(e)}" return result.dict() - Handler function implementing list_docker_images tool in server_monitor_sse. Runs 'docker images' via SSH and delegates parsing to ServerInspector.parse_docker_images, returns InspectionResult.
@handle_exceptions def list_docker_images( hostname: str, username: str, password: str = "", port: int = 22, 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() # 执行命令 stdin, stdout, stderr = ssh.exec_command("docker images", timeout=timeout) images_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() # 解析镜像信息 images = ServerInspector.parse_docker_images(images_output) # 设置结果 result.status = "success" result.data = {"images": images} result.raw_outputs = {"image_list": images_output} image_count = len(images) result.summary = f"找到 {image_count} 个Docker镜像" except Exception as e: result.status = "error" result.error = f"获取镜像列表失败: {str(e)}" return result.dict() - server_monitor/main.py:42-66 (registration)Registration of list_docker_images in tools_dict and dynamic registration using mcp.tool() decorator in FastMCP server.
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 } # 使用装饰器动态注册所有工具 for name, func in tools_dict.items(): mcp.tool()(func) - server_monitor_sse/server.py:234-246 (registration)Tool dispatch registration in @app.call_tool() handler: calls list_docker_images when name matches.
elif name == "list_docker_images": required_args = ["hostname", "username"] for arg in required_args: if arg not in arguments: raise ValueError(f"Missing required argument '{arg}'") result = list_docker_images( hostname=arguments["hostname"], username=arguments["username"], password=arguments.get("password", ""), port=arguments.get("port", 22), timeout=arguments.get("timeout", 30) ) - Schema defining tool name constant DOCKER_IMAGES = "list_docker_images" in ServerTools enum.
DOCKER_CONTAINERS = "list_docker_containers" # 列出Docker容器 DOCKER_IMAGES = "list_docker_images" # 列出Docker镜像 DOCKER_VOLUMES = "list_docker_volumes" # 列出Docker卷 CONTAINER_LOGS = "get_container_logs" # 获取容器日志 CONTAINER_STATS = "monitor_container_stats" # 监控容器状态 DOCKER_HEALTHCHECK = "check_docker_health" # 检查Docker服务健康状态