Skip to main content

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.

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: 
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. 
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: 
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: 
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: 
IMAGE=<your-registry>/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

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://<your-bucket-endpoint>
BUCKET_ACCESS_KEY_ID=<your-access-key-id>
BUCKET_SECRET_ACCESS_KEY=<your-secret-access-key>
BUCKET_NAME=<your-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

​
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: 
return {"error": "invalid input"}