Comment appeler flux-1-kontext-max sur HiAPI : tutoriel curl et Python

À quoi ressemble une requête flux-1-kontext-max

Avant d'écrire la moindre ligne de code, il est utile de comprendre la forme de l'interaction. Sur HiAPI, un appel d'édition d'image vers flux-1-kontext-max n'est pas une simple requête/réponse qui renvoie l'image dans le même appel HTTP. C'est une tâche en deux phases :

1. Soumission — vous envoyez en POST une charge utile JSON décrivant votre image d'entrée et l'édition souhaitée. L'API accepte la tâche et renvoie un identifiant de tâche.
2. Interrogation — vous envoyez une requête GET sur l'état de la tâche avec cet identifiant jusqu'à ce qu'elle signale son achèvement, puis vous lisez l'URL de sortie dans la réponse.

Ce mode est utilisé parce que l'édition d'image est gourmande en calcul. Maintenir une seule connexion HTTP ouverte pendant tout le rendu est fragile ; un handle de tâche vous permet de vous déconnecter, de réessayer et de revenir plus tard.

La charge utile se divise conceptuellement en trois parties :

• Sélecteur de modèle — un champ qui désigne flux-1-kontext-max comme le modèle à exécuter.
• Entrée — généralement une image (URL ou charge utile base64, selon le schéma documenté) accompagnée d'une instruction textuelle décrivant l'édition.
• Contrôles optionnels — des paramètres qui ajustent le comportement de génération, comme le format d'image ou la graine, à condition que la documentation du modèle les mentionne.

Tout le reste — en-têtes d'authentification, endpoint de soumission, endpoint d'état — est le même pipeline que celui utilisé pour chaque tâche HiAPI. Une fois ce rythme en trois étapes intériorisé (soumettre → interroger → télécharger), les versions curl et Python deviennent interchangeables et vous pouvez passer de l'une à l'autre sans réapprendre le flux.

Prérequis et authentification

Avant que la première requête n'aboutisse, plusieurs choses doivent être en place.

Ce dont vous avez besoin

• Un compte HiAPI et une clé API créée depuis le tableau de bord.
• L'URL de base de l'API HiAPI, telle que documentée par la plateforme.
• Une image source accessible par l'API — le plus souvent une URL HTTPS publique. Si votre image est privée, téléversez-la là où l'API peut la récupérer, ou transmettez-la en ligne si le schéma du modèle accepte l'entrée base64.
• curl pour l'exercice en shell, ainsi que Python 3 et la bibliothèque requests pour l'exercice en code.

Authentification

HiAPI authentifie les requêtes en envoyant votre clé API dans l'en-tête d'autorisation. La convention est le jeton bearer :

Authorization: Bearer $HIAPI_API_KEY

et chaque requête JSON doit déclarer son type de contenu :

Content-Type: application/json

Garder la clé hors du code source

Stockez la clé dans une variable d'environnement plutôt que de la coller dans un script :

export HIAPI_API_KEY="your_key_here"

En Python, lisez-la avec os.environ["HIAPI_API_KEY"]. Dans curl, référencez-la par $HIAPI_API_KEY. Cette seule habitude prévient la cause la plus courante de fuite d'identifiants — une clé commitée dans un dépôt, puis contrainte d'être renouvelée dans l'urgence.

Si une requête renvoie une erreur d'authentification, vérifiez trois choses dans l'ordre : la clé existe et est valide, l'orthographe de l'en-tête correspond exactement à la documentation, et aucune espace ou saut de ligne superflu ne s'est glissé dans la variable d'environnement.

Envoyer la requête avec curl

Voici l'appel de soumission. Remplacez l'URL d'image et l'instruction d'édition par les vôtres.

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"
    }
  }'

Le rôle de chaque partie

• -X POST — la soumission est une opération de création, donc on utilise POST.
• Endpoint — l'endpoint jobs accepte les nouvelles tâches. L'URL de base provient de votre environnement, ce qui rend la commande portable entre environnements.
• Authorization — votre jeton bearer, lu depuis une variable d'environnement plutôt qu'en ligne.
• Content-Type: application/json — obligatoire ; sans lui, le serveur peut rejeter ou mal interpréter le corps de la requête.
• -d — la charge utile JSON. model sélectionne flux-1-kontext-max, input.image pointe vers votre image source, input.prompt décrit l'édition.

Lire la réponse

Une soumission réussie renvoie un corps JSON contenant l'identifiant de tâche — généralement un champ comme id ou job_id — ainsi qu'un état indiquant que la tâche est en file d'attente ou en cours de traitement. Conservez cet identifiant ; l'interrogation en dépend.

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

