backup_critical_files
Backup critical system configuration files from remote servers via SSH to protect against data loss or corruption.
Instructions
备份重要系统配置文件
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| hostname | Yes | ||
| username | Yes | ||
| password | No | ||
| port | No | ||
| files | No | ||
| backup_dir | No | /tmp/backup | |
| timeout | No |
Implementation Reference
- Primary handler implementation for the backup_critical_files tool. Connects to remote server via SSH, creates a timestamped backup directory, and copies specified critical files (default: /etc/passwd, /etc/shadow, etc.) to backups with .bak extension.
def backup_critical_files( hostname: str, username: str, password: str = "", port: int = 22, files: list[str] = ["/etc/passwd", "/etc/shadow", "/etc/fstab", "/etc/hosts"], backup_dir: str = "/tmp/backup", timeout: int = 60 ) -> dict: """备份重要系统配置文件""" result = {"status": "unknown", "backups": [], "error": ""} try: with SSHManager(hostname, username, password, port, timeout) as ssh: # 创建备份目录 mkdir_command = f"mkdir -p {backup_dir}" stdin, stdout, stderr = ssh.exec_command(mkdir_command, timeout=timeout) # 获取当前时间作为备份标识 date_command = "date +%Y%m%d_%H%M%S" stdin, stdout, stderr = ssh.exec_command(date_command, timeout=timeout) date_string = stdout.read().decode().strip() backups = [] for file_path in files: # 提取文件名 file_name = file_path.split("/")[-1] backup_path = f"{backup_dir}/{file_name}.{date_string}.bak" # 检查文件是否存在 check_command = f"[ -f {file_path} ] && echo 'exists' || echo 'not found'" stdin, stdout, stderr = ssh.exec_command(check_command, timeout=timeout) file_exists = stdout.read().decode().strip() == "exists" if file_exists: # 复制文件 copy_command = f"cp {file_path} {backup_path}" stdin, stdout, stderr = ssh.exec_command(copy_command, timeout=timeout) # 检查备份是否成功 check_backup = f"[ -f {backup_path} ] && echo 'success' || echo 'failed'" stdin, stdout, stderr = ssh.exec_command(check_backup, timeout=timeout) backup_status = stdout.read().decode().strip() == "success" backups.append({ "original_file": file_path, "backup_file": backup_path, "status": "success" if backup_status else "failed" }) else: backups.append({ "original_file": file_path, "backup_file": "", "status": "file not found" }) result["backups"] = backups result["status"] = "success" except Exception as e: result["status"] = "error" result["error"] = str(e) return result - SSE variant handler for backup_critical_files. Similar to primary but uses files_to_backup param and simpler .bak naming without timestamp.
def backup_critical_files( hostname: str, username: str, password: str = "", port: int = 22, backup_dir: str = "/tmp/backup", files_to_backup: list[str] = ["/etc/passwd", "/etc/shadow", "/etc/ssh/sshd_config"], timeout: int = 30 ) -> dict: """备份关键系统文件""" result = {"status": "unknown", "backup_results": [], "error": ""} try: with SSHManager(hostname, username, password, port, timeout) as ssh: # 创建备份目录 stdin, stdout, stderr = ssh.exec_command(f"mkdir -p {backup_dir}", timeout=timeout) error = stderr.read().decode().strip() if error: result["status"] = "error" result["error"] = f"创建备份目录失败: {error}" return result backup_results = [] for file_path in files_to_backup: # 检查文件是否存在 stdin, stdout, stderr = ssh.exec_command(f"ls {file_path}", timeout=timeout) error = stderr.read().decode().strip() if error: backup_results.append({ "file": file_path, "status": "error", "message": f"文件不存在: {error}" }) continue # 获取文件名 file_name = file_path.split('/')[-1] backup_path = f"{backup_dir}/{file_name}.bak" # 备份文件 stdin, stdout, stderr = ssh.exec_command(f"cp {file_path} {backup_path}", timeout=timeout) error = stderr.read().decode().strip() if error: backup_results.append({ "file": file_path, "status": "error", "message": f"备份失败: {error}" }) else: backup_results.append({ "file": file_path, "status": "success", "backup_path": backup_path }) result["backup_results"] = backup_results result["status"] = "success" except Exception as e: result["status"] = "error" result["error"] = str(e) return result - server_monitor/tools/utils.py:95-103 (schema)Tool schema definition including parameters and defaults for backup_critical_files.
{"name": "backup_critical_files", "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": "files", "type": "list[str]", "default": ["/etc/passwd", "/etc/shadow", "/etc/fstab", "/etc/hosts"]}, {"name": "backup_dir", "type": "str", "default": "/tmp/backup"}, {"name": "timeout", "type": "int", "default": 60} ]}, - server_monitor/main.py:52-52 (registration)Registration of backup_critical_files in the MCP tools dictionary for dynamic registration using mcp.tool() decorator.
'backup_critical_files': backup_critical_files, - server_monitor_sse/server.py:172-186 (registration)Tool dispatch registration in SSE server: handles call to backup_critical_files by parsing arguments and invoking the handler.
elif name == "backup_critical_files": required_args = ["hostname", "username"] for arg in required_args: if arg not in arguments: raise ValueError(f"Missing required argument '{arg}'") result = backup_critical_files( hostname=arguments["hostname"], username=arguments["username"], password=arguments.get("password", ""), port=arguments.get("port", 22), backup_dir=arguments.get("backup_dir", "/tmp/backup"), files_to_backup=arguments.get("files_to_backup", ["/etc/passwd", "/etc/shadow", "/etc/ssh/sshd_config"]), timeout=arguments.get("timeout", 30) )