check_service_status
Monitor and verify the operational status of services on remote servers through SSH connections to identify running or stopped processes.
Instructions
检查指定服务的运行状态
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| hostname | Yes | ||
| username | Yes | ||
| password | No | ||
| port | No | ||
| services | No | ||
| timeout | No |
Implementation Reference
- Core handler function implementing check_service_status: connects via SSH, checks systemctl status for services or lists running ones.
@handle_exceptions def check_service_status( hostname: str, username: str, password: str = "", port: int = 22, services: list[str] = [], timeout: int = 30 ) -> dict: """检查指定服务的运行状态""" result = {"status": "unknown", "services": [], "error": ""} try: with SSHManager(hostname, username, password, port, timeout) as ssh: if services: # 检查特定服务 service_statuses = [] for service in services: command = f"systemctl status {service}" stdin, stdout, stderr = ssh.exec_command(command, timeout=timeout) output = stdout.read().decode().strip() # 分析输出判断服务状态 service_status = { "name": service, "status": "unknown", "active": False, "enabled": False } if "Active: active" in output: service_status["status"] = "running" service_status["active"] = True elif "Active: inactive" in output: service_status["status"] = "stopped" elif "not-found" in output or "could not be found" in output: service_status["status"] = "not found" # 检查是否开机启动 enabled_command = f"systemctl is-enabled {service}" stdin, stdout, stderr = ssh.exec_command(enabled_command, timeout=timeout) enabled_output = stdout.read().decode().strip() service_status["enabled"] = enabled_output == "enabled" service_statuses.append(service_status) result["services"] = service_statuses else: # 列出所有活跃的服务 command = "systemctl list-units --type=service --state=running" stdin, stdout, stderr = ssh.exec_command(command, timeout=timeout) raw_output = stdout.read().decode().strip() # 解析服务状态 result["services"] = ServerInspector.parse_services(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)Registration of the check_service_status tool in the MCP server, imported and added to tools_dict then decorated.
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/tools/utils.py:70-77 (schema)Tool schema definition including name, description, and parameters for check_service_status.
{"name": "check_service_status", "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": "services", "type": "list[str]", "default": []}, {"name": "timeout", "type": "int", "default": 30} ]}, - SSE variant handler for check_service_status: similar SSH-based service status check with slight differences in param and logic.
@handle_exceptions def check_service_status( hostname: str, username: str, password: str = "", port: int = 22, service_names: list[str] = [], # 为空则检查所有服务 timeout: int = 30 ) -> dict: """检查服务状态""" result = {"status": "unknown", "services": [], "error": ""} try: with SSHManager(hostname, username, password, port, timeout) as ssh: # 构建命令 if service_names: # 检查指定服务 services_str = " ".join(service_names) command = f"systemctl status {services_str}" else: # 列出所有服务 command = "systemctl list-units --type=service --all" stdin, stdout, stderr = ssh.exec_command(command, timeout=timeout) raw_output = stdout.read().decode().strip() # 解析服务状态 result["services"] = ServerInspector.parse_services(raw_output) result["status"] = "success" except Exception as e: result["status"] = "error" result["error"] = str(e) return result - server_monitor_sse/server.py:98-111 (registration)Manual dispatching/registration in SSE server's tool_handler for check_service_status call.
elif name == "check_service_status": required_args = ["hostname", "username"] for arg in required_args: if arg not in arguments: raise ValueError(f"Missing required argument '{arg}'") result = check_service_status( hostname=arguments["hostname"], username=arguments["username"], password=arguments.get("password", ""), port=arguments.get("port", 22), service_names=arguments.get("service_names", []), timeout=arguments.get("timeout", 30) )