Guía de uso del SDK de GPUs

Esta guía explica cómo usar el SDK novita-gpus para llamar a un Async Serverless Endpoint y cómo personalizar un handler de worker.

1. Instalar el SDK

Instala el SDK en tu entorno de cliente o en el entorno de ejecución del worker:

pip install novita-gpus

2. Enviar trabajos con el SDK

La URL de solicitud predeterminada del SDK novita-gpus es https://async-public.serverless.novita.ai/v1. Para llamar a un Endpoint, configura tu clave de API y crea un cliente con el nombre del 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. Ejemplo de trabajo de ComfyUI

Para novitalabs/comfyui-worker:v0.0.1, la entrada del trabajo debe incluir un flujo de trabajo de ComfyUI. El siguiente ejemplo mínimo coincide con el caso probado:

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

En el lado del worker, llama a novita_gpus.start({"handler": handler}) para iniciar el bucle de tareas. La plataforma pasa los datos de la tarea a handler(job):

job["id"]: id del trabajo actual
job["input"]: contenido de entrada enviado por el cliente
El valor de retorno del handler se usa como salida del trabajo
Si el dict devuelto contiene un campo error, el trabajo se marca como fallido

Ejemplo 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. Ejemplos completos de workers

Para ver el código fuente completo de workers, instrucciones para compilar imágenes Docker y scripts de envío de tareas, consulta Novita GPUs Python examples. El repositorio incluye dos ejemplos:

comfyui-worker: ejecuta ComfyUI desde handler.py y devuelve las imágenes generadas.
sleep-worker: un handler.py mínimo que espera durante la duración solicitada y devuelve un resultado JSON.

Cada directorio de ejemplo contiene:

handler.py: define handler(job) e inicia el worker con novita_gpus.start({"handler": handler}).
Dockerfile: compila la imagen del worker.
requirements.txt: instala novita-gpus.
submit_task.py: envía una tarea con el SDK de cliente novita-gpus.

Compila y sube tu imagen de worker a tu propio registro:

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

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

El envío de tareas usa el formato de nombre de Endpoint <endpoint-id>-<app-name>. Por ejemplo, el ID de Endpoint o8UJWkag5WTn y el nombre de app async producen:

o8UJWkag5WTn-async

Los ejemplos de submit_task.py aceptan el ID de Endpoint y el nombre de app por separado, y luego componen el nombre final del 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 del script, el cliente del SDK se crea con el nombre de Endpoint compuesto:

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

6. Devolver imágenes o archivos

La API status de Async Serverless Endpoint tiene un límite de tamaño de salida. Para archivos grandes, como imágenes y videos, súbelos primero a un almacenamiento de objetos y devuelve URL en la salida. Configura las variables de entorno del almacenamiento de objetos en el 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>

Sube una imagen en el 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})

También puedes subir archivos normales o 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

¿Necesito configurar una clave de API en el worker?

Normalmente no. Las claves de API se usan principalmente por los clientes para enviar, consultar y cancelar trabajos.

¿Qué valor de retorno del handler marca un trabajo como fallido?

Si el handler devuelve un dict que contiene un campo error, el trabajo se marca como fallido:

return {"error": "invalid input"}

Get started

Model APIs

Agent Sandbox

GPUs

Observability

Resources

Guía de uso del SDK de GPUs

1. Instalar el SDK

2. Enviar trabajos con el SDK

3. Ejemplo de trabajo de ComfyUI

4. Handler personalizado

5. Ejemplos completos de workers

6. Devolver imágenes o archivos

7. FAQ

¿Necesito configurar una clave de API en el worker?

¿Qué valor de retorno del handler marca un trabajo como fallido?

​1. Instalar el SDK

​2. Enviar trabajos con el SDK

​3. Ejemplo de trabajo de ComfyUI

​4. Handler personalizado

​5. Ejemplos completos de workers

​6. Devolver imágenes o archivos

​7. FAQ

​¿Necesito configurar una clave de API en el worker?

​¿Qué valor de retorno del handler marca un trabajo como fallido?

1. Instalar el SDK

2. Enviar trabajos con el SDK

3. Ejemplo de trabajo de ComfyUI

4. Handler personalizado

5. Ejemplos completos de workers

6. Devolver imágenes o archivos

7. FAQ

¿Necesito configurar una clave de API en el worker?

¿Qué valor de retorno del handler marca un trabajo como fallido?