list_docker_containers
Retrieve and display Docker container details on remote servers using hostname and credentials. Options include showing all containers and setting timeout limits for inspection.
Instructions
列出Docker容器及其信息
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| hostname | Yes | ||
| password | No | ||
| port | No | ||
| show_all | 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"
},
"show_all": {
"default": false,
"title": "Show All",
"type": "boolean"
},
"timeout": {
"default": 30,
"title": "Timeout",
"type": "integer"
},
"username": {
"title": "Username",
"type": "string"
}
},
"required": [
"hostname",
"username"
],
"title": "list_docker_containersArguments",
"type": "object"
}
Implementation Reference
- The primary handler function for the 'list_docker_containers' tool. It connects via SSH, checks for Docker, runs 'docker ps' (with -a if show_all), parses output using ServerInspector, and returns structured results.@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()
- Alternative handler implementation in the non-SSE version, with manual parsing and docker stats integration for CPU/memory usage.@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_sse/server.py:219-232 (registration)Tool registration and dispatching logic in the MCP server handler, mapping tool name to function call.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) )
- Tool schema definition used for listing available tools and generating input schemas for MCP.{"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/tools/__init__.py:24-31 (registration)Import and export of the list_docker_containers function in the tools package init file, making it available for server registration.from .docker_tools import ( list_docker_containers, list_docker_images, list_docker_volumes, get_container_logs, monitor_container_stats, check_docker_health )