文件下载到沙箱
将文件下载到 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 配置文件写入成功")
}最佳实践
选择合适的方法
-
大多数场景:使用 FileCopy API
- 优势:高效、稳定、支持各种平台
- 需要:文件有可访问的 HTTP(S) URL
-
超大文件或特殊需求:使用 curl 命令
- 优势:灵活控制、支持断点续传
- 需要:Linux 沙箱,配合长期任务使用
-
小配置文件:使用 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("✗ 文件下载验证失败")
}
}安全建议
- 使用 HTTPS:确保下载链接使用 HTTPS 协议
- 验证文件:下载后验证文件完整性(checksum、文件大小等)
- 权限控制:设置合适的文件权限,避免敏感文件暴露
- 临时 URL:对于敏感文件,使用有时效的预签名 URL