MCP SSH Tools Server

Instructions

Implementation Reference

• main.py:281-367 (handler)

  Main handler function for the download_file tool, decorated with @mcp.tool() for registration in MCP. Handles single file and directory (recursive) downloads via SFTP.

  @mcp.tool()
  def download_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:
          sftp = client.open_sftp()
          
          # 检查远程路径是否存在
          try:
              file_attr = sftp.stat(remote_path)
          except IOError:
              sftp.close()
              return {
                  "success": False,
                  "error": f"远程路径不存在: {remote_path}",
                  "server": server_name
              }
          
          # 判断是文件还是目录
          if stat.S_ISDIR(file_attr.st_mode):
              # 是目录，递归下载
              result = download_directory_recursive(sftp, remote_path, local_src)
              sftp.close()
              
              if result["success"]:
                  return {
                      "success": True,
                      "server": server_name,
                      "ip": mcp_manager.server_configs[server_name].ssh_ip,
                      "type": "directory",
                      "message": f"目录下载完成: {remote_path} -> {local_src}",
                      "downloaded_files": result["downloaded_files"],
                      "failed_files": result["failed_files"],
                      "summary": {
                          "total_downloaded": len(result["downloaded_files"]),
                          "total_failed": len(result["failed_files"])
                      }
                  }
              else:
                  return {
                      "success": False,
                      "error": result["error"],
                      "server": server_name
                  }
          else:
              # 是文件，直接下载
              # 如果本地路径是目录，则在该目录下创建同名文件
              if os.path.isdir(local_src):
                  local_file_path = os.path.join(local_src, os.path.basename(remote_path))
              else:
                  local_file_path = local_src
                  # 创建本地目录
                  os.makedirs(os.path.dirname(local_file_path), exist_ok=True)
              
              sftp.get(remote_path, local_file_path)
              sftp.close()
              
              return {
                  "success": True,
                  "server": server_name,
                  "ip": mcp_manager.server_configs[server_name].ssh_ip,
                  "type": "file",
                  "message": f"文件下载成功: {remote_path} -> {local_file_path}",
                  "remote_path": remote_path,
                  "local_path": local_file_path,
                  "size": file_attr.st_size
              }
              
      except Exception as e:
          return {
              "success": False,
              "error": str(e),
              "server": server_name
          }

• main.py:131-178 (helper)

  Supporting recursive function for downloading entire directories, invoked by the main handler when the remote path is a directory.

  def download_directory_recursive(sftp, remote_path: str, local_path: str) -> Dict[str, Any]:
      """递归下载目录"""
      try:
          # 创建本地目录
          os.makedirs(local_path, exist_ok=True)
          
          # 获取远程目录内容
          files = sftp.listdir_attr(remote_path)
          downloaded_files = []
          failed_files = []
          
          for file_attr in files:
              remote_file_path = f"{remote_path}/{file_attr.filename}"
              local_file_path = os.path.join(local_path, file_attr.filename)
              
              try:
                  if stat.S_ISDIR(file_attr.st_mode):
                      # 如果是目录，递归下载
                      result = download_directory_recursive(sftp, remote_file_path, local_file_path)
                      downloaded_files.extend(result.get("downloaded_files", []))
                      failed_files.extend(result.get("failed_files", []))
                  else:
                      # 如果是文件，直接下载
                      sftp.get(remote_file_path, local_file_path)
                      downloaded_files.append({
                          "remote": remote_file_path,
                          "local": local_file_path,
                          "size": file_attr.st_size
                      })
                      logger.info(f"已下载文件: {remote_file_path} -> {local_file_path}")
              except Exception as e:
                  failed_files.append({
                      "remote": remote_file_path,
                      "local": local_file_path,
                      "error": str(e)
                  })
                  logger.error(f"下载失败: {remote_file_path} - {str(e)}")
          
          return {
              "success": True,
              "downloaded_files": downloaded_files,
              "failed_files": failed_files
          }
      except Exception as e:
          return {
              "success": False,
              "error": str(e)
          }

• main.py:282-289 (schema)

  Input schema defined by function parameters (server_name: str, remote_path: str, local_src: str) and descriptive docstring.

  def download_file(server_name: str, remote_path: str, local_src: str) -> Dict[str, Any]:
      """
      下载远程服务器的文件或者目录，保存到指定目录下，如果是目录，则递归下载
      参数:
      - server_name: 服务器名称
      - remote_path: 远程文件/目录路径
      - local_src: 本地文件/目录路径
      """