Le passage par python -m json.tool embellit la réponse et rend l'ID de tâche facile à repérer. Placer le corps de la requête dans payload.json garde la commande lisible à mesure que le prompt s'allonge.

À ce stade, vous avez soumis la tâche mais n'avez pas encore reçu l'image. L'image n'arrive qu'après l'interrogation.

La même requête en Python

La version Python porte la même logique, avec une ergonomie plus confortable pour tout ce que vous prévoyez de conserver à long terme.

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"])

Pourquoi cela reflète exactement l'appel curl

Le paramètre json=payload sérialise le dictionnaire et définit le type de contenu pour vous — les octets envoyés sont identiques à ceux de curl. timeout=30 limite la requête de soumission, de sorte qu'une connexion suspendue échoue bruyamment au lieu de bloquer votre processus. raise_for_status() transforme un 4xx ou 5xx en exception que vous pouvez intercepter, plutôt qu'en document d'erreur renvoyé silencieusement.

Petites différences importantes

• Dictionnaire au lieu de chaîne — vous construisez la charge utile comme un objet Python, ce qui facilite le calcul de prompts ou l'injection de variables.
• Gestion des exceptions — enveloppez l'appel dans try/except requests.HTTPError pour journaliser le corps de la réponse ; les charges utiles d'erreur expliquent souvent ce qui cloche dans l'entrée.
• Aucun échappement shell — les longs prompts contenant des guillemets ne nécessitent plus d'échappement, ce qui élimine toute une classe d'erreurs.

L'ID de tâche renvoyé ici alimente directement l'étape d'interrogation ci-dessous. La tâche ne change en rien du fait qu'elle est soumise depuis Python plutôt que curl.

Gérer les tâches asynchrones et interroger le résultat

La soumission revient vite ; le rendu, non. Vous attendez en interrogeant l'endpoint d'état de l'ID de tâche.

Boucle d'interrogation

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")

Interpréter l'état de la tâche

Attendez-vous à un petit vocabulaire : un état queued/pending avant le début du travail, un état processing/running pendant le rendu par le modèle, et un état terminal — succès ou échec. Certaines API utilisent des orthographes légèrement différentes ; examinez la réponse que vous recevez réellement et mappez ces chaînes, sans supposer. La règle essentielle : ne considérez une tâche comme terminée que si l'état est un état final de succès et qu'une URL de sortie est présente. Un 200 renvoyé par l'endpoint d'état ne signifie pas que l'image est prête.

Bonnes manières d'interrogation

• Espacez vos requêtes. Pour les tâches d'image, quelques secondes entre chaque interrogation suffisent. Interroger toutes les 100 ms gaspille les limites de débit sans aucun bénéfice.
• Bornez l'attente. max_attempts limite la boucle, de sorte qu'une tâche bloquée fait échouer votre code au lieu de le suspendre indéfiniment.
• Reculez sur 429. Si vous atteignez la limite de débit, dormez plus longtemps avant de réessayer plutôt que d'insister.
• Préférez un webhook s'il est proposé. Si la documentation de la plateforme décrit un mécanisme de callback, il supprime entièrement l'interrogation et est plus adapté à l'échelle.

Pour les tests en shell, une boucle rapide fonctionne de la même manière :

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

Une fois sorti de la boucle, la réponse terminale contient la référence de sortie. C'est l'étape suivante.

Télécharger et enregistrer la sortie

La réponse d'une tâche terminée pointe vers l'image rendue, généralement une URL (parfois une liste). Effectuez une seconde requête pour la récupérer et écrivez les octets sur le disque.

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

Points d'attention

• Ne codez pas la forme en dur. Le champ de sortie varie selon les modèles — tantôt une URL sous forme de chaîne, tantôt un objet avec une clé url, tantôt une liste. Examinez la réponse réelle et normalisez avant de télécharger.
• Les URL de sortie peuvent expirer. Traitez l'URL comme de courte durée de vie : téléchargez immédiatement, ou copiez sans tarder vers votre propre stockage. Ne stockez pas l'URL comme enregistrement permanent de l'image.
• Validez les octets. Vérifiez que le Content-Type commence par image/ et que la taille du fichier est raisonnable ; une page d'erreur enregistrée en .png est un échec silencieux.
• Réessayez les erreurs de téléchargement transitoires. Enveloppez le GET dans une nouvelle tentative avec un petit recul — les URL signées renvoient parfois un 403 à cause d'une dérive d'horloge, et le réseau vacille.

