list_docker_volumes
Retrieve and list Docker volumes on a remote server by specifying hostname, username, and optional parameters for port, password, and timeout.
Instructions
列出Docker卷
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| hostname | Yes | ||
| password | No | ||
| port | No | ||
| timeout | No | ||
| username | Yes |
Input Schema (JSON Schema)
{
"properties": {
"hostname": {
"title": "Hostname",
"type": "string"
},
"password": {
"default": "",
"title": "Password",
"type": "string"
},
"port": {
"default": 22,
"title": "Port",
"type": "integer"
},
"timeout": {
"default": 30,
"title": "Timeout",
"type": "integer"
},
"username": {
"title": "Username",
"type": "string"
}
},
"required": [
"hostname",
"username"
],
"title": "list_docker_volumesArguments",
"type": "object"
}
Implementation Reference
- Primary handler implementation for list_docker_volumes tool. Executes SSH commands to list Docker volumes, parses output, estimates size using du, and fetches creation times via docker volume inspect.@handle_exceptions def list_docker_volumes( 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 volume ls --format '{{.Name}}|{{.Driver}}|{{.Mountpoint}}'" stdin, stdout, stderr = ssh.exec_command(cmd) volume_output = stdout.read().decode('utf-8') # 处理结果 volumes = [] # 解析卷列表 for line in volume_output.strip().split('\n'): if line: parts = line.split('|') if len(parts) >= 3: # 尝试获取卷大小(非标准功能,可能需要自定义脚本) size = None try: stdin, stdout, stderr = ssh.exec_command(f"sudo du -sh {parts[2]}") size_output = stdout.read().decode('utf-8').strip() if size_output: size = size_output.split()[0] except: pass volume_info = VolumeInfo( name=parts[0], driver=parts[1], mountpoint=parts[2], created="N/A", # Docker命令不直接提供创建时间 size=size ) volumes.append(volume_info) # 获取更详细的卷信息(包括创建时间) for volume in volumes: try: stdin, stdout, stderr = ssh.exec_command(f"docker volume inspect {volume['name']}") inspect_output = stdout.read().decode('utf-8') if "CreatedAt" in inspect_output: inspect_data = json.loads(inspect_output) if inspect_data and len(inspect_data) > 0 and "CreatedAt" in inspect_data[0]: volume["created"] = inspect_data[0]["CreatedAt"] except: pass # 设置结果 result.status = "success" result.data = {"volumes": volumes} result.raw_outputs = {"volume_list": volume_output} result.summary = f"发现 {len(volumes)} 个Docker卷" except Exception as e: result.status = "error" result.error = f"获取Docker卷信息失败: {str(e)}" return result.dict()
- Alternative handler for list_docker_volumes in SSE version. Runs docker volume ls and delegates parsing to ServerInspector.parse_docker_volumes.@handle_exceptions def list_docker_volumes( 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 volume ls", timeout=timeout) volumes_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() # 解析卷信息 volumes = ServerInspector.parse_docker_volumes(volumes_output) # 设置结果 result.status = "success" result.data = {"volumes": volumes} result.raw_outputs = {"volume_list": volumes_output} volume_count = len(volumes) result.summary = f"找到 {volume_count} 个Docker卷" except Exception as e: result.status = "error" result.error = f"获取卷列表失败: {str(e)}" return result.dict()
- server_monitor_sse/server.py:248-260 (registration)Tool registration and dispatch logic in the SSE server. Maps tool name 'list_docker_volumes' to the handler function call within the MCP tool_handler.elif name == "list_docker_volumes": required_args = ["hostname", "username"] for arg in required_args: if arg not in arguments: raise ValueError(f"Missing required argument '{arg}'") result = list_docker_volumes( hostname=arguments["hostname"], username=arguments["username"], password=arguments.get("password", ""), port=arguments.get("port", 22), timeout=arguments.get("timeout", 30) )
- server_monitor/tools/__init__.py:25-28 (registration)Export/registration of docker tools including list_docker_volumes in the tools package __init__ for server_monitor.from .docker_tools import ( list_docker_containers, list_docker_images, list_docker_volumes,
- server_monitor_sse/server.py:8-27 (registration)Import of the list_docker_volumes handler into the SSE server main file.from tools import ( remote_server_inspection, get_system_load, monitor_processes, check_service_status, get_os_details, check_ssh_risk_logins, check_firewall_config, security_vulnerability_scan, backup_critical_files, inspect_network, analyze_logs, list_docker_containers, list_docker_images, list_docker_volumes, get_container_logs, monitor_container_stats, check_docker_health, list_available_tools )