OPS MCP Server

Instructions

Implementation Reference

• server_monitor/tools/system_tools.py:140-175 (handler)

  Core handler function for the monitor_processes MCP tool. Establishes SSH connection to remote server, runs sorted 'ps aux' command to get top processes by CPU/memory/time, parses output using ServerInspector.parse_processes, and returns status and process list.

  def monitor_processes(
      hostname: str,
      username: str,
      password: str = "",
      port: int = 22,
      top_n: int = 10,
      sort_by: str = "cpu",
      timeout: int = 30
  ) -> dict:
      """监控远程服务器进程，返回占用资源最多的进程"""
      result = {"status": "unknown", "processes": [], "error": ""}
  
      sort_options = {
          "cpu": "-pcpu",
          "memory": "-pmem",
          "time": "-time"
      }
  
      sort_param = sort_options.get(sort_by, "-pcpu")
  
      try:
          with SSHManager(hostname, username, password, port, timeout) as ssh:
              # 使用ps命令获取进程信息，并按指定条件排序
              command = f"ps aux --sort={sort_param} | head -n {top_n + 1}"  # +1 是为了包含标题行
              stdin, stdout, stderr = ssh.exec_command(command, timeout=timeout)
              raw_output = stdout.read().decode().strip()
  
              # 解析进程信息
              result["processes"] = ServerInspector.parse_processes(raw_output)
              result["status"] = "success"
  
      except Exception as e:
          result["status"] = "error"
          result["error"] = str(e)
  
      return result

• server_monitor_sse/tools/system_tools.py:106-141 (handler)

  Identical core handler function for the monitor_processes tool in the SSE variant of the server monitor.

  def monitor_processes(
      hostname: str,
      username: str,
      password: str = "",
      port: int = 22,
      top_n: int = 10,
      sort_by: str = "cpu",
      timeout: int = 30
  ) -> dict:
      """监控远程服务器进程，返回占用资源最多的进程"""
      result = {"status": "unknown", "processes": [], "error": ""}
  
      sort_options = {
          "cpu": "-pcpu",
          "memory": "-pmem",
          "time": "-time"
      }
  
      sort_param = sort_options.get(sort_by, "-pcpu")
  
      try:
          with SSHManager(hostname, username, password, port, timeout) as ssh:
              # 使用ps命令获取进程信息，并按指定条件排序
              command = f"ps aux --sort={sort_param} | head -n {top_n + 1}"  # +1 是为了包含标题行
              stdin, stdout, stderr = ssh.exec_command(command, timeout=timeout)
              raw_output = stdout.read().decode().strip()
  
              # 解析进程信息
              result["processes"] = ServerInspector.parse_processes(raw_output)
              result["status"] = "success"
  
      except Exception as e:
          result["status"] = "error"
          result["error"] = str(e)
  
      return result

• server_monitor/main.py:42-66 (registration)

  MCP tool registration in FastMCP server. The monitor_processes function is included in tools_dict (line 46) and dynamically registered using mcp.tool().

  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:82-96 (registration)

  Manual dispatch and registration of monitor_processes tool call within the @app.call_tool() handler in the SSE MCP server.

  elif name == "monitor_processes":
      required_args = ["hostname", "username"]
      for arg in required_args:
          if arg not in arguments:
              raise ValueError(f"Missing required argument '{arg}'")
  
      result = monitor_processes(
          hostname=arguments["hostname"],
          username=arguments["username"],
          password=arguments.get("password", ""),
          port=arguments.get("port", 22),
          top_n=arguments.get("top_n", 10),
          sort_by=arguments.get("sort_by", "cpu"),
          timeout=arguments.get("timeout", 30)
      )

• server_monitor/tools/utils.py:61-69 (schema)

  JSON schema definition for the monitor_processes tool, including parameters, types, and defaults, used in list_available_tools.

  {"name": "monitor_processes", "description": "监控远程服务器进程，返回占用资源最多的进程", "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": "top_n", "type": "int", "default": 10},
      {"name": "sort_by", "type": "str", "default": "cpu"},
      {"name": "timeout", "type": "int", "default": 30}
  ]},