Pour une vérification manuelle rapide, curl suffit :

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

L'option -L suit les redirections, que les sorties hébergées sur CDN utilisent souvent. Une fois le fichier écrit, ouvrez-le et confirmez que l'édition correspond bien au prompt — c'est votre validation de bout en bout, qui mérite d'être automatisée par une vérification visuelle ou de hash dans un pipeline où l'image doit être régénérée en cas d'échec.

Points d'échec courants et corrections

La plupart des échecs d'intégration initiale se produisent de l'une de ces quelques manières.

Authentification et en-têtes

• 401/403 — clé erronée, expirée ou contenant des espaces superflus. Réexportez la variable d'environnement et confirmez que la forme de l'en-tête correspond à la documentation.
• Content-Type manquant — envoyer du JSON sans cet en-tête peut produire des erreurs 4xx déroutantes. Incluez-le, ou utilisez json= en Python.

Problèmes d'entrée

• Image inaccessible — l'API récupère votre image côté serveur. Une URL localhost ou un lien nécessitant une authentification échouera. Utilisez une URL HTTPS publique, ou du base64 en ligne si c'est pris en charge.
• Prompt ignoré ou rejeté — des instructions vagues produisent des résultats vagues. Décrivez ce qui doit changer et ce qui doit rester inchangé, comme dans l'exemple du coucher de soleil. Les prompts très longs peuvent être tronqués.
• Paramètres non pris en charge — transmettre des champs que le modèle n'accepte pas peut provoquer des erreurs de validation. Commencez par une charge utile minimale, puis ajoutez les contrôles documentés un par un.

Erreurs de traitement asynchrone

• Lire l'image dans la réponse de soumission — l'appel de soumission renvoie un handle de tâche, pas des pixels. Interrogez d'abord.
• Considérer tout 200 comme un achèvement — l'endpoint d'état renvoie aussi 200 pour une tâche encore en cours. Vérifiez le champ d'état et la présence de l'URL de sortie.
• Interroger trop vite ou indéfiniment — ajoutez un intervalle et un plafond, sinon vous épuiserez la limite de débit et suspendrez votre worker.

Problèmes de téléchargement

• Enregistrer du HTML comme image — vérifiez toujours le type de contenu et la taille.
• URL de sortie expirée — téléchargez immédiatement après l'achèvement ; ne mettez pas l'URL en file d'attente pour plus tard.

Erreurs structurelles

• Incohérence d'ID de tâche — si vous soumettez plusieurs tâches, étiquetez chaque ID avec votre propre ID de requête, afin que les résultats d'interrogation ne se mélangent pas.
• Absence de timeout — chaque requête devrait en avoir un. Sans cela, un appel suspendu peut bloquer tout un lot.

Transformer l'exercice en fonction utilitaire réutilisable

Les trois étapes — soumettre, interroger, télécharger — appartiennent à une seule fonction, afin que l'appelant n'ait pas à répéter l'orchestration.

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")

Ce qui la rend prête pour la production

• Point d'entrée unique. L'appelant fournit une image et un prompt, et obtient un chemin de fichier ou une exception. Les détails du flux de travail restent cachés.
• Échec explicite. Chaque requête lève une exception en cas d'erreur HTTP, et l'interrogateur lève une exception en cas d'échec ou de dépassement de délai de la tâche, ce qui permet à votre file d'attente ou à votre appelant de réessayer.
• Timing configurable. poll_interval et max_attempts sont des paramètres, si bien que vous pouvez les assouplir pour des éditions plus lourdes sans toucher à la logique.
• Nommage de sortie prévisible. Utiliser l'ID de tâche comme nom de fichier rend la corrélation entre fichiers et journaux facile.

Où aller ensuite

Enveloppez l'appel dans des préoccupations applicatives : journalisez l'ID de tâche et la latence, réessayez les soumissions échouées avec un peu de gigue, et poussez les octets téléchargés directement vers un stockage objet plutôt que sur le disque local. Si vous traitez beaucoup d'éditions, soumettez d'abord un lot de tâches, puis interrogez-les ensemble — la soumission est bon marché, la concurrence réside dans l'attente, pas dans l'envoi.

La page flux-1-kontext-max de la documentation HiAPI est la référence pour les champs acceptés et les paramètres actuels. Gardez la fonction utilitaire concise, laissez la documentation dicter le schéma, et la même fonction couvrira à la fois les scripts exploratoires et les pipelines réels.