Lybic Docs

文件下载到沙箱

将文件下载到 Lybic 沙箱中的多种方法

本文介绍如何将文件从外部来源下载到 Lybic 沙箱中。根据不同的场景和文件类型,你可以选择最合适的方法。

方法对比

方法适用场景支持系统优势限制
FileCopy API (推荐)较小的文件Windows/Linux/Android高效、稳定、支持多种协议需要可访问的 URL
curl 命令大文件、长时间下载Linux/Windows灵活、支持断点续传需要长期任务支持
stdin 写入小文本文件(< 1MB)Windows/Linux/Android简单、无需外部 URL仅适合小文件

方法一:FileCopy API(推荐)

FileCopy API 是下载文件到沙箱的推荐方式,支持从 HTTP/HTTPS URL 直接下载文件到沙箱指定路径。

详细文档请参考:文件传输

基本用法

curl -X POST "https://api.lybic.cn/api/orgs/{orgId}/sandboxes/{sandboxId}/file/copy" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "files": [
      {
        "id": "download-dataset",
        "src": {
          "type": "httpGetLocation",
          "url": "https://example.com/dataset.csv"
        },
        "dest": {
          "type": "sandboxFileLocation",
          "path": "/home/agent/data/dataset.csv"
        }
      }
    ]
  }'
import asyncio
from lybic import LybicClient, LybicAuth
from lybic.dto import (
    SandboxFileCopyRequestDto,
    FileCopyItem,
    SandboxFileLocation,
    HttpGetLocation
)

async def main():
    async with LybicClient(
        LybicAuth(
            org_id="ORG-xxxx",
            api_key="lysk-xxxxxxxxxxx",
            endpoint="https://api.lybic.cn/"
        )
    ) as client:
        # 下载文件到沙箱
        response = await client.sandbox.copy_files(
            "SBX-xxxx",
            SandboxFileCopyRequestDto(files=[
                FileCopyItem(
                    id="download-dataset",
                    src=HttpGetLocation(
                        url="https://example.com/dataset.csv"
                    ),
                    dest=SandboxFileLocation(
                        path="/home/agent/data/dataset.csv"
                    )
                )
            ])
        )
        print("文件下载成功:", response)

if __name__ == "__main__":
    asyncio.run(main())
import { LybicClient } from '@lybic/core'

const lybic = new LybicClient({
  baseUrl: 'https://api.lybic.cn',
  orgId: 'ORG-xxxx',
  apiKey: 'lysk-your-api-key-here',
})

// 下载文件到沙箱
const result = await lybic.copyFilesWithSandbox('SBX-xxxx', {
  files: [
    {
      id: 'download-dataset',
      src: {
        type: 'httpGetLocation',
        url: 'https://example.com/dataset.csv',
      },
      dest: {
        type: 'sandboxFileLocation',
        path: '/home/agent/data/dataset.csv',
      },
    },
  ],
})

console.log('文件下载成功:', result.data)
package main

import (
    "context"
    "fmt"
    "github.com/lybic/lybic-sdk-go"
)

func main() {
    ctx := context.Background()
    client, _ := lybic.NewClient(nil)

    copyDto := lybic.SandboxFileCopyRequestDto{
        Files: []lybic.SandboxFileCopyRequestDtoFiles{
            {
                Id: "download-dataset",
                Src: map[string]string{
                    "type": "httpGetLocation",
                    "url":  "https://example.com/dataset.csv",
                },
                Dest: map[string]any{
                    "type": "sandboxFileLocation",
                    "path": "/home/agent/data/dataset.csv",
                },
            },
        },
    }

    result, err := client.CopyFilesWithSandbox(ctx, "SBX-xxxx", copyDto)
    if err != nil {
        fmt.Println("下载文件出错:", err)
        return
    }
    fmt.Printf("文件下载成功: %+v\n", result)
}

携带自定义 Headers

如果下载需要认证或自定义请求头:

curl -X POST "https://api.lybic.cn/api/orgs/{orgId}/sandboxes/{sandboxId}/file/copy" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "files": [
      {
        "id": "download-private-file",
        "src": {
          "type": "httpGetLocation",
          "url": "https://storage.example.com/private/data.zip",
          "headers": {
            "Authorization": "Bearer STORAGE_TOKEN",
            "X-Custom-Header": "custom-value"
          }
        },
        "dest": {
          "type": "sandboxFileLocation",
          "path": "/home/agent/downloads/data.zip"
        }
      }
    ]
  }'
import asyncio
from lybic import LybicClient, LybicAuth
from lybic.dto import (
    SandboxFileCopyRequestDto,
    FileCopyItem,
    SandboxFileLocation,
    HttpGetLocation
)

