OPS MCP Server

Instructions

Implementation Reference

• server_monitor/tools/system_tools.py:121-138 (handler)

  Core handler implementation for the get_system_load tool. Uses SSH to execute 'uptime' command and parse load average.

  @handle_exceptions
  def get_system_load(
      hostname: str,
      username: str,
      password: str = "",
      port: int = 22,
      timeout: int = 30
  ) -> dict:
      """获取系统负载信息"""
      try:
          with SSHManager(hostname, username, password, port, timeout) as ssh:
              stdin, stdout, stderr = ssh.exec_command("uptime")
              load_output = stdout.read().decode().strip()
              load_avg = re.search(r'load average: (.*)', load_output)
              return {"status": "success", "load_average": load_avg.group(1) if load_avg else "unknown"}
      except Exception as e:
          return {"status": "error", "error": str(e)}

• server_monitor_sse/tools/system_tools.py:87-104 (handler)

  Core handler implementation for the get_system_load tool in SSE variant. Identical logic using SSH 'uptime'.

  @handle_exceptions
  def get_system_load(
      hostname: str,
      username: str,
      password: str = "",
      port: int = 22,
      timeout: int = 30
  ) -> dict:
      """获取系统负载信息"""
      try:
          with SSHManager(hostname, username, password, port, timeout) as ssh:
              stdin, stdout, stderr = ssh.exec_command("uptime")
              load_output = stdout.read().decode().strip()
              load_avg = re.search(r'load average: (.*)', load_output)
              return {"status": "success", "load_average": load_avg.group(1) if load_avg else "unknown"}
      except Exception as e:
          return {"status": "error", "error": str(e)}

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

  Tool registration in FastMCP server, including get_system_load in tools_dict and dynamic @mcp.tool() decoration.

  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:68-81 (registration)

  Manual dispatcher/registration case for get_system_load tool call in SSE server handler.

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

• server_monitor/tools/utils.py:53-59 (schema)

  JSON schema definition for get_system_load tool parameters and description used in list_available_tools.

  {"name": "get_system_load", "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": "timeout", "type": "int", "default": 30}
  ]},