OPS MCP Server

Instructions

Implementation Reference

• server_monitor_sse/tools/docker_tools.py:111-155 (handler)

  The primary handler function for the 'list_docker_volumes' tool. It connects via SSH to the target server, checks for Docker, runs 'docker volume ls', parses the output using ServerInspector.parse_docker_volumes, and returns structured results including volumes list and summary.
  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/tools/utils.py:116-122 (schema)

  The JSON schema definition for the MCP tool 'list_docker_volumes', including description and input parameters (hostname, username, password, port, timeout). Used by list_tools() to generate the tool schema.
  {"name": "list_docker_volumes", "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": "timeout", "type": "int", "default": 30} ]},
• server_monitor_sse/server.py:248-260 (registration)

  Registration and dispatching logic in the MCP app.call_tool() method. Checks required args and calls the list_docker_volumes handler with provided arguments.
  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_sse/tools/__init__.py:24-27 (registration)

  Import registration of the list_docker_volumes function from docker_tools.py into the tools package, making it available for use in the MCP server.
  from .docker_tools import ( list_docker_containers, list_docker_images, list_docker_volumes,
• server_monitor_sse/models/schemas.py:94-99 (schema)

  Pydantic/TypedDict model for VolumeInfo used in the output data structure of list_docker_volumes.
  class VolumeInfo(TypedDict): """卷信息数据结构""" volume_name: str driver: str mountpoint: str