Cómo llamar a flux-1-kontext-max en HiAPI: tutorial con curl y Python

Cómo es una solicitud a flux-1-kontext-max

Antes de escribir cualquier código, resulta útil conocer la forma de la interacción. En HiAPI, una llamada de edición de imágenes a flux-1-kontext-max no es una solicitud/respuesta simple que devuelve la imagen en la misma llamada HTTP. Es una tarea de dos fases:

1. Envío —— haces un POST con una carga útil JSON que describe tu imagen de entrada y la edición deseada. La API acepta la tarea y devuelve un identificador de tarea.
2. Sondeo —— haces un GET del estado de la tarea con ese identificador hasta que informe que ha terminado, y luego lees la URL de salida en la respuesta.

Este patrón se adopta porque la edición de imágenes consume muchos recursos. Mantener abierta una única conexión HTTP durante todo el renderizado es frágil; un identificador de tarea te permite desconectarte, reintentar y volver más tarde.

La carga útil en sí se divide conceptualmente en tres partes:

• Selector de modelo —— un campo que indica flux-1-kontext-max como el modelo a ejecutar.
• Entrada —— normalmente una imagen (URL o carga útil base64, según el esquema documentado) más una instrucción de texto que describe la edición.
• Controles opcionales —— parámetros que ajustan el comportamiento de la generación, como la relación de aspecto o la semilla, siempre que la documentación del modelo los describa.

Todo lo demás —— las cabeceras de autenticación, el endpoint de envío, el endpoint de estado —— es la misma tubería que usas en cada tarea de HiAPI. Una vez que interiorizas este ritmo de tres pasos (enviar → sondear → descargar), las versiones curl y Python son intercambiables, y puedes alternar entre ambas sin volver a aprender el flujo.

Requisitos previos y autenticación

Antes de que la primera solicitud tenga éxito, necesitas tener varias cosas listas.

Qué necesitas

• Una cuenta de HiAPI y una API key creada desde el panel de control.
• La URL base de la API de HiAPI, tal como se indica en la documentación de la plataforma.
• Una imagen de origen accesible para la API —— lo más común es una URL HTTPS pública. Si tu imagen es privada, súbela a un lugar al que la API pueda acceder o, si el esquema del modelo admite entrada base64, pásala en línea.
• curl para la demostración en shell, y Python 3 con la biblioteca requests para la demostración en código.

Autenticación

HiAPI autentica las solicitudes enviando tu API key en la cabecera de autorización. La convención es un token bearer:

Authorization: Bearer $HIAPI_API_KEY

Y toda solicitud JSON debe declarar su tipo de contenido:

Content-Type: application/json

Mantén la clave fuera del código fuente

Guarda la clave en una variable de entorno en lugar de pegarla en el script:

export HIAPI_API_KEY="your_key_here"

En Python, léela con os.environ["HIAPI_API_KEY"]. En curl, referénciala como $HIAPI_API_KEY. Este único hábito previene la causa más común de filtración de credenciales: una clave comprometida en un repositorio y luego rotada bajo presión.

Si una solicitud devuelve un error de autenticación, revisa tres cosas en orden: que la clave exista y sea válida, que la cabecera esté escrita exactamente como en la documentación, y que no se hayan colado espacios o saltos de línea extra en la variable de entorno.

Enviar la solicitud con curl

Aquí está la llamada de envío. Sustituye la URL de tu imagen y la instrucción de edición por las tuyas.

curl -X POST "$HIAPI_BASE_URL/jobs" \
  -H "Authorization: Bearer $HIAPI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "flux-1-kontext-max",
    "input": {
      "image": "https://example.com/input.jpg",
      "prompt": "Replace the sky with a dramatic sunset, keep the buildings unchanged"
    }
  }'

Qué hace cada parte

• -X POST —— el envío es una operación de creación, así que usa POST.
• Endpoint —— el endpoint de jobs acepta nuevas tareas. La URL base viene de tu entorno, así que el comando es portable entre entornos.
• Authorization —— tu token bearer, leído de la variable de entorno en lugar de estar en línea.
• Content-Type: application/json —— obligatorio; sin él, el servidor puede rechazar o analizar mal el cuerpo de la solicitud.
• -d —— la carga útil JSON. model selecciona flux-1-kontext-max, input.image apunta a tu imagen de origen, y input.prompt describe la edición.

Leer la respuesta

Un envío exitoso devuelve un cuerpo JSON que contiene el identificador de tarea —— normalmente un campo como id o job_id —— y un estado que indica que la tarea está en cola o en procesamiento. Guarda ese identificador; el sondeo depende de él.

curl -s -X POST "$HIAPI_BASE_URL/jobs" \
  -H "Authorization: Bearer $HIAPI_API_KEY" \
  -H "Content-Type: application/json" \
  -d @payload.json | python -m json.tool

Canalizar la salida por python -m json.tool imprime la respuesta con formato, lo que facilita encontrar el ID de la tarea. Colocar el cuerpo de la solicitud en payload.json mantiene el comando legible a medida que el prompt se alarga.

En este punto has enviado la tarea, pero aún no has recibido la imagen. La imagen solo se devuelve después del sondeo.

