monitor_processes
Monitor remote server processes to identify resource-intensive applications by analyzing CPU and memory usage for system optimization.
Instructions
监控远程服务器进程，返回占用资源最多的进程
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| hostname | Yes | ||
| username | Yes | ||
| password | No | ||
| port | No | ||
| top_n | No | ||
| sort_by | No | cpu | |
| timeout | No |
Implementation Reference
- 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
- 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} ]},