执行长期异步任务

本篇文档适用于 Windows 沙箱。如果你使用的是 Linux 沙箱，请参考在 Linux 沙箱中运行长期任务。

提示：对于快速执行的短命令（5 秒内完成），请使用 execSandboxProcess API。

工作原理

在 Windows 中，有三种主要方式运行长期后台任务：

PowerShell Jobs (Start-Job)：最简单的方式，适合短期到中期的后台任务
Scheduled Tasks (schtasks)：最可靠的方式，适合长期任务和需要持久化的任务
Start-Process：适合简单的后台进程启动，但监控能力较弱

本文档主要介绍前两种方式。

方法 1：使用 PowerShell Jobs (推荐用于中短期任务)

PowerShell Jobs 是最简单直接的方式，适合运行时间在几分钟到几小时的任务。

步骤概览

准备脚本或程序：将要运行的脚本/程序传输到沙箱
启动任务：使用 Start-Job 启动后台任务
查看状态：查询任务运行状态和结果

步骤 1：准备脚本或程序

有两种方式准备要运行的脚本或程序：

方式 A：通过文件拷贝

使用文件传输 API 将脚本上传到沙箱：

# 首先，将脚本上传到对象存储（获取下载 URL）
# 然后，从 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": "upload-training-script",
        "src": {
          "type": "httpGetLocation",
          "url": "https://storage.example.com/scripts/train.ps1"
        },
        "dest": {
          "type": "sandboxFileLocation",
          "path": "C:\\Users\\Administrator\\scripts\\train.ps1"
        }
      }
    ]
  }'

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:
        # 上传脚本到沙箱
        await client.sandbox.copy_files(
            "SBX-xxxx",
            SandboxFileCopyRequestDto(files=[
                FileCopyItem(
                    id="upload-training-script",
                    src=HttpGetLocation(
                        url="https://storage.example.com/scripts/train.ps1"
                    ),
                    dest=SandboxFileLocation(
                        path="C:\\Users\\Administrator\\scripts\\train.ps1"
                    )
                )
            ])
        )
        print("脚本上传成功")

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

// 上传脚本到沙箱
await lybic.copyFilesWithSandbox('SBX-xxxx', {
  files: [
    {
      id: 'upload-training-script',
      src: {
        type: 'httpGetLocation',
        url: 'https://storage.example.com/scripts/train.ps1',
      },
      dest: {
        type: 'sandboxFileLocation',
        path: 'C:\\Users\\Administrator\\scripts\\train.ps1',
      },
    },
  ],
})

console.log('脚本上传成功')

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: "upload-training-script",
                Src: map[string]string{
                    "type": "httpGetLocation",
                    "url":  "https://storage.example.com/scripts/train.ps1",
                },
                Dest: map[string]any{
                    "type": "sandboxFileLocation",
                    "path": "C:\\Users\\Administrator\\scripts\\train.ps1",
                },
            },
        },
    }

    _, err := client.CopyFilesWithSandbox(ctx, "SBX-xxxx", copyDto)
    if err != nil {
        fmt.Println("上传脚本出错:", err)
        return
    }
    fmt.Println("脚本上传成功")
}

方式 B：通过 stdin 写入

使用 execSandboxProcess 将脚本内容通过管道写入文件：

