MCP SSH Tools Server

Instructions

Implementation Reference

• main.py:370-452 (handler)

  The main handler function for the 'upload_file' MCP tool. Decorated with @mcp.tool() for registration. Handles uploading local files or directories (recursively) to a remote SSH server via SFTP. Includes input validation, connection management, and detailed success/failure reporting.
  def upload_file(server_name: str, remote_path: str, local_src: str) -> Dict[str, Any]: """ 将本地的文件或者目录上传到远程服务器 参数: - server_name: 服务器名称 - remote_path: 远程文件/目录路径 - local_src: 本地文件/目录路径 """ client = mcp_manager.get_connection(server_name) if not client: return { "success": False, "error": "SSH连接失败", "server": server_name } try: # 检查本地路径是否存在 if not os.path.exists(local_src): return { "success": False, "error": f"本地路径不存在: {local_src}", "server": server_name } sftp = client.open_sftp() if os.path.isdir(local_src): # 是目录，递归上传 result = upload_directory_recursive(sftp, local_src, remote_path) sftp.close() if result["success"]: return { "success": True, "server": server_name, "ip": mcp_manager.server_configs[server_name].ssh_ip, "type": "directory", "message": f"目录上传完成: {local_src} -> {remote_path}", "uploaded_files": result["uploaded_files"], "failed_files": result["failed_files"], "summary": { "total_uploaded": len(result["uploaded_files"]), "total_failed": len(result["failed_files"]) } } else: return { "success": False, "error": result["error"], "server": server_name } else: # 是文件，直接上传 # 创建远程目录 remote_dir = os.path.dirname(remote_path) if remote_dir: try: sftp.mkdir(remote_dir) except IOError: # 目录可能已存在，忽略错误 pass sftp.put(local_src, remote_path) sftp.close() return { "success": True, "server": server_name, "ip": mcp_manager.server_configs[server_name].ssh_ip, "type": "file", "message": f"文件上传成功: {local_src} -> {remote_path}", "local_path": local_src, "remote_path": remote_path, "size": os.path.getsize(local_src) } except Exception as e: return { "success": False, "error": str(e), "server": server_name }
• main.py:180-240 (helper)

  Helper function for recursive directory upload via SFTP. Called by the upload_file handler when uploading directories. Handles creating remote directories and uploading files with error tracking.
  def upload_directory_recursive(sftp, local_path: str, remote_path: str) -> Dict[str, Any]: """递归上传目录""" try: # 创建远程目录 try: sftp.mkdir(remote_path) except IOError: # 目录可能已存在，忽略错误 pass uploaded_files = [] failed_files = [] for root, dirs, files in os.walk(local_path): # 计算相对路径 rel_path = os.path.relpath(root, local_path) if rel_path == '.': remote_root = remote_path else: remote_root = f"{remote_path}/{rel_path.replace(os.sep, '/')}" # 创建远程目录 for dir_name in dirs: remote_dir = f"{remote_root}/{dir_name}" try: sftp.mkdir(remote_dir) except IOError: # 目录可能已存在，忽略错误 pass # 上传文件 for file_name in files: local_file_path = os.path.join(root, file_name) remote_file_path = f"{remote_root}/{file_name}" try: sftp.put(local_file_path, remote_file_path) uploaded_files.append({ "local": local_file_path, "remote": remote_file_path, "size": os.path.getsize(local_file_path) }) logger.info(f"已上传文件: {local_file_path} -> {remote_file_path}") except Exception as e: failed_files.append({ "local": local_file_path, "remote": remote_file_path, "error": str(e) }) logger.error(f"上传失败: {local_file_path} - {str(e)}") return { "success": True, "uploaded_files": uploaded_files, "failed_files": failed_files } except Exception as e: return { "success": False, "error": str(e) }
• main.py:371-377 (schema)

  Docstring within the upload_file handler providing the tool description and parameter documentation, which serves as the schema for MCP tool invocation.
  """ 将本地的文件或者目录上传到远程服务器 参数: - server_name: 服务器名称 - remote_path: 远程文件/目录路径 - local_src: 本地文件/目录路径 """