> ## Documentation Index
> Fetch the complete documentation index at: https://ppio.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# GPUs SDK 使用指南

本文介绍如何使用 `ppio-gpus` SDK 调用 Async Serverless Endpoint，以及如何自定义 worker handler。

## 1. 安装 SDK

在客户端环境或 worker 运行环境中安装 SDK：

```bash Bash icon="terminal" theme={null}
pip install ppio-gpus
```

## 2. 使用 SDK 提交任务

`ppio-gpus` SDK 默认请求地址为 `https://async-public.serverless.ppinfra.com/v1`。调用时只需要设置 API Key，并使用 Endpoint 名称创建客户端。这里的 API Key 用于客户端提交、查询、取消任务，不用于 worker 拉取任务或回传结果。

```python Python icon="python" theme={null}
import ppio_gpus

ppio_gpus.api_key = "sk_xxxx"

endpoint = ppio_gpus.Endpoint("0f43a6867e05fddd")
job = endpoint.run({
    "prompt": "a red apple on a table"
})

print(job.status())
output = job.output(timeout=300)
print(output)
```

## 3. ComfyUI 任务示例

对于 `image.ppinfra.com/prod-gpucloudpublic/comfyui-worker:v0.0.1`，任务输入需要包含 ComfyUI workflow。下面是一个最小示例：

```python Python icon="python" theme={null}
import ppio_gpus

ppio_gpus.api_key = "sk_xxxx"

endpoint = ppio_gpus.Endpoint("0f43a6867e05fddd")

job = endpoint.run({
    "workflow": {
        "4": {
            "class_type": "CheckpointLoaderSimple",
            "inputs": {"ckpt_name": "flux1-dev-fp8.safetensors"},
        },
        "5": {
            "class_type": "EmptyLatentImage",
            "inputs": {"width": 512, "height": 512, "batch_size": 1},
        },
        "6": {
            "class_type": "CLIPTextEncode",
            "inputs": {"clip": ["4", 1], "text": "a red apple on a table"},
        },
        "7": {
            "class_type": "CLIPTextEncode",
            "inputs": {"clip": ["4", 1], "text": "blurry, low quality"},
        },
        "3": {
            "class_type": "KSampler",
            "inputs": {
                "model": ["4", 0],
                "positive": ["6", 0],
                "negative": ["7", 0],
                "latent_image": ["5", 0],
                "seed": 42,
                "steps": 10,
                "cfg": 7,
                "sampler_name": "euler",
                "scheduler": "normal",
                "denoise": 1,
            },
        },
        "8": {
            "class_type": "VAEDecode",
            "inputs": {"samples": ["3", 0], "vae": ["4", 2]},
        },
        "9": {
            "class_type": "SaveImage",
            "inputs": {"filename_prefix": "test", "images": ["8", 0]},
        },
    },
    "output_node_id": "9",
})

print(job.status())
print(job.output(timeout=300))
```

## 4. 自定义 Handler

worker 侧使用 `ppio_gpus.start({"handler": handler})` 启动任务循环。平台会把任务内容传入 `handler(job)`：

* `job["id"]`：当前 job id
* `job["input"]`：提交任务时的 input 内容
* handler 返回的内容会作为任务 output
* 如果返回 dict 中包含 `error` 字段，任务会被标记为失败

最小 handler 示例：

```python Python icon="python" theme={null}
import time
import ppio_gpus


def handler(job: dict) -> dict:
    job_id = job["id"]
    job_input = job.get("input", {})
    prompt = job_input.get("prompt", "hello")

    ppio_gpus.progress_update(job, {
        "status": "running",
        "message": "job accepted",
    })

    time.sleep(1)

    return {
        "job_id": job_id,
        "prompt": prompt,
        "result": "ok",
    }


if __name__ == "__main__":
    ppio_gpus.start({"handler": handler})
```

## 5. 返回图片或文件

Async Serverless Endpoint 的 `status` 接口返回结果有大小限制。图片、视频等大文件建议先上传到对象存储，再在 output 中返回 URL。

在 Endpoint 环境变量中配置对象存储：

```bash Bash icon="terminal" theme={null}
BUCKET_ENDPOINT_URL=https://<your-bucket-endpoint>
BUCKET_ACCESS_KEY_ID=<your-access-key-id>
BUCKET_SECRET_ACCESS_KEY=<your-secret-access-key>
BUCKET_NAME=<your-bucket-name>
```

handler 中上传图片：

```python Python icon="python" theme={null}
import ppio_gpus


def handler(job: dict) -> dict:
    image_url = ppio_gpus.upload_image(job["id"], "/tmp/output.png")

    return {
        "images": [
            {
                "filename": "output.png",
                "url": image_url,
            }
        ]
    }


if __name__ == "__main__":
    ppio_gpus.start({"handler": handler})
```

## 6. 常见问题

### handler 返回什么会被认为失败？

如果 handler 返回 dict 且包含 `error` 字段，任务会被标记为失败：

```python Python icon="python" theme={null}
return {"error": "invalid input"}
```