# Base64 编码脚本内容
SCRIPT_CONTENT=$(cat << 'EOF' | base64 -w0
Write-Host "开始训练模型..."
for ($i = 1; $i -le 100; $i++) {
    Start-Sleep -Seconds 1
    Write-Host "进度: $i/100"
}
Write-Host "训练完成!"
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": "powershell.exe",
    "args": ["-Command", "$input | Out-File -FilePath C:\\Users\\Administrator\\scripts\\train.ps1 -Encoding UTF8"],
    "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 = """Write-Host "开始训练模型..."
for ($i = 1; $i -le 100; $i++) {
    Start-Sleep -Seconds 1
    Write-Host "进度: $i/100"
}
Write-Host "训练完成!"
"""
        
        # Base64 编码
        stdin_data = base64.b64encode(script_content.encode('utf-8')).decode()
        
        # 写入文件
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="powershell.exe",
            args=["-Command", "$input | Out-File -FilePath C:\\Users\\Administrator\\scripts\\train.ps1 -Encoding UTF8"],
            stdinBase64=stdin_data
        )
        
        if result.exitCode == 0:
            print("脚本创建成功")
        else:
            print(f"创建失败: {base64.b64decode(result.stderrBase64 or '').decode()}")

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 = `Write-Host "开始训练模型..."
for ($i = 1; $i -le 100; $i++) {
    Start-Sleep -Seconds 1
    Write-Host "进度: $i/100"
}
Write-Host "训练完成!"
`

// Base64 编码
const stdinBase64 = btoa(scriptContent)

// 写入文件
const result = await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'powershell.exe',
  args: ['-Command', '$input | Out-File -FilePath C:\\Users\\Administrator\\scripts\\train.ps1 -Encoding UTF8'],
  stdinBase64: stdinBase64,
})

if (result.data?.exitCode === 0) {
  console.log('脚本创建成功')
} else {
  console.log('创建失败:', atob(result.data?.stderrBase64 || ''))
}

package main

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

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

    // 准备脚本内容
    scriptContent := `Write-Host "开始训练模型..."
for ($i = 1; $i -le 100; $i++) {
    Start-Sleep -Seconds 1
    Write-Host "进度: $i/100"
}
Write-Host "训练完成!"
`

    // Base64 编码
    stdinBase64 := base64.StdEncoding.EncodeToString([]byte(scriptContent))
    
    // 写入文件
    cmd := "$input | Out-File -FilePath C:\\Users\\Administrator\\scripts\\train.ps1 -Encoding UTF8"
    processDto := lybic.SandboxProcessRequestDto{
        Executable:  "powershell.exe",
        Args:        []string{"-Command", cmd},
        StdinBase64: &stdinBase64,
    }

    result, _ := client.ExecSandboxProcess(ctx, "SBX-xxxx", processDto)
    
    if result.ExitCode == 0 {
        fmt.Println("脚本创建成功")
    } else {
        stderr, _ := base64.StdEncoding.DecodeString(result.StderrBase64)
        fmt.Printf("创建失败: %s\n", string(stderr))
    }
}

步骤 2：使用 Start-Job 启动后台任务

使用 PowerShell 的 Start-Job 启动后台任务。该命令会立即返回，任务在后台继续运行。

# 启动后台任务
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": "powershell.exe",
    "args": [
      "-Command",
      "Start-Job -Name MyTrainingTask -FilePath C:\\Users\\Administrator\\scripts\\train.ps1"
    ]
  }'

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:
        # 启动后台任务
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="powershell.exe",
            args=[
                "-Command",
                "Start-Job -Name MyTrainingTask -FilePath C:\\Users\\Administrator\\scripts\\train.ps1"
            ]
        )
        
        output = base64.b64decode(result.stdoutBase64 or '').decode()
        print(f"任务已启动: {output}")
        print(f"退出代码: {result.exitCode}")

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: 'powershell.exe',
  args: [
    '-Command',
    'Start-Job -Name MyTrainingTask -FilePath C:\\Users\\Administrator\\scripts\\train.ps1',
  ],
})

const output = atob(result.data?.stdoutBase64 || '')
console.log('任务已启动:', output)
console.log('退出代码:', result.data?.exitCode)

package main

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

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

    // 启动后台任务
    processDto := lybic.SandboxProcessRequestDto{
        Executable: "powershell.exe",
        Args: []string{
            "-Command",
            "Start-Job -Name MyTrainingTask -FilePath C:\\Users\\Administrator\\scripts\\train.ps1",
        },
    }

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

    stdout, _ := base64.StdEncoding.DecodeString(result.StdoutBase64)
    fmt.Printf("任务已启动: %s\n", string(stdout))
    fmt.Printf("退出代码: %d\n", result.ExitCode)
}

重要参数说明：

-Name <名称>：指定任务名称，用于后续查询状态和结果
-FilePath <路径>：指定要执行的脚本文件
-ScriptBlock { ... }：或者直接提供脚本块（代替 -FilePath）

输出示例：

Id     Name               PSJobTypeName   State         HasMoreData     Location             Command
--     ----               -------------   -----         -----------     --------             -------
1      MyTrainingTask     BackgroundJob   Running       True            localhost            C:\Users\Administrator\scripts\train.ps1

步骤 3：查看任务状态

3.1 查看所有后台任务

使用 Get-Job 查看所有后台任务：

# 查看所有任务
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": "powershell.exe",
    "args": ["-Command", "Get-Job | Format-Table -AutoSize"]
  }'

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:
        # 查看所有任务
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="powershell.exe",
            args=["-Command", "Get-Job | Format-Table -AutoSize"]
        )
        
        jobs = base64.b64decode(result.stdoutBase64 or '').decode()
        print("所有任务:")
        print(jobs)

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: 'powershell.exe',
  args: ['-Command', 'Get-Job | Format-Table -AutoSize'],
})

const jobs = atob(result.data?.stdoutBase64 || '')
console.log('所有任务:')
console.log(jobs)

package main

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

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

    // 查看所有任务
    processDto := lybic.SandboxProcessRequestDto{
        Executable: "powershell.exe",
        Args:       []string{"-Command", "Get-Job | Format-Table -AutoSize"},
    }

    result, _ := client.ExecSandboxProcess(ctx, "SBX-xxxx", processDto)
    
    stdout, _ := base64.StdEncoding.DecodeString(result.StdoutBase64)
    fmt.Println("所有任务:")
    fmt.Println(string(stdout))
}

3.2 查看特定任务状态

使用 Get-Job -Name 查看特定任务的详细状态：

# 查看特定任务状态
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": "powershell.exe",
    "args": ["-Command", "Get-Job -Name MyTrainingTask | Format-List"]
  }'

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:
        # 查看任务状态
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="powershell.exe",
            args=["-Command", "Get-Job -Name MyTrainingTask | Format-List"]
        )
        
        status = base64.b64decode(result.stdoutBase64 or '').decode()
        print("任务状态:")
        print(status)

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: 'powershell.exe',
  args: ['-Command', 'Get-Job -Name MyTrainingTask | Format-List'],
})

const status = atob(result.data?.stdoutBase64 || '')
console.log('任务状态:')
console.log(status)

package main

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

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

    // 查看任务状态
    processDto := lybic.SandboxProcessRequestDto{
        Executable: "powershell.exe",
        Args:       []string{"-Command", "Get-Job -Name MyTrainingTask | Format-List"},
    }

    result, _ := client.ExecSandboxProcess(ctx, "SBX-xxxx", processDto)
    
    stdout, _ := base64.StdEncoding.DecodeString(result.StdoutBase64)
    fmt.Println("任务状态:")
    fmt.Println(string(stdout))
}

状态输出示例：

Id         : 1
Name       : MyTrainingTask
PSJobTypeName : BackgroundJob
State      : Running
HasMoreData: True
Location   : localhost
Command    : C:\Users\Administrator\scripts\train.ps1

任务状态说明：

Running：任务正在运行
Completed：任务已完成
Failed：任务失败
Stopped：任务被停止

3.3 获取任务输出

使用 Receive-Job 获取任务的输出：

# 获取任务输出（不删除输出）
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": "powershell.exe",
    "args": ["-Command", "Receive-Job -Name MyTrainingTask -Keep"]
  }'

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:
        # 获取任务输出
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="powershell.exe",
            args=["-Command", "Receive-Job -Name MyTrainingTask -Keep"]
        )
        
        output = base64.b64decode(result.stdoutBase64 or '').decode()
        print("任务输出:")
        print(output)

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: 'powershell.exe',
  args: ['-Command', 'Receive-Job -Name MyTrainingTask -Keep'],
})

const output = atob(result.data?.stdoutBase64 || '')
console.log('任务输出:')
console.log(output)

package main

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

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

    // 获取任务输出
    processDto := lybic.SandboxProcessRequestDto{
        Executable: "powershell.exe",
        Args:       []string{"-Command", "Receive-Job -Name MyTrainingTask -Keep"},
    }

    result, _ := client.ExecSandboxProcess(ctx, "SBX-xxxx", processDto)
    
    stdout, _ := base64.StdEncoding.DecodeString(result.StdoutBase64)
    fmt.Println("任务输出:")
    fmt.Println(string(stdout))
}

重要参数：

-Keep：保留输出，下次调用 Receive-Job 时仍能获取（不使用此参数会清除已读取的输出）
-Wait：等待任务完成（不推荐，会阻塞直到任务结束）

3.4 等待任务完成并获取结果

使用 Wait-Job 和 Receive-Job 等待任务完成：

import asyncio
import base64
from lybic import LybicClient, LybicAuth

async def wait_for_job(client, sandbox_id: str, job_name: str):
    """等待任务完成并获取结果"""
    while True:
        # 检查任务状态
        result = await client.sandbox.execute_process(
            sandbox_id,
            executable="powershell.exe",
            args=["-Command", f"(Get-Job -Name {job_name}).State"]
        )
        
        state = base64.b64decode(result.stdoutBase64 or '').decode().strip()
        print(f"任务状态: {state}")
        
        if state in ["Completed", "Failed", "Stopped"]:
            # 获取任务输出
            result = await client.sandbox.execute_process(
                sandbox_id,
                executable="powershell.exe",
                args=["-Command", f"Receive-Job -Name {job_name}"]
            )
            
            output = base64.b64decode(result.stdoutBase64 or '').decode()
            return state, output
        
        # 等待 5 秒后重新检查
        await asyncio.sleep(5)

async def main():
    async with LybicClient(
        LybicAuth(
            org_id="ORG-xxxx",
            api_key="lysk-xxxxxxxxxxx",
            endpoint="https://api.lybic.cn/"
        )
    ) as client:
        state, output = await wait_for_job(client, "SBX-xxxx", "MyTrainingTask")
        print(f"\n任务完成，状态: {state}")
        print(f"输出:\n{output}")

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

async function waitForJob(sandboxId: string, jobName: string) {
  while (true) {
    // 检查任务状态
    const statusResult = await lybic.execSandboxProcess(sandboxId, {
      executable: 'powershell.exe',
      args: ['-Command', `(Get-Job -Name ${jobName}).State`],
    })
    
    const state = atob(statusResult.data?.stdoutBase64 || '').trim()
    console.log('任务状态:', state)
    
    if (['Completed', 'Failed', 'Stopped'].includes(state)) {
      // 获取任务输出
      const outputResult = await lybic.execSandboxProcess(sandboxId, {
        executable: 'powershell.exe',
        args: ['-Command', `Receive-Job -Name ${jobName}`],
      })
      
      const output = atob(outputResult.data?.stdoutBase64 || '')
      return { state, output }
    }
    
    // 等待 5 秒后重新检查
    await new Promise(resolve => setTimeout(resolve, 5000))
  }
}

const { state, output } = await waitForJob('SBX-xxxx', 'MyTrainingTask')
console.log('\n任务完成，状态:', state)
console.log('输出:\n', output)

方法 2：使用 Scheduled Tasks (推荐用于长期任务)

Scheduled Tasks（计划任务）是最可靠的方式，适合需要长期运行或需要持久化的任务。计划任务即使在沙箱 重启(而非销毁) 后也能保留。

创建计划任务

使用 schtasks 命令创建立即运行的计划任务：

# 创建并启动计划任务
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": "schtasks",
    "args": [
      "/Create",
      "/TN", "MyTrainingTask",
      "/TR", "powershell.exe -File C:\\Users\\Administrator\\scripts\\train.ps1",
      "/SC", "ONCE",
      "/ST", "00:00",
      "/F"
    ]
  }'

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:
        # 创建计划任务
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="schtasks",
            args=[
                "/Create",
                "/TN", "MyTrainingTask",
                "/TR", "powershell.exe -File C:\\Users\\Administrator\\scripts\\train.ps1",
                "/SC", "ONCE",
                "/ST", "00:00",
                "/F"
            ]
        )
        
        output = base64.b64decode(result.stdoutBase64 or '').decode()
        print(f"计划任务创建结果: {output}")
        
        # 立即运行任务
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="schtasks",
            args=["/Run", "/TN", "MyTrainingTask"]
        )
        
        output = base64.b64decode(result.stdoutBase64 or '').decode()
        print(f"任务启动结果: {output}")

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

// 创建计划任务
let result = await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'schtasks',
  args: [
    '/Create',
    '/TN', 'MyTrainingTask',
    '/TR', 'powershell.exe -File C:\\Users\\Administrator\\scripts\\train.ps1',
    '/SC', 'ONCE',
    '/ST', '00:00',
    '/F',
  ],
})

console.log('计划任务创建结果:', atob(result.data?.stdoutBase64 || ''))

// 立即运行任务
result = await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'schtasks',
  args: ['/Run', '/TN', 'MyTrainingTask'],
})

console.log('任务启动结果:', atob(result.data?.stdoutBase64 || ''))

package main

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

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

    // 创建计划任务
    processDto := lybic.SandboxProcessRequestDto{
        Executable: "schtasks",
        Args: []string{
            "/Create",
            "/TN", "MyTrainingTask",
            "/TR", "powershell.exe -File C:\\Users\\Administrator\\scripts\\train.ps1",
            "/SC", "ONCE",
            "/ST", "00:00",
            "/F",
        },
    }

    result, _ := client.ExecSandboxProcess(ctx, "SBX-xxxx", processDto)
    stdout, _ := base64.StdEncoding.DecodeString(result.StdoutBase64)
    fmt.Printf("计划任务创建结果: %s\n", string(stdout))
    
    // 立即运行任务
    processDto = lybic.SandboxProcessRequestDto{
        Executable: "schtasks",
        Args:       []string{"/Run", "/TN", "MyTrainingTask"},
    }
    
    result, _ = client.ExecSandboxProcess(ctx, "SBX-xxxx", processDto)
    stdout, _ = base64.StdEncoding.DecodeString(result.StdoutBase64)
    fmt.Printf("任务启动结果: %s\n", string(stdout))
}

重要参数说明：

/Create：创建新任务
/TN <任务名>：任务名称
/TR <命令>：要执行的命令
/SC ONCE：运行一次（也可以用 DAILY、WEEKLY 等）
/ST <时间>：计划开始时间（格式：HH:MM）
/F：强制创建，如果任务已存在则覆盖
/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": "schtasks",
    "args": ["/Query", "/TN", "MyTrainingTask", "/V", "/FO", "LIST"]
  }'

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:
        # 查看任务状态
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="schtasks",
            args=["/Query", "/TN", "MyTrainingTask", "/V", "/FO", "LIST"]
        )
        
        status = base64.b64decode(result.stdoutBase64 or '').decode()
        print("任务状态:")
        print(status)

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: 'schtasks',
  args: ['/Query', '/TN', 'MyTrainingTask', '/V', '/FO', 'LIST'],
})

const status = atob(result.data?.stdoutBase64 || '')
console.log('任务状态:')
console.log(status)

package main

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

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

    // 查看任务状态
    processDto := lybic.SandboxProcessRequestDto{
        Executable: "schtasks",
        Args:       []string{"/Query", "/TN", "MyTrainingTask", "/V", "/FO", "LIST"},
    }

    result, _ := client.ExecSandboxProcess(ctx, "SBX-xxxx", processDto)
    
    stdout, _ := base64.StdEncoding.DecodeString(result.StdoutBase64)
    fmt.Println("任务状态:")
    fmt.Println(string(stdout))
}

参数说明：

/Query：查询任务
/V：详细输出
/FO LIST：格式化为列表输出

任务状态说明：

Ready：就绪（未运行）
Running：正在运行
Disabled：已禁用

输出重定向到文件

由于计划任务的输出不像 PowerShell Jobs 那样容易获取，推荐将输出重定向到文件：

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:
        # 创建带输出重定向的计划任务
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="schtasks",
            args=[
                "/Create",
                "/TN", "MyTrainingTask",
                "/TR", "powershell.exe -File C:\\Users\\Administrator\\scripts\\train.ps1 > C:\\Users\\Administrator\\logs\\output.log 2>&1",
                "/SC", "ONCE",
                "/ST", "00:00",
                "/F"
            ]
        )
        
        print("任务创建成功")
        
        # 启动任务
        await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="schtasks",
            args=["/Run", "/TN", "MyTrainingTask"]
        )
        
        print("任务已启动")
        
        # 等待一段时间后读取日志
        await asyncio.sleep(10)
        
        # 读取输出日志
        result = await client.sandbox.execute_process(
            "SBX-xxxx",
            executable="powershell.exe",
            args=["-Command", "Get-Content C:\\Users\\Administrator\\logs\\output.log"]
        )
        
        output = base64.b64decode(result.stdoutBase64 or '').decode()
        print("任务输出:")
        print(output)

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

// 创建带输出重定向的计划任务
await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'schtasks',
  args: [
    '/Create',
    '/TN', 'MyTrainingTask',
    '/TR', 'powershell.exe -File C:\\Users\\Administrator\\scripts\\train.ps1 > C:\\Users\\Administrator\\logs\\output.log 2>&1',
    '/SC', 'ONCE',
    '/ST', '00:00',
    '/F',
  ],
})

console.log('任务创建成功')

// 启动任务
await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'schtasks',
  args: ['/Run', '/TN', 'MyTrainingTask'],
})

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

// 等待一段时间后读取日志
await new Promise(resolve => setTimeout(resolve, 10000))

// 读取输出日志
const result = await lybic.execSandboxProcess('SBX-xxxx', {
  executable: 'powershell.exe',
  args: ['-Command', 'Get-Content C:\\Users\\Administrator\\logs\\output.log'],
})

const output = atob(result.data?.stdoutBase64 || '')
console.log('任务输出:')
console.log(output)

完整示例：运行 Python Web 服务器

以下是一个完整的示例，展示如何在 Windows 沙箱中运行一个长期的 Python Web 服务器：

import asyncio
import base64
from lybic import LybicClient, LybicAuth

async def run_web_server():
    async with LybicClient(
        LybicAuth(
            org_id="ORG-xxxx",
            api_key="lysk-xxxxxxxxxxx",
            endpoint="https://api.lybic.cn/"
        )
    ) as client:
        sandbox_id = "SBX-xxxx"
        
        # 1. 创建 Web 服务器脚本
        server_code = """
from http.server import HTTPServer, BaseHTTPRequestHandler
import json

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        response = {'status': 'ok', 'message': 'Hello from Lybic Windows!'}
        self.wfile.write(json.dumps(response).encode())
    
    def log_message(self, format, *args):
        print(f"{self.address_string()} - {format % args}")

if __name__ == '__main__':
    server = HTTPServer(('0.0.0.0', 8080), Handler)
    print("服务器启动在 http://0.0.0.0:8080")
    server.serve_forever()
"""
        
        stdin_data = base64.b64encode(server_code.encode('utf-8')).decode()
        
        # 创建目录和脚本文件
        result = await client.sandbox.execute_process(
            sandbox_id,
            executable="powershell.exe",
            args=["-Command", "New-Item -ItemType Directory -Force -Path C:\\Users\\Administrator\\server; $input | Out-File -FilePath C:\\Users\\Administrator\\server\\app.py -Encoding UTF8"],
            stdinBase64=stdin_data
        )
        
        if result.exitCode != 0:
            print("创建服务器脚本失败")
            return
        
        print("✓ 服务器脚本创建成功")
        
        # 2. 使用 Start-Job 启动服务器
        result = await client.sandbox.execute_process(
            sandbox_id,
            executable="powershell.exe",
            args=[
                "-Command",
                "Start-Job -Name WebServer -ScriptBlock { python C:\\Users\\Administrator\\server\\app.py }"
            ]
        )
        
        output = base64.b64decode(result.stdoutBase64 or '').decode()
        print(f"✓ Web 服务器已启动: {output}")
        
        # 3. 等待几秒让服务器启动
        await asyncio.sleep(3)
        
        # 4. 检查服务器状态
        result = await client.sandbox.execute_process(
            sandbox_id,
            executable="powershell.exe",
            args=["-Command", "(Get-Job -Name WebServer).State"]
        )
        
        status = base64.b64decode(result.stdoutBase64 or '').decode().strip()
        print(f"✓ 服务器状态: {status}")
        
        # 5. 查看任务输出
        result = await client.sandbox.execute_process(
            sandbox_id,
            executable="powershell.exe",
            args=["-Command", "Receive-Job -Name WebServer -Keep"]
        )
        
        logs = base64.b64decode(result.stdoutBase64 or '').decode()
        print("✓ 服务器日志:")
        print(logs)
        
        # 6. 测试服务器
        result = await client.sandbox.execute_process(
            sandbox_id,
            executable="powershell.exe",
            args=["-Command", "Invoke-WebRequest -Uri http://localhost:8080 -UseBasicParsing | Select-Object -ExpandProperty Content"]
        )
        
        response = base64.b64decode(result.stdoutBase64 or '').decode()
        print(f"✓ 服务器响应: {response}")

if __name__ == "__main__":
    asyncio.run(run_web_server())

import { LybicClient } from '@lybic/core'

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

async function runWebServer() {
  const sandboxId = 'SBX-xxxx'
  
  // 1. 创建 Web 服务器脚本
  const serverCode = `
from http.server import HTTPServer, BaseHTTPRequestHandler
import json

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        response = {'status': 'ok', 'message': 'Hello from Lybic Windows!'}
        self.wfile.write(json.dumps(response).encode())
    
    def log_message(self, format, *args):
        print(f"{self.address_string()} - {format % args}")

if __name__ == '__main__':
    server = HTTPServer(('0.0.0.0', 8080), Handler)
    print("服务器启动在 http://0.0.0.0:8080")
    server.serve_forever()
`
  
  const stdinBase64 = btoa(serverCode)
  
  // 创建目录和脚本文件
  let result = await lybic.execSandboxProcess(sandboxId, {
    executable: 'powershell.exe',
    args: ['-Command', 'New-Item -ItemType Directory -Force -Path C:\\Users\\Administrator\\server; $input | Out-File -FilePath C:\\Users\\Administrator\\server\\app.py -Encoding UTF8'],
    stdinBase64: stdinBase64,
  })
  
  if (result.data?.exitCode !== 0) {
    console.log('创建服务器脚本失败')
    return
  }
  
  console.log('✓ 服务器脚本创建成功')
  
  // 2. 使用 Start-Job 启动服务器
  result = await lybic.execSandboxProcess(sandboxId, {
    executable: 'powershell.exe',
    args: [
      '-Command',
      'Start-Job -Name WebServer -ScriptBlock { python C:\\Users\\Administrator\\server\\app.py }',
    ],
  })
  
  const output = atob(result.data?.stdoutBase64 || '')
  console.log('✓ Web 服务器已启动:', output)
  
  // 3. 等待几秒让服务器启动
  await new Promise(resolve => setTimeout(resolve, 3000))
  
  // 4. 检查服务器状态
  result = await lybic.execSandboxProcess(sandboxId, {
    executable: 'powershell.exe',
    args: ['-Command', '(Get-Job -Name WebServer).State'],
  })
  
  const status = atob(result.data?.stdoutBase64 || '').trim()
  console.log('✓ 服务器状态:', status)
  
  // 5. 查看任务输出
  result = await lybic.execSandboxProcess(sandboxId, {
    executable: 'powershell.exe',
    args: ['-Command', 'Receive-Job -Name WebServer -Keep'],
  })
  
  const logs = atob(result.data?.stdoutBase64 || '')
  console.log('✓ 服务器日志:')
  console.log(logs)
  
  // 6. 测试服务器
  result = await lybic.execSandboxProcess(sandboxId, {
    executable: 'powershell.exe',
    args: ['-Command', 'Invoke-WebRequest -Uri http://localhost:8080 -UseBasicParsing | Select-Object -ExpandProperty Content'],
  })
  
  const response = atob(result.data?.stdoutBase64 || '')
  console.log('✓ 服务器响应:', response)
}

runWebServer()

高级用法

停止 PowerShell Job

# 停止特定任务
Stop-Job -Name MyTrainingTask

# 删除任务
Remove-Job -Name MyTrainingTask

停止计划任务

# 停止任务
schtasks /End /TN MyTrainingTask

# 删除任务
schtasks /Delete /TN MyTrainingTask /F

列出所有任务

# 列出所有 PowerShell Jobs
Get-Job

# 列出所有计划任务
schtasks /Query /FO LIST

使用 Start-Process 启动后台进程

对于更简单的场景，可以使用 Start-Process：

Start-Process -FilePath "python" -ArgumentList "C:\Users\Administrator\scripts\train.py" -WindowStyle Hidden -RedirectStandardOutput "C:\Users\Administrator\logs\output.log"

注意：Start-Process 启动的进程较难追踪和管理，推荐使用 PowerShell Jobs 或 Scheduled Tasks。

方法对比

特性	PowerShell Jobs	Scheduled Tasks	Start-Process
易用性	⭐⭐⭐	⭐⭐	⭐⭐⭐
状态查询	⭐⭐⭐	⭐⭐	⭐
输出获取	⭐⭐⭐	⭐	⭐
持久化	❌	✅	❌
适合长期任务	⭐⭐	⭐⭐⭐	⭐⭐
资源隔离	⭐⭐	⭐⭐⭐	⭐⭐