MCP SSH Tools Server

Instructions

Implementation Reference

• main.py:369-453 (handler)

  The core handler function for the 'upload_file' tool. It is registered via the @mcp.tool() decorator. Handles uploading local files or directories (recursively using the helper function) to a remote SSH server via SFTP.

  @mcp.tool()
  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)

  Supporting helper function that performs recursive directory upload via SFTP, used by the upload_file handler when uploading directories.

  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)
          }