GPUs SDK 利用ガイド

このガイドでは、novita-gpus SDK を使用して Async Serverless Endpoint を呼び出す方法と、worker handler をカスタマイズする方法を説明します。

1. SDK のインストール

クライアント環境または worker ランタイム環境に SDK をインストールします。

pip install novita-gpus

2. SDK でジョブを送信する

novita-gpus SDK のデフォルトリクエスト URL は https://async-public.serverless.novita.ai/v1 です。Endpoint を呼び出すには、API Key を設定し、Endpoint 名を指定してクライアントを作成します。

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 を含める必要があります。以下の最小例は、テスト済みのケースと一致します。

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 の例:

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 にプッシュします。

IMAGE=<your-registry>/comfyui-worker:v0.0.1

docker buildx build --platform linux/amd64 \
  -t "$IMAGE" \
  --push .

タスク送信では、Endpoint 名の形式 <endpoint-id>-<app-name> を使用します。たとえば、Endpoint ID o8UJWkag5WTn と app 名 async から、次のようになります。

o8UJWkag5WTn-async

submit_task.py の例では、Endpoint ID と app 名を別々に受け取り、最終的な Endpoint 名を組み立てます。

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 クライアントを作成します。

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 でオブジェクトストレージの環境変数を設定します。

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 で画像をアップロードします。

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 をアップロードすることもできます。

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 を返すと、ジョブは失敗としてマークされます。

return {"error": "invalid input"}

Get started

Model APIs

Agent Sandbox

GPUs

Observability

Resources

GPUs SDK 利用ガイド

1. SDK のインストール

2. SDK でジョブを送信する

3. ComfyUI ジョブの例

4. カスタム Handler

5. 完全な Worker 例

6. 画像またはファイルを返す

7. FAQ

worker で API Key を設定する必要がありますか？

どの handler 戻り値がジョブを失敗としてマークしますか？

​1. SDK のインストール

​2. SDK でジョブを送信する

​3. ComfyUI ジョブの例

​4. カスタム Handler

​5. 完全な Worker 例

​6. 画像またはファイルを返す

​7. FAQ

​worker で API Key を設定する必要がありますか？

​どの handler 戻り値がジョブを失敗としてマークしますか？

1. SDK のインストール

2. SDK でジョブを送信する

3. ComfyUI ジョブの例

4. カスタム Handler

5. 完全な Worker 例

6. 画像またはファイルを返す

7. FAQ

worker で API Key を設定する必要がありますか？

どの handler 戻り値がジョブを失敗としてマークしますか？