# GPUs SDK Usage Guide - Documentation

> For the complete documentation index, see [llms.txt](/llms.txt). Markdown is available with `Accept: text/markdown` and `.md` URL variants.

Source: /docs/guides/serverless-gpus-sdk-usage

# GPUs SDK Usage Guide

This guide explains how to use the `novita-gpus` SDK to call an Async Serverless Endpoint and how to customize a worker handler.

##

[​](#1-install-sdk)

1. Install SDK

Install the SDK in your client environment or worker runtime environment:

```
pip install novita-gpus
```

##

[​](#2-submit-jobs-with-sdk)

2. Submit Jobs with SDK

The `novita-gpus` SDK default request URL is `https://async-public.serverless.novita.ai/v1`. To call an Endpoint, set your API Key and create a client with the Endpoint name.

```
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-job-example)

3. ComfyUI Job Example

For `novitalabs/comfyui-worker:v0.0.1`, the job input must include a ComfyUI workflow. The following minimal example matches the tested case:

```
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-custom-handler)

4. Custom Handler

On the worker side, call `novita_gpus.start({"handler": handler})` to start the task loop. The platform passes task data to `handler(job)`:

- `job["id"]`: current job id

- `job["input"]`: input content submitted by the client

- The handler return value is used as the job output

- If the returned dict contains an `error` field, the job is marked as failed

Minimal handler example:

```
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-complete-worker-examples)

5. Complete Worker Examples

For complete worker source code, Docker image build instructions, and task submission scripts, see the [Novita GPUs Python examples](https://github.com/novitalabs/gpus-python-example).
The repository includes two examples:

- [comfyui-worker](https://github.com/novitalabs/gpus-python-example/tree/main/workers/comfyui-worker): runs ComfyUI from `handler.py` and returns generated images.

- [sleep-worker](https://github.com/novitalabs/gpus-python-example/tree/main/workers/sleep-worker): a minimal `handler.py` that waits for a requested duration and returns a JSON result.

Each example directory contains:

- `handler.py`: defines `handler(job)` and starts the worker with `novita_gpus.start({"handler": handler})`.

- `Dockerfile`: builds the worker image.

- `requirements.txt`: installs `novita-gpus`.

- `submit_task.py`: submits a task with the `novita-gpus` client SDK.

Build and push your worker image to your own registry:

```
IMAGE=/comfyui-worker:v0.0.1

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

Task submission uses the Endpoint name format `<endpoint-id>-<app-name>`.
For example, Endpoint ID `o8UJWkag5WTn` and app name `async` produce:

```
o8UJWkag5WTn-async
```

The `submit_task.py` examples accept the Endpoint ID and app name separately, then compose the final Endpoint name:

```
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"
```

Inside the script, the SDK client is created with the composed Endpoint name:

```
endpoint_name = f"{args.endpoint}-{args.app_name}"
endpoint = novita_gpus.Endpoint(endpoint_name, api_key=args.api_key)
```

##

[​](#6-return-images-or-files)

6. Return Images or Files

The Async Serverless Endpoint `status` API has an output size limit. For large files such as images and videos, upload them to object storage first and return URLs in the output.
Configure object storage environment variables in the Endpoint:

```
BUCKET_ENDPOINT_URL=https://
BUCKET_ACCESS_KEY_ID=
BUCKET_SECRET_ACCESS_KEY=
BUCKET_NAME=
```

Upload an image in the 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})
```

You can also upload regular files or 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)

7. FAQ

###

[​](#do-i-need-to-configure-an-api-key-in-the-worker)

Do I need to configure an API Key in the worker?

Usually no. API Keys are mainly used by clients to submit, query, and cancel jobs.

###

[​](#what-handler-return-value-marks-a-job-as-failed)

What handler return value marks a job as failed?

If the handler returns a dict that contains an `error` field, the job is marked as failed:

```
return {"error": "invalid input"}
```

Last modified on May 28, 2026