async def main():
    async with LybicClient(
        LybicAuth(
            org_id="ORG-xxxx",
            api_key="lysk-xxxxxxxxxxx",
            endpoint="https://api.lybic.cn/"
        )
    ) as client:
        response = await client.sandbox.copy_files(
            "SBX-xxxx",
            SandboxFileCopyRequestDto(files=[
                FileCopyItem(
                    id="download-private-file",
                    src=HttpGetLocation(
                        url="https://storage.example.com/private/data.zip",
                        headers={
                            "Authorization": "Bearer STORAGE_TOKEN",
                            "X-Custom-Header": "custom-value"
                        }
                    ),
                    dest=SandboxFileLocation(
                        path="/home/agent/downloads/data.zip"
                    )
                )
            ])
        )
        print("私有文件下载成功:", response)

if __name__ == "__main__":
    asyncio.run(main())
import { LybicClient } from '@lybic/core'

const lybic = new LybicClient({
  baseUrl: 'https://api.lybic.cn',
  orgId: 'ORG-xxxx',
  apiKey: 'lysk-your-api-key-here',
})

const result = await lybic.copyFilesWithSandbox('SBX-xxxx', {
  files: [
    {
      id: 'download-private-file',
      src: {
        type: 'httpGetLocation',
        url: 'https://storage.example.com/private/data.zip',
        headers: {
          'Authorization': 'Bearer STORAGE_TOKEN',
          'X-Custom-Header': 'custom-value',
        },
      },
      dest: {
        type: 'sandboxFileLocation',
        path: '/home/agent/downloads/data.zip',
      },
    },
  ],
})

console.log('私有文件下载成功:', result.data)
package main

import (
    "context"
    "fmt"
    "github.com/lybic/lybic-sdk-go"
)

func main() {
    ctx := context.Background()
    client, _ := lybic.NewClient(nil)

    copyDto := lybic.SandboxFileCopyRequestDto{
        Files: []lybic.SandboxFileCopyRequestDtoFiles{
            {
                Id: "download-private-file",
                Src: map[string]any{
                    "type": "httpGetLocation",
                    "url":  "https://storage.example.com/private/data.zip",
                    "headers": map[string]string{
                        "Authorization":   "Bearer STORAGE_TOKEN",
                        "X-Custom-Header": "custom-value",
                    },
                },
                Dest: map[string]any{
                    "type": "sandboxFileLocation",
                    "path": "/home/agent/downloads/data.zip",
                },
            },
        },
    }

    result, err := client.CopyFilesWithSandbox(ctx, "SBX-xxxx", copyDto)
    if err != nil {
        fmt.Println("下载私有文件出错:", err)
        return
    }
    fmt.Printf("私有文件下载成功: %+v\n", result)
}

批量下载多个文件

curl -X POST "https://api.lybic.cn/api/orgs/{orgId}/sandboxes/{sandboxId}/file/copy" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "files": [
      {
        "id": "download-model",
        "src": {
          "type": "httpGetLocation",
          "url": "https://models.example.com/model.pth"
        },
        "dest": {
          "type": "sandboxFileLocation",
          "path": "/home/agent/models/model.pth"
        }
      },
      {
        "id": "download-config",
        "src": {
          "type": "httpGetLocation",
          "url": "https://configs.example.com/config.yaml"
        },
        "dest": {
          "type": "sandboxFileLocation",
          "path": "/home/agent/configs/config.yaml"
        }
      },
      {
        "id": "download-dataset",
        "src": {
          "type": "httpGetLocation",
          "url": "https://data.example.com/train.csv"
        },
        "dest": {
          "type": "sandboxFileLocation",
          "path": "/home/agent/data/train.csv"
        }
      }
    ]
  }'
import asyncio
from lybic import LybicClient, LybicAuth
from lybic.dto import (
    SandboxFileCopyRequestDto,
    FileCopyItem,
    SandboxFileLocation,
    HttpGetLocation
)

async def main():
    async with LybicClient(
        LybicAuth(
            org_id="ORG-xxxx",
            api_key="lysk-xxxxxxxxxxx",
            endpoint="https://api.lybic.cn/"
        )
    ) as client:
        # 批量下载多个文件
        response = await client.sandbox.copy_files(
            "SBX-xxxx",
            SandboxFileCopyRequestDto(files=[
                FileCopyItem(
                    id="download-model",
                    src=HttpGetLocation(url="https://models.example.com/model.pth"),
                    dest=SandboxFileLocation(path="/home/agent/models/model.pth")
                ),
                FileCopyItem(
                    id="download-config",
                    src=HttpGetLocation(url="https://configs.example.com/config.yaml"),
                    dest=SandboxFileLocation(path="/home/agent/configs/config.yaml")
                ),
                FileCopyItem(
                    id="download-dataset",
                    src=HttpGetLocation(url="https://data.example.com/train.csv"),
                    dest=SandboxFileLocation(path="/home/agent/data/train.csv")
                )
            ])
        )
        
        # 检查每个文件的下载结果
        for result in response.results:
            if result.success:
                print(f"✓ {result.id} 下载成功")
            else:
                print(f"✗ {result.id} 下载失败: {result.error}")

