> ## Documentation Index
> Fetch the complete documentation index at: https://novita.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# GPUs SDK 利用ガイド
このガイドでは、`novita-gpus` SDK を使用して Async Serverless Endpoint を呼び出す方法と、worker handler をカスタマイズする方法を説明します。
## 1. SDK のインストール
クライアント環境または worker ランタイム環境に SDK をインストールします。
```bash theme={"system"}
pip install novita-gpus
```
## 2. SDK でジョブを送信する
`novita-gpus` SDK のデフォルトリクエスト URL は `https://async-public.serverless.novita.ai/v1` です。Endpoint を呼び出すには、API Key を設定し、Endpoint 名を指定してクライアントを作成します。
```python theme={"system"}
import novita_gpus
novita_gpus.api_key = "sk_xxxx"
endpoint = novita_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 ジョブの例
`novitalabs/comfyui-worker:v0.0.1` の場合、ジョブ入力には ComfyUI workflow を含める必要があります。以下の最小例は、テスト済みのケースと一致します。
```python theme={"system"}
import novita_gpus
novita_gpus.api_key = "sk_xxxx"
endpoint = novita_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 側では、`novita_gpus.start({"handler": handler})` を呼び出してタスクループを開始します。プラットフォームはタスクデータを `handler(job)` に渡します。
* `job["id"]`: 現在のジョブ ID
* `job["input"]`: クライアントによって送信された入力内容
* handler の戻り値はジョブ出力として使用されます
* 返された dict に `error` フィールドが含まれる場合、ジョブは失敗としてマークされます
最小 handler の例:
```python theme={"system"}
import time
import novita_gpus
def handler(job: dict) -> dict:
job_id = job["id"]
job_input = job.get("input", {})
prompt = job_input.get("prompt", "hello")
novita_gpus.progress_update(job, {
"status": "running",
"message": "job accepted",
})
time.sleep(1)
return {
"job_id": job_id,
"prompt": prompt,
"result": "ok",
}
if __name__ == "__main__":
novita_gpus.start({"handler": handler})
```
## 5. 完全な Worker 例
完全な worker ソースコード、Docker イメージのビルド手順、タスク送信スクリプトについては、Novita GPUs Python examples を参照してください。
このリポジトリには 2 つの例が含まれています。
* comfyui-worker: `handler.py` から ComfyUI を実行し、生成された画像を返します。
* sleep-worker: 指定された時間だけ待機し、JSON 結果を返す最小限の `handler.py` です。
各サンプルディレクトリには以下が含まれています。
* `handler.py`: `handler(job)` を定義し、`novita_gpus.start({"handler": handler})` で worker を開始します。
* `Dockerfile`: worker イメージをビルドします。
* `requirements.txt`: `novita-gpus` をインストールします。
* `submit_task.py`: `novita-gpus` クライアント SDK を使用してタスクを送信します。
worker イメージをビルドし、自分の registry にプッシュします。
```bash theme={"system"}
IMAGE=/comfyui-worker:v0.0.1
docker buildx build --platform linux/amd64 \
-t "$IMAGE" \
--push .
```
タスク送信では、Endpoint 名の形式 `-` を使用します。
たとえば、Endpoint ID `o8UJWkag5WTn` と app 名 `async` から、次のようになります。
```text theme={"system"}
o8UJWkag5WTn-async
```
`submit_task.py` の例では、Endpoint ID と app 名を別々に受け取り、最終的な Endpoint 名を組み立てます。
```bash theme={"system"}
export NOVITA_API_KEY="sk_xxxx"
export NOVITA_ENDPOINT_ID="o8UJWkag5WTn"
export NOVITA_APP_NAME="async"
python submit_task.py \
--endpoint "$NOVITA_ENDPOINT_ID" \
--app-name "$NOVITA_APP_NAME" \
--api-key "$NOVITA_API_KEY"
```
スクリプト内では、組み立てた Endpoint 名を使用して SDK クライアントを作成します。
```python theme={"system"}
endpoint_name = f"{args.endpoint}-{args.app_name}"
endpoint = novita_gpus.Endpoint(endpoint_name, api_key=args.api_key)
```
## 6. 画像またはファイルを返す
Async Serverless Endpoint `status` API には出力サイズの制限があります。画像や動画などの大きなファイルについては、まずオブジェクトストレージにアップロードし、出力で URL を返してください。
Endpoint でオブジェクトストレージの環境変数を設定します。
```bash theme={"system"}
BUCKET_ENDPOINT_URL=https://
BUCKET_ACCESS_KEY_ID=
BUCKET_SECRET_ACCESS_KEY=
BUCKET_NAME=
```
handler で画像をアップロードします。
```python theme={"system"}
import novita_gpus
def handler(job: dict) -> dict:
image_url = novita_gpus.upload_image(job["id"], "/tmp/output.png")
return {
"images": [
{
"filename": "output.png",
"url": image_url,
}
]
}
if __name__ == "__main__":
novita_gpus.start({"handler": handler})
```
通常のファイルや bytes をアップロードすることもできます。
```python theme={"system"}
import novita_gpus
file_url = novita_gpus.upload_file("result.json", "/tmp/result.json")
bytes_url = novita_gpus.upload_bytes("result.txt", b"hello")
```
## 7. FAQ
### worker で API Key を設定する必要がありますか？
通常は不要です。API Key は主に、クライアントがジョブを送信、照会、キャンセルするために使用されます。
### どの handler 戻り値がジョブを失敗としてマークしますか？
handler が `error` フィールドを含む dict を返すと、ジョブは失敗としてマークされます。
```python theme={"system"}
return {"error": "invalid input"}
```