Guia de Uso do SDK de GPUs

Este guia explica como usar o SDK novita-gpus para chamar um Endpoint Serverless Assíncrono e como personalizar um handler de worker.

1. Instalar o SDK

Instale o SDK no seu ambiente de cliente ou no ambiente de runtime do worker:

pip install novita-gpus

2. Enviar Jobs com o SDK

A URL de solicitação padrão do SDK novita-gpus é https://async-public.serverless.novita.ai/v1. Para chamar um Endpoint, defina sua API Key e crie um cliente com o nome do 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. Exemplo de Job do ComfyUI

Para novitalabs/comfyui-worker:v0.0.1, a entrada do job deve incluir um workflow do ComfyUI. O exemplo mínimo a seguir corresponde ao caso testado:

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 Personalizado

No lado do worker, chame novita_gpus.start({"handler": handler}) para iniciar o loop de tarefas. A plataforma passa os dados da tarefa para handler(job):

job["id"]: ID do job atual
job["input"]: conteúdo de entrada enviado pelo cliente
O valor de retorno do handler é usado como saída do job
Se o dict retornado contiver um campo error, o job será marcado como falho

Exemplo mínimo de 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. Exemplos Completos de Worker

Para ver o código-fonte completo do worker, instruções de build da imagem Docker e scripts de envio de tarefas, consulte os exemplos Python de GPUs da Novita. O repositório inclui dois exemplos:

comfyui-worker: executa o ComfyUI a partir de handler.py e retorna imagens geradas.
sleep-worker: um handler.py mínimo que aguarda uma duração solicitada e retorna um resultado JSON.

Cada diretório de exemplo contém:

handler.py: define handler(job) e inicia o worker com novita_gpus.start({"handler": handler}).
Dockerfile: cria a imagem do worker.
requirements.txt: instala novita-gpus.
submit_task.py: envia uma tarefa com o SDK de cliente novita-gpus.

Faça o build e envie a imagem do seu worker para o seu próprio registry:

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

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

O envio de tarefas usa o formato de nome do Endpoint <endpoint-id>-<app-name>. Por exemplo, o Endpoint ID o8UJWkag5WTn e o nome do app async produzem:

o8UJWkag5WTn-async

Os exemplos submit_task.py aceitam o Endpoint ID e o nome do app separadamente e, em seguida, compõem o nome final do 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"

Dentro do script, o cliente do SDK é criado com o nome do Endpoint composto:

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

6. Retornar Imagens ou Arquivos

A API status do Endpoint Serverless Assíncrono tem um limite de tamanho de saída. Para arquivos grandes, como imagens e vídeos, faça primeiro o upload deles para um armazenamento de objetos e retorne URLs na saída. Configure as variáveis de ambiente de armazenamento de objetos no 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>

Faça upload de uma imagem no 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})

Você também pode fazer upload de arquivos comuns ou 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

Preciso configurar uma API Key no worker?

Geralmente, não. API Keys são usadas principalmente por clientes para enviar, consultar e cancelar jobs.

Qual valor de retorno do handler marca um job como falho?

Se o handler retornar um dict que contenha um campo error, o job será marcado como falho:

return {"error": "invalid input"}

Get started

Model APIs

Agent Sandbox

GPUs

Observability

Resources

Guia de Uso do SDK de GPUs

1. Instalar o SDK

2. Enviar Jobs com o SDK

3. Exemplo de Job do ComfyUI

4. Handler Personalizado

5. Exemplos Completos de Worker

6. Retornar Imagens ou Arquivos

7. FAQ

Preciso configurar uma API Key no worker?

Qual valor de retorno do handler marca um job como falho?

​1. Instalar o SDK

​2. Enviar Jobs com o SDK

​3. Exemplo de Job do ComfyUI

​4. Handler Personalizado

​5. Exemplos Completos de Worker

​6. Retornar Imagens ou Arquivos

​7. FAQ

​Preciso configurar uma API Key no worker?

​Qual valor de retorno do handler marca um job como falho?

1. Instalar o SDK

2. Enviar Jobs com o SDK

3. Exemplo de Job do ComfyUI

4. Handler Personalizado

5. Exemplos Completos de Worker

6. Retornar Imagens ou Arquivos

7. FAQ

Preciso configurar uma API Key no worker?

Qual valor de retorno do handler marca um job como falho?