if __name__ == "__main__":
    asyncio.run(main())
import { LybicClient } from '@lybic/core'

const lybic = new LybicClient({
  baseUrl: 'https://api.lybic.cn',
  orgId: 'ORG-xxxx',
  apiKey: 'lysk-your-api-key-here',
})

// 批量下载多个文件
const result = await lybic.copyFilesWithSandbox('SBX-xxxx', {
  files: [
    {
      id: 'download-model',
      src: {
        type: 'httpGetLocation',
        url: 'https://models.example.com/model.pth',
      },
      dest: {
        type: 'sandboxFileLocation',
        path: '/home/agent/models/model.pth',
      },
    },
    {
      id: 'download-config',
      src: {
        type: 'httpGetLocation',
        url: 'https://configs.example.com/config.yaml',
      },
      dest: {
        type: 'sandboxFileLocation',
        path: '/home/agent/configs/config.yaml',
      },
    },
    {
      id: 'download-dataset',
      src: {
        type: 'httpGetLocation',
        url: 'https://data.example.com/train.csv',
      },
      dest: {
        type: 'sandboxFileLocation',
        path: '/home/agent/data/train.csv',
      },
    },
  ],
})

// 检查每个文件的下载结果
result.data.results.forEach((r) => {
  if (r.success) {
    console.log(`✓ ${r.id} 下载成功`)
  } else {
    console.log(`✗ ${r.id} 下载失败: ${r.error}`)
  }
})
package main

import (
    "context"
    "fmt"
    "github.com/lybic/lybic-sdk-go"
)

func main() {
    ctx := context.Background()
    client, _ := lybic.NewClient(nil)

    copyDto := lybic.SandboxFileCopyRequestDto{
        Files: []lybic.SandboxFileCopyRequestDtoFiles{
            {
                Id: "download-model",
                Src: map[string]string{
                    "type": "httpGetLocation",
                    "url":  "https://models.example.com/model.pth",
                },
                Dest: map[string]any{
                    "type": "sandboxFileLocation",
                    "path": "/home/agent/models/model.pth",
                },
            },
            {
                Id: "download-config",
                Src: map[string]string{
                    "type": "httpGetLocation",
                    "url":  "https://configs.example.com/config.yaml",
                },
                Dest: map[string]any{
                    "type": "sandboxFileLocation",
                    "path": "/home/agent/configs/config.yaml",
                },
            },
            {
                Id: "download-dataset",
                Src: map[string]string{
                    "type": "httpGetLocation",
                    "url":  "https://data.example.com/train.csv",
                },
                Dest: map[string]any{
                    "type": "sandboxFileLocation",
                    "path": "/home/agent/data/train.csv",
                },
            },
        },
    }

    result, err := client.CopyFilesWithSandbox(ctx, "SBX-xxxx", copyDto)
    if err != nil {
        fmt.Println("批量下载出错:", err)
        return
    }

    // 检查每个文件的下载结果
    for _, r := range result.Results {
        if r.Success {
            fmt.Printf("✓ %s 下载成功\n", r.Id)
        } else {
            fmt.Printf("✗ %s 下载失败: %s\n", r.Id, r.Error)
        }
    }
}

方法二:使用 curl 命令

对于需要更灵活控制的场景,可以在沙箱中直接执行 curl 命令下载文件。由于下载可能耗时较长,需要配合 长期任务 机制使用。

详细文档请参考:执行长期异步任务

使用 systemd 运行 curl 下载

# 使用 systemd --user run 运行下载任务
curl -X POST "https://api.lybic.cn/api/orgs/{orgId}/sandboxes/{sandboxId}/process" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "executable": "systemd-run",
    "args": [
      "--user",
      "--unit=download-large-file",
      "--description=下载大文件",
      "curl",
      "-L",
      "-o",
      "/home/agent/downloads/large-file.zip",
      "https://example.com/large-file.zip"
    ]
  }'
import asyncio
from lybic import LybicClient, LybicAuth

async def main():
    async with LybicClient(
        LybicAuth(
            org_id="ORG-xxxx",
            api_key="lysk-xxxxxxxxxxx",
            endpoint="https://api.lybic.cn/"
        )
    ) as client:
        # 启动下载任务
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="systemd-run",
            args=[
                "--user",
                "--unit=download-large-file",
                "--description=下载大文件",
                "curl",
                "-L",
                "-o",
                "/home/agent/downloads/large-file.zip",
                "https://example.com/large-file.zip"
            ]
        )
        print("下载任务已启动:", result.stdout)
        
        # 查看任务状态
        status = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="systemctl",
            args=["--user", "status", "download-large-file"]
        )
        print("任务状态:", status.stdout)