La misma solicitud en Python

La versión en Python conlleva la misma lógica y, para cualquier cosa que planees conservar a largo plazo, ofrece una ergonomía más cómoda.

import os
import requests

BASE_URL = os.environ["HIAPI_BASE_URL"]
API_KEY = os.environ["HIAPI_API_KEY"]

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

payload = {
    "model": "flux-1-kontext-max",
    "input": {
        "image": "https://example.com/input.jpg",
        "prompt": "Replace the sky with a dramatic sunset, keep the buildings unchanged",
    },
}

response = requests.post(
    f"{BASE_URL}/jobs",
    headers=headers,
    json=payload,
    timeout=30,
)
response.raise_for_status()
job = response.json()
print("job id:", job["id"])

Por qué esto refleja la llamada con curl

El parámetro json=payload serializa el diccionario y establece el tipo de contenido por ti —— los bytes enviados son los mismos que con curl. timeout=30 limita la solicitud de envío, de modo que una conexión colgada falla de forma ruidosa en lugar de bloquear tu proceso. raise_for_status() convierte un 4xx o 5xx en una excepción que puedes capturar, en lugar de devolver silenciosamente un documento de error.

Pequeñas diferencias importantes

• Diccionario en lugar de cadena —— construyes la carga útil como un objeto Python, lo que facilita calcular prompts o inyectar variables.
• Manejo de excepciones —— envuelve la llamada en try/except requests.HTTPError para registrar el cuerpo de la respuesta; las cargas útiles de error suelen explicar qué falló en la entrada.
• Sin necesidad de escape de shell —— los prompts largos entre comillas ya no necesitan escape, lo que elimina toda una clase de errores.

El ID de tarea devuelto aquí pasa directamente al paso de sondeo que sigue. La tarea no cambia en nada por haber sido enviada desde Python en lugar de curl.

Manejar tareas asíncronas y sondear resultados

El envío regresa rápido; el renderizado no. Esperas sondeando el endpoint de estado con el ID de tarea.

Bucle de sondeo

import time

def wait_for_job(job_id, headers, interval=3, max_attempts=100):
    url = f"{BASE_URL}/jobs/{job_id}"
    for attempt in range(max_attempts):
        r = requests.get(url, headers=headers, timeout=30)
        r.raise_for_status()
        data = r.json()
        status = data.get("status")
        if status == "succeeded" or status == "completed":
            return data
        if status == "failed" or status == "error":
            raise RuntimeError(f"Job failed: {data}")
        time.sleep(interval)
    raise TimeoutError(f"Job {job_id} did not finish in time")

Interpretar los estados de la tarea

Espera un vocabulario pequeño: un estado queued/pending antes de que empiece el trabajo, un estado processing/running durante el renderizado del modelo, y un estado terminal —— éxito o fallo. Algunas API usan grafías ligeramente distintas; examina la respuesta que realmente recibes y mapea esas cadenas, no lo des por supuesto. La regla clave: considera la tarea completada solo cuando el estado sea un estado terminal de éxito y exista una URL de salida. Que el endpoint de estado devuelva 200 no significa que la imagen esté lista.

Etiqueta de sondeo

• Espacia tus solicitudes. Unos pocos segundos entre sondeos es suficiente para tareas de imagen. Sondear cada 100 ms desperdicia límites de tasa sin ninguna ganancia.
• Limita el tiempo de espera. max_attempts acota el bucle, de modo que una tarea atascada hace fallar tu código en lugar de colgarse para siempre.
• Haz retroceso ante 429. Si alcanzas un límite de tasa, duerme más antes de reintentar en lugar de insistir.
• Prefiere webhooks si están disponibles. Si la documentación de la plataforma describe un mecanismo de callback, elimina el sondeo por completo y es mucho más amigable a escala.

Para pruebas en shell, un bucle rápido funciona de la misma manera:

JOB_ID="your_job_id"
while true; do
  BODY=$(curl -s "$HIAPI_BASE_URL/jobs/$JOB_ID" -H "Authorization: Bearer $HIAPI_API_KEY")
  echo "$BODY"
  echo "$BODY" | grep -q '"succeeded"' && break
  sleep 3
done

Una vez que sales del bucle, la respuesta terminal contiene la referencia de salida. Ese es el siguiente paso.

Descargar y guardar la salida

La respuesta de una tarea completada apunta a la imagen renderizada, normalmente una URL (a veces una lista). Haz una segunda solicitud para obtenerla y escribe los bytes en disco.

import pathlib

def download_output(job_data, path="edit.png"):
    out = job_data.get("output")
    if isinstance(out, list):
        out = out[0]
    url = out["url"] if isinstance(out, dict) else out

    r = requests.get(url, timeout=60)
    r.raise_for_status()
    pathlib.Path(path).write_bytes(r.content)
    return path

Puntos a tener en cuenta

