> ## 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 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
Install the SDK in your client environment or worker runtime environment:
```bash theme={"system"}
pip install novita-gpus
```
## 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.
```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 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:
```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. 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:
```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. Complete Worker Examples
For complete worker source code, Docker image build instructions, and task submission scripts, see the Novita GPUs Python examples.
The repository includes two examples:
* comfyui-worker: runs ComfyUI from `handler.py` and returns generated images.
* 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:
```bash theme={"system"}
IMAGE=/comfyui-worker:v0.0.1
docker buildx build --platform linux/amd64 \
-t "$IMAGE" \
--push .
```
Task submission uses the Endpoint name format `-`.
For example, Endpoint ID `o8UJWkag5WTn` and app name `async` produce:
```text theme={"system"}
o8UJWkag5WTn-async
```
The `submit_task.py` examples accept the Endpoint ID and app name separately, then compose the final Endpoint name:
```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"
```
Inside the script, the SDK client is created with the composed Endpoint name:
```python theme={"system"}
endpoint_name = f"{args.endpoint}-{args.app_name}"
endpoint = novita_gpus.Endpoint(endpoint_name, api_key=args.api_key)
```
## 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:
```bash theme={"system"}
BUCKET_ENDPOINT_URL=https://
BUCKET_ACCESS_KEY_ID=
BUCKET_SECRET_ACCESS_KEY=
BUCKET_NAME=
```
Upload an image in the 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})
```
You can also upload regular files or 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
### 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?
If the handler returns a dict that contains an `error` field, the job is marked as failed:
```python theme={"system"}
return {"error": "invalid input"}
```