if __name__ == "__main__":
    asyncio.run(main())
import { LybicClient } from '@lybic/core'

const lybic = new LybicClient({
  baseUrl: 'https://api.lybic.cn',
  orgId: 'ORG-xxxx',
  apiKey: 'lysk-your-api-key-here',
})

// 启动下载任务
const result = await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'systemd-run',
  args: [
    '--user',
    '--unit=download-large-file',
    '--description=下载大文件',
    'curl',
    '-L',
    '-o',
    '/home/agent/downloads/large-file.zip',
    'https://example.com/large-file.zip',
  ],
})

console.log('下载任务已启动:', result.data.stdout)

// 查看任务状态
const status = await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'systemctl',
  args: ['--user', 'status', 'download-large-file'],
})

console.log('任务状态:', status.data.stdout)
package main

import (
    "context"
    "fmt"
    "github.com/lybic/lybic-sdk-go"
)

func main() {
    ctx := context.Background()
    client, _ := lybic.NewClient(nil)

    // 启动下载任务
    execDto := lybic.ExecSandboxProcessRequestDto{
        Executable: "systemd-run",
        Args: []string{
            "--user",
            "--unit=download-large-file",
            "--description=下载大文件",
            "curl",
            "-L",
            "-o",
            "/home/agent/downloads/large-file.zip",
            "https://example.com/large-file.zip",
        },
    }

    result, err := client.ExecSandboxProcess(ctx, "SBX-xxxx", execDto)
    if err != nil {
        fmt.Println("启动下载任务出错:", err)
        return
    }
    fmt.Println("下载任务已启动:", result.Stdout)

    // 查看任务状态
    statusDto := lybic.ExecSandboxProcessRequestDto{
        Executable: "systemctl",
        Args:       []string{"--user", "status", "download-large-file"},
    }

    status, err := client.ExecSandboxProcess(ctx, "SBX-xxxx", statusDto)
    if err != nil {
        fmt.Println("查询任务状态出错:", err)
        return
    }
    fmt.Println("任务状态:", status.Stdout)
}

带进度条的下载

# 使用 curl 的进度条功能
curl -X POST "https://api.lybic.cn/api/orgs/{orgId}/sandboxes/{sandboxId}/process" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "executable": "systemd-run",
    "args": [
      "--user",
      "--unit=download-with-progress",
      "bash",
      "-c",
      "curl -L --progress-bar -o /home/agent/downloads/dataset.tar.gz https://example.com/dataset.tar.gz 2>&1 | tee /home/agent/downloads/progress.log"
    ]
  }'
import asyncio
from lybic import LybicClient, LybicAuth

async def main():
    async with LybicClient(
        LybicAuth(
            org_id="ORG-xxxx",
            api_key="lysk-xxxxxxxxxxx",
            endpoint="https://api.lybic.cn/"
        )
    ) as client:
        # 启动带进度的下载任务
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="systemd-run",
            args=[
                "--user",
                "--unit=download-with-progress",
                "bash",
                "-c",
                "curl -L --progress-bar -o /home/agent/downloads/dataset.tar.gz "
                "https://example.com/dataset.tar.gz 2>&1 | tee /home/agent/downloads/progress.log"
            ]
        )
        print("下载任务已启动")
        
        # 等待一段时间后查看进度
        await asyncio.sleep(5)
        
        # 查看进度日志
        log = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="tail",
            args=["-n", "10", "/home/agent/downloads/progress.log"]
        )
        print("下载进度:", log.stdout)

if __name__ == "__main__":
    asyncio.run(main())
import { LybicClient } from '@lybic/core'

const lybic = new LybicClient({
  baseUrl: 'https://api.lybic.cn',
  orgId: 'ORG-xxxx',
  apiKey: 'lysk-your-api-key-here',
})

// 启动带进度的下载任务
const result = await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'systemd-run',
  args: [
    '--user',
    '--unit=download-with-progress',
    'bash',
    '-c',
    'curl -L --progress-bar -o /home/agent/downloads/dataset.tar.gz ' +
      'https://example.com/dataset.tar.gz 2>&1 | tee /home/agent/downloads/progress.log',
  ],
})

console.log('下载任务已启动')

// 等待一段时间后查看进度
await new Promise((resolve) => setTimeout(resolve, 5000))

// 查看进度日志
const log = await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'tail',
  args: ['-n', '10', '/home/agent/downloads/progress.log'],
})

console.log('下载进度:', log.data.stdout)
package main

import (
    "context"
    "fmt"
    "time"
    "github.com/lybic/lybic-sdk-go"
)