• No codifiques la forma en duro. El campo de salida varía según el modelo —— a veces es una cadena URL, a veces un objeto con una clave url, a veces una lista. Inspecciona la respuesta real y normalízala antes de descargar.
• Las URL de salida pueden caducar. Trata la URL como de vida corta: descarga de inmediato o cópiala enseguida a tu propio almacenamiento. No guardes la URL como registro permanente de la imagen.
• Valida los bytes. Comprueba que Content-Type empiece por image/ y que el tamaño del archivo sea razonable; una página de error guardada como .png es un fallo silencioso.
• Reintenta errores transitorios de descarga. Envuelve el GET en reintentos con un pequeño retroceso —— las URL firmadas a veces devuelven 403 por desfase de reloj, y la red también fluctúa.

Para una comprobación manual rápida, curl es suficiente:

curl -L -o edit.png "$OUTPUT_URL"

El flag -L sigue redirecciones, que las salidas alojadas en CDN suelen usar con frecuencia. Una vez que el archivo está en disco, ábrelo y confirma que la edición coincide realmente con el prompt —— esta es tu verificación de extremo a extremo, y en tuberías donde la imagen debe regenerarse si falla, vale la pena automatizar una comprobación visual o de hash.

Puntos comunes de fallo y sus soluciones

La mayoría de los fallos de una primera integración se producen de una de unas pocas maneras.

Autenticación y cabeceras

• 401/403 —— clave incorrecta, caducada o con espacios extra. Vuelve a exportar la variable de entorno y confirma que la forma de la cabecera coincide con la documentación.
• Falta Content-Type —— enviar JSON sin esa cabecera puede producir errores 4xx confusos. Inclúyela, o usa json= en Python.

Problemas de entrada

• La imagen no es accesible —— la API obtiene tu imagen en el servidor. Una URL de localhost o un enlace que requiere autenticación fallará. Usa una URL HTTPS pública o, cuando se admita, base64 en línea.
• El prompt se ignora o se rechaza —— instrucciones vagas producen resultados vagos. Describe qué cambiar y qué conservar, como muestra el ejemplo del atardecer. Los prompts muy largos pueden truncarse.
• Parámetros no admitidos —— pasar campos que el modelo no acepta puede causar errores de validación. Empieza con la carga útil mínima y luego añade los controles documentados uno a uno.

Errores de procesamiento asíncrono

• Leer la imagen de la respuesta de envío —— la llamada de envío devuelve un identificador de tarea, no píxeles. Primero hay que sondear.
• Tratar cualquier 200 como completado —— el endpoint de estado también devuelve 200 para tareas aún en ejecución. Comprueba el campo de estado y que exista la URL de salida.
• Sondear demasiado rápido o para siempre —— añade un intervalo y un límite, o agotarás los límites de tasa y colgarás tu worker.

Problemas de descarga

• Guardar HTML como imagen —— comprueba siempre el tipo de contenido y el tamaño.
• URL de salida caducada —— descarga inmediatamente tras completar; no dejes la URL en cola para más tarde.

Errores estructurales

• Desajuste de ID de tarea —— si envías varias tareas, etiqueta cada ID con tu propio ID de solicitud para que los resultados del sondeo no se crucen.
• Sin timeout —— toda solicitud debería tener uno. Sin él, una llamada colgada puede bloquear un lote.

Convertir la demostración en una función auxiliar reutilizable

Los tres pasos —— enviar, sondear, descargar —— pertenecen a una sola función, de modo que quien la llame no tenga que repetir la orquestación.

def edit_image(image_url, prompt, model="flux-1-kontext-max",
               headers=None, poll_interval=3, max_attempts=100):
    payload = {"model": model,
               "input": {"image": image_url, "prompt": prompt}}
    r = requests.post(f"{BASE_URL}/jobs", headers=headers,
                      json=payload, timeout=30)
    r.raise_for_status()
    job_id = r.json()["id"]

    data = wait_for_job(job_id, headers,
                        interval=poll_interval,
                        max_attempts=max_attempts)
    return download_output(data, path=f"{job_id}.png")

Qué la hace apta para producción

• Un único punto de entrada. Quien la llama pasa una imagen y un prompt, y recibe una ruta de archivo o una excepción. Los detalles del flujo de trabajo quedan ocultos.
• Fallo explícito. Cada solicitud lanza error ante un fallo HTTP, y el sondeador lanza error cuando la tarea falla o se agota el tiempo, de modo que tu cola o quien llama pueden reintentar.
• Temporización configurable. poll_interval y max_attempts son parámetros, así que puedes relajarlos para ediciones más pesadas sin tocar la lógica.
• Nombres de salida predecibles. Usar el ID de tarea como nombre de archivo facilita correlacionar archivos con registros.

Hacia dónde ir después

Envuelve la llamada con preocupaciones de la capa de aplicación: registra los ID de tarea y la latencia, reintenta envíos fallidos con jitter y empuja los bytes descargados directamente al almacenamiento de objetos en lugar del disco local. Si gestionas muchas ediciones, envía un lote de tareas primero y luego sondea todas juntas —— el envío es barato; la concurrencia está en la espera, no en la subida.

La página de flux-1-kontext-max en la documentación de HiAPI es la referencia de los campos aceptados y los parámetros actuales. Mantén la función auxiliar ligera, deja que la documentación guíe el esquema, y la misma función cubrirá tanto scripts exploratorios como tuberías reales.