func main() {
    ctx := context.Background()
    client, _ := lybic.NewClient(nil)

    // 启动带进度的下载任务
    execDto := lybic.ExecSandboxProcessRequestDto{
        Executable: "systemd-run",
        Args: []string{
            "--user",
            "--unit=download-with-progress",
            "bash",
            "-c",
            "curl -L --progress-bar -o /home/agent/downloads/dataset.tar.gz " +
                "https://example.com/dataset.tar.gz 2>&1 | tee /home/agent/downloads/progress.log",
        },
    }

    result, err := client.ExecSandboxProcess(ctx, "SBX-xxxx", execDto)
    if err != nil {
        fmt.Println("启动下载任务出错:", err)
        return
    }
    fmt.Println("下载任务已启动:", result.Stdout)

    // 等待一段时间后查看进度
    time.Sleep(5 * time.Second)

    // 查看进度日志
    logDto := lybic.ExecSandboxProcessRequestDto{
        Executable: "tail",
        Args:       []string{"-n", "10", "/home/agent/downloads/progress.log"},
    }

    log, err := client.ExecSandboxProcess(ctx, "SBX-xxxx", logDto)
    if err != nil {
        fmt.Println("查看进度日志出错:", err)
        return
    }
    fmt.Println("下载进度:", log.Stdout)
}

带认证的下载

# 使用 Header 进行认证下载
curl -X POST "https://api.lybic.cn/api/orgs/{orgId}/sandboxes/{sandboxId}/process" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "executable": "systemd-run",
    "args": [
      "--user",
      "--unit=download-authenticated",
      "curl",
      "-L",
      "-H",
      "Authorization: Bearer STORAGE_TOKEN",
      "-o",
      "/home/agent/downloads/private-data.zip",
      "https://storage.example.com/private/data.zip"
    ]
  }'
import asyncio
from lybic import LybicClient, LybicAuth

async def main():
    async with LybicClient(
        LybicAuth(
            org_id="ORG-xxxx",
            api_key="lysk-xxxxxxxxxxx",
            endpoint="https://api.lybic.cn/"
        )
    ) as client:
        # 下载需要认证的文件
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="systemd-run",
            args=[
                "--user",
                "--unit=download-authenticated",
                "curl",
                "-L",
                "-H",
                "Authorization: Bearer STORAGE_TOKEN",
                "-o",
                "/home/agent/downloads/private-data.zip",
                "https://storage.example.com/private/data.zip"
            ]
        )
        print("认证下载任务已启动:", result.stdout)

if __name__ == "__main__":
    asyncio.run(main())
import { LybicClient } from '@lybic/core'

const lybic = new LybicClient({
  baseUrl: 'https://api.lybic.cn',
  orgId: 'ORG-xxxx',
  apiKey: 'lysk-your-api-key-here',
})

// 下载需要认证的文件
const result = await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'systemd-run',
  args: [
    '--user',
    '--unit=download-authenticated',
    'curl',
    '-L',
    '-H',
    'Authorization: Bearer STORAGE_TOKEN',
    '-o',
    '/home/agent/downloads/private-data.zip',
    'https://storage.example.com/private/data.zip',
  ],
})

console.log('认证下载任务已启动:', result.data.stdout)
package main

import (
    "context"
    "fmt"
    "github.com/lybic/lybic-sdk-go"
)

func main() {
    ctx := context.Background()
    client, _ := lybic.NewClient(nil)

    // 下载需要认证的文件
    execDto := lybic.ExecSandboxProcessRequestDto{
        Executable: "systemd-run",
        Args: []string{
            "--user",
            "--unit=download-authenticated",
            "curl",
            "-L",
            "-H",
            "Authorization: Bearer STORAGE_TOKEN",
            "-o",
            "/home/agent/downloads/private-data.zip",
            "https://storage.example.com/private/data.zip",
        },
    }

    result, err := client.ExecSandboxProcess(ctx, "SBX-xxxx", execDto)
    if err != nil {
        fmt.Println("启动认证下载任务出错:", err)
        return
    }
    fmt.Println("认证下载任务已启动:", result.Stdout)
}

方法三:使用 stdin 写入小文件

对于小的文本文件(通常小于 1MB),可以直接通过 stdinBase64 参数将内容写入沙箱文件。

适用场景

  • 配置文件(JSON、YAML、INI 等)
  • 脚本文件(Python、Shell、JavaScript 等)
  • 文本数据文件
  • 环境变量文件

基本用法

# Base64 编码文件内容
FILE_CONTENT=$(cat << 'EOF' | base64 -w0
{
  "model": "gpt-4",
  "temperature": 0.7,
  "max_tokens": 2000
}
EOF
)

# 写入文件到沙箱
curl -X POST "https://api.lybic.cn/api/orgs/{orgId}/sandboxes/{sandboxId}/process" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "executable": "bash",
    "args": ["-c", "cat > /home/agent/config/settings.json"],
    "stdinBase64": "'$FILE_CONTENT'"
  }'
import asyncio
import base64
from lybic import LybicClient, LybicAuth

async def main():
    async with LybicClient(
        LybicAuth(
            org_id="ORG-xxxx",
            api_key="lysk-xxxxxxxxxxx",
            endpoint="https://api.lybic.cn/"
        )
    ) as client:
        # 准备配置文件内容
        config_content = """{
  "model": "gpt-4",
  "temperature": 0.7,
  "max_tokens": 2000
}"""
        
        # Base64 编码
        stdin_data = base64.b64encode(config_content.encode()).decode()
        
        # 写入文件到沙箱
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="bash",
            args=["-c", "cat > /home/agent/config/settings.json"],
            stdin_base64=stdin_data
        )
        print("配置文件写入成功:", result.stdout)

if __name__ == "__main__":
    asyncio.run(main())
import { LybicClient } from '@lybic/core'

const lybic = new LybicClient({
  baseUrl: 'https://api.lybic.cn',
  orgId: 'ORG-xxxx',
  apiKey: 'lysk-your-api-key-here',
})

// 准备配置文件内容
const configContent = `{
  "model": "gpt-4",
  "temperature": 0.7,
  "max_tokens": 2000
}`

// Base64 编码
const stdinData = Buffer.from(configContent).toString('base64')

// 写入文件到沙箱
const result = await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'bash',
  args: ['-c', 'cat > /home/agent/config/settings.json'],
  stdinBase64: stdinData,
})

console.log('配置文件写入成功:', result.data.stdout)
package main

import (
    "context"
    "encoding/base64"
    "fmt"
    "github.com/lybic/lybic-sdk-go"
)

func main() {
    ctx := context.Background()
    client, _ := lybic.NewClient(nil)

    // 准备配置文件内容
    configContent := `{
  "model": "gpt-4",
  "temperature": 0.7,
  "max_tokens": 2000
}`

    // Base64 编码
    stdinData := base64.StdEncoding.EncodeToString([]byte(configContent))

    // 写入文件到沙箱
    execDto := lybic.ExecSandboxProcessRequestDto{
        Executable:  "bash",
        Args:        []string{"-c", "cat > /home/agent/config/settings.json"},
        StdinBase64: &stdinData,
    }

    result, err := client.ExecSandboxProcess(ctx, "SBX-xxxx", execDto)
    if err != nil {
        fmt.Println("写入配置文件出错:", err)
        return
    }
    fmt.Println("配置文件写入成功:", result.Stdout)
}

写入脚本文件

# Base64 编码脚本内容
SCRIPT_CONTENT=$(cat << 'EOF' | base64 -w0
#!/usr/bin/env python3
import sys

def main():
    print("Hello from sandbox!")
    print(f"Python version: {sys.version}")

if __name__ == "__main__":
    main()
EOF
)

# 写入脚本并添加执行权限
curl -X POST "https://api.lybic.cn/api/orgs/{orgId}/sandboxes/{sandboxId}/process" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "executable": "bash",
    "args": ["-c", "cat > /home/agent/scripts/hello.py && chmod +x /home/agent/scripts/hello.py"],
    "stdinBase64": "'$SCRIPT_CONTENT'"
  }'
import asyncio
import base64
from lybic import LybicClient, LybicAuth

async def main():
    async with LybicClient(
        LybicAuth(
            org_id="ORG-xxxx",
            api_key="lysk-xxxxxxxxxxx",
            endpoint="https://api.lybic.cn/"
        )
    ) as client:
        # 准备脚本内容
        script_content = """#!/usr/bin/env python3
import sys

def main():
    print("Hello from sandbox!")
    print(f"Python version: {sys.version}")

if __name__ == "__main__":
    main()
"""
        
        # Base64 编码
        stdin_data = base64.b64encode(script_content.encode()).decode()
        
        # 写入脚本并添加执行权限
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="bash",
            args=["-c", "cat > /home/agent/scripts/hello.py && chmod +x /home/agent/scripts/hello.py"],
            stdin_base64=stdin_data
        )
        print("脚本文件写入成功")
        
        # 验证脚本可以执行
        run_result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="python3",
            args=["/home/agent/scripts/hello.py"]
        )
        print("脚本执行结果:", run_result.stdout)

if __name__ == "__main__":
    asyncio.run(main())
import { LybicClient } from '@lybic/core'

const lybic = new LybicClient({
  baseUrl: 'https://api.lybic.cn',
  orgId: 'ORG-xxxx',
  apiKey: 'lysk-your-api-key-here',
})

// 准备脚本内容
const scriptContent = `#!/usr/bin/env python3
import sys

def main():
    print("Hello from sandbox!")
    print(f"Python version: {sys.version}")

if __name__ == "__main__":
    main()
`

// Base64 编码
const stdinData = Buffer.from(scriptContent).toString('base64')

// 写入脚本并添加执行权限
await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'bash',
  args: ['-c', 'cat > /home/agent/scripts/hello.py && chmod +x /home/agent/scripts/hello.py'],
  stdinBase64: stdinData,
})

console.log('脚本文件写入成功')

// 验证脚本可以执行
const runResult = await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'python3',
  args: ['/home/agent/scripts/hello.py'],
})

console.log('脚本执行结果:', runResult.data.stdout)
package main

import (
    "context"
    "encoding/base64"
    "fmt"
    "github.com/lybic/lybic-sdk-go"
)

func main() {
    ctx := context.Background()
    client, _ := lybic.NewClient(nil)

    // 准备脚本内容
    scriptContent := `#!/usr/bin/env python3
import sys

def main():
    print("Hello from sandbox!")
    print(f"Python version: {sys.version}")

if __name__ == "__main__":
    main()
`

    // Base64 编码
    stdinData := base64.StdEncoding.EncodeToString([]byte(scriptContent))

    // 写入脚本并添加执行权限
    execDto := lybic.ExecSandboxProcessRequestDto{
        Executable:  "bash",
        Args:        []string{"-c", "cat > /home/agent/scripts/hello.py && chmod +x /home/agent/scripts/hello.py"},
        StdinBase64: &stdinData,
    }

    result, err := client.ExecSandboxProcess(ctx, "SBX-xxxx", execDto)
    if err != nil {
        fmt.Println("写入脚本文件出错:", err)
        return
    }
    fmt.Println("脚本文件写入成功")

    // 验证脚本可以执行
    runDto := lybic.ExecSandboxProcessRequestDto{
        Executable: "python3",
        Args:       []string{"/home/agent/scripts/hello.py"},
    }

    runResult, err := client.ExecSandboxProcess(ctx, "SBX-xxxx", runDto)
    if err != nil {
        fmt.Println("执行脚本出错:", err)
        return
    }
    fmt.Println("脚本执行结果:", runResult.Stdout)
}

写入多行配置文件

# Base64 编码 YAML 配置
YAML_CONTENT=$(cat << 'EOF' | base64 -w0
database:
  host: localhost
  port: 5432
  name: myapp
  user: admin
  password: secret

redis:
  host: localhost
  port: 6379
  db: 0

logging:
  level: INFO
  format: json
EOF
)

# 写入配置文件
curl -X POST "https://api.lybic.cn/api/orgs/{orgId}/sandboxes/{sandboxId}/process" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "executable": "bash",
    "args": ["-c", "cat > /home/agent/config/app.yaml"],
    "stdinBase64": "'$YAML_CONTENT'"
  }'
import asyncio
import base64
from lybic import LybicClient, LybicAuth

async def main():
    async with LybicClient(
        LybicAuth(
            org_id="ORG-xxxx",
            api_key="lysk-xxxxxxxxxxx",
            endpoint="https://api.lybic.cn/"
        )
    ) as client:
        # 准备 YAML 配置内容
        yaml_content = """database:
  host: localhost
  port: 5432
  name: myapp
  user: admin
  password: secret

redis:
  host: localhost
  port: 6379
  db: 0

logging:
  level: INFO
  format: json
"""
        
        # Base64 编码
        stdin_data = base64.b64encode(yaml_content.encode()).decode()
        
        # 写入配置文件
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="bash",
            args=["-c", "cat > /home/agent/config/app.yaml"],
            stdin_base64=stdin_data
        )
        print("YAML 配置文件写入成功")

if __name__ == "__main__":
    asyncio.run(main())
import { LybicClient } from '@lybic/core'

const lybic = new LybicClient({
  baseUrl: 'https://api.lybic.cn',
  orgId: 'ORG-xxxx',
  apiKey: 'lysk-your-api-key-here',
})

// 准备 YAML 配置内容
const yamlContent = `database:
  host: localhost
  port: 5432
  name: myapp
  user: admin
  password: secret

redis:
  host: localhost
  port: 6379
  db: 0

logging:
  level: INFO
  format: json
`

// Base64 编码
const stdinData = Buffer.from(yamlContent).toString('base64')

// 写入配置文件
await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'bash',
  args: ['-c', 'cat > /home/agent/config/app.yaml'],
  stdinBase64: stdinData,
})

console.log('YAML 配置文件写入成功')
package main

import (
    "context"
    "encoding/base64"
    "fmt"
    "github.com/lybic/lybic-sdk-go"
)

func main() {
    ctx := context.Background()
    client, _ := lybic.NewClient(nil)

    // 准备 YAML 配置内容
    yamlContent := `database:
  host: localhost
  port: 5432
  name: myapp
  user: admin
  password: secret

redis:
  host: localhost
  port: 6379
  db: 0

logging:
  level: INFO
  format: json
`

    // Base64 编码
    stdinData := base64.StdEncoding.EncodeToString([]byte(yamlContent))

    // 写入配置文件
    execDto := lybic.ExecSandboxProcessRequestDto{
        Executable:  "bash",
        Args:        []string{"-c", "cat > /home/agent/config/app.yaml"},
        StdinBase64: &stdinData,
    }

    result, err := client.ExecSandboxProcess(ctx, "SBX-xxxx", execDto)
    if err != nil {
        fmt.Println("写入 YAML 配置文件出错:", err)
        return
    }
    fmt.Println("YAML 配置文件写入成功")
}

最佳实践

选择合适的方法

  1. 大多数场景:使用 FileCopy API

    • 优势:高效、稳定、支持各种平台
    • 需要:文件有可访问的 HTTP(S) URL
  2. 超大文件或特殊需求:使用 curl 命令

    • 优势:灵活控制、支持断点续传
    • 需要:Linux 沙箱,配合长期任务使用
  3. 小配置文件:使用 stdin 写入

    • 优势:简单直接、无需外部 URL
    • 限制:仅适合小文件(< 1MB)

错误处理

建议在下载后验证文件:

# 验证文件是否存在且非空
curl -X POST "https://api.lybic.cn/api/orgs/{orgId}/sandboxes/{sandboxId}/process" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "executable": "bash",
    "args": ["-c", "test -s /home/agent/downloads/file.zip && echo \"文件下载成功\" || echo \"文件下载失败\""]
  }'
import asyncio
import base64
from lybic import LybicClient, LybicAuth

async def verify_file_download(client, sandbox_id, file_path):
    """验证文件是否下载成功"""
    result = await client.sandbox.execute_process(
        sandbox_id,
        executable="bash",
        args=["-c", f"test -s {file_path} && echo '文件存在' || echo '文件不存在'"]
    )
    return "文件存在" in base64.b64decode(result.stdoutBase64 or '').decode()

async def main():
    async with LybicClient(
        LybicAuth(
            org_id="ORG-xxxx",
            api_key="lysk-xxxxxxxxxxx",
            endpoint="https://api.lybic.cn/"
        )
    ) as client:
        # 下载文件
        await client.sandbox.copy_files("SBX-xxxx", ...)
        
        # 验证下载
        if await verify_file_download(client, "SBX-xxxx", "/home/agent/downloads/file.zip"):
            print("✓ 文件下载验证成功")
        else:
            print("✗ 文件下载验证失败")

if __name__ == "__main__":
    asyncio.run(main())
import { LybicClient } from '@lybic/core'

async function verifyFileDownload(
  lybic: LybicClient,
  sandboxId: string,
  filePath: string
): Promise<boolean> {
  const result = await lybic.execSandboxProcess(sandboxId, {
    executable: 'bash',
    args: ['-c', `test -s ${filePath} && echo '文件存在' || echo '文件不存在'`],
  })
  return atob(result.data?.stdoutBase64 || '').includes('文件存在')
}

const lybic = new LybicClient({
  baseUrl: 'https://api.lybic.cn',
  orgId: 'ORG-xxxx',
  apiKey: 'lysk-your-api-key-here',
})

// 下载文件
await lybic.copyFilesWithSandbox('SBX-xxxx', { ... })

// 验证下载
if (await verifyFileDownload(lybic, 'SBX-xxxx', '/home/agent/downloads/file.zip')) {
  console.log('✓ 文件下载验证成功')
} else {
  console.log('✗ 文件下载验证失败')
}
package main

import (
    "context"
    "fmt"
    "base64"
    "strings"
    "github.com/lybic/lybic-sdk-go"
)

func verifyFileDownload(ctx context.Context, client *lybic.Client, sandboxID, filePath string) (bool, error) {
    execDto := lybic.ExecSandboxProcessRequestDto{
        Executable: "bash",
        Args:       []string{"-c", fmt.Sprintf("test -s %s && echo '文件存在' || echo '文件不存在'", filePath)},
    }

    result, err := client.ExecSandboxProcess(ctx, sandboxID, execDto)
    if err != nil {
        return false, err
    }
    stdout, _ := base64.StdEncoding.DecodeString(result.StdoutBase64)
    return strings.Contains(stdout, "文件存在"), nil
}

func main() {
    ctx := context.Background()
    client, _ := lybic.NewClient(nil)

    // 下载文件
    // ...

    // 验证下载
    success, err := verifyFileDownload(ctx, client, "SBX-xxxx", "/home/agent/downloads/file.zip")
    if err != nil {
        fmt.Println("验证出错:", err)
        return
    }

    if success {
        fmt.Println("✓ 文件下载验证成功")
    } else {
        fmt.Println("✗ 文件下载验证失败")
    }
}

安全建议

  1. 使用 HTTPS:确保下载链接使用 HTTPS 协议
  2. 验证文件:下载后验证文件完整性(checksum、文件大小等)
  3. 权限控制:设置合适的文件权限,避免敏感文件暴露
  4. 临时 URL:对于敏感文件,使用有时效的预签名 URL

相关文档

本页内容