Come chiamare flux-1-kontext-max su HiAPI: tutorial con curl e Python

Che aspetto ha una richiesta a flux-1-kontext-max

Prima di scrivere qualsiasi codice, è utile capire la forma dell'interazione. Su HiAPI, una chiamata di modifica immagine a flux-1-kontext-max non è una semplice richiesta/risposta che restituisce l'immagine nello stesso invocazione HTTP. È un'attività in due fasi:

1. Invio —— fai una POST di un payload JSON che descrive la tua immagine di input e la modifica desiderata. L'API accetta l'attività e restituisce un identificatore dell'attività.
2. Polling —— con quell'identificatore fai una GET dello stato dell'attività finché non segnala il completamento, poi leggi l'URL di output dalla risposta.

Questo schema è adottato perché la modifica delle immagini è ad alta intensità di calcolo. Mantenere aperta una singola connessione HTTP per tutta la durata del rendering è fragile; un handle di attività ti permette di disconnetterti, riprovare e tornare più tardi.

Il payload stesso si divide concettualmente in tre parti:

• Selettore del modello —— un campo che indica flux-1-kontext-max come modello da eseguire.
• Input —— di solito un'immagine (URL o payload base64, a seconda dello schema documentato) più un'istruzione testuale che descrive la modifica.
• Controlli opzionali —— parametri che regolano il comportamento della generazione, come il rapporto d'aspetto o il seed, se indicati nella documentazione del modello.

Tutto il resto —— header di autenticazione, endpoint di invio, endpoint di stato —— è la stessa pipeline che usi per ogni attività HiAPI. Una volta interiorizzato il ritmo in tre passi (invio → polling → download), le versioni curl e Python sono intercambiabili e puoi passare dall'una all'altra senza dover reimparare il flusso.

Prerequisiti e autenticazione

Prima che la prima richiesta abbia successo, devi preparare alcune cose.

Di cosa hai bisogno

• Un account HiAPI e una API key creata dalla dashboard.
• La base URL dell'API HiAPI, come indicato nella documentazione della piattaforma.
• Un'immagine di origine raggiungibile dall'API —— il più delle volte un URL HTTPS pubblico. Se la tua immagine è privata, caricala dove l'API può recuperarla, oppure passala inline se lo schema del modello supporta input base64.
• curl per l'esercitazione da shell, e Python 3 con la libreria requests per l'esercitazione lato codice.

Autenticazione

HiAPI autentica le richieste inviando la tua API key nell'header di autorizzazione. La convenzione è il bearer token:

Authorization: Bearer $HIAPI_API_KEY

E ogni richiesta JSON dovrebbe dichiarare il proprio tipo di contenuto:

Content-Type: application/json

Tieni la chiave lontana dal codice sorgente

Conserva la chiave in una variabile d'ambiente invece di incollarla nello script:

export HIAPI_API_KEY="your_key_here"

In Python, leggila con os.environ["HIAPI_API_KEY"]. In curl, referenziala come $HIAPI_API_KEY. Questa sola abitudine previene la causa più comune di fuga di credenziali —— una chiave committata nel repository e poi costretta a rotazione sotto pressione.

Se una richiesta restituisce un errore di autenticazione, controlla tre cose in ordine: la chiave esiste ed è valida, l'header è scritto esattamente come nella documentazione, e nella variabile d'ambiente non si sono infiltrati spazi o newline superflui.

Inviare la richiesta con curl

Di seguito è riportata la chiamata di invio. Sostituisci con il tuo URL immagine e la tua istruzione di modifica.

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

Il ruolo di ogni parte

• -X POST —— l'invio è un'operazione di creazione, quindi si usa POST.
• Endpoint —— l'endpoint jobs accetta nuove attività. La base URL viene dal tuo ambiente, così il comando è portabile tra ambienti diversi.
• Authorization —— il tuo bearer token, letto da una variabile d'ambiente invece che inline.
• Content-Type: application/json —— obbligatorio; senza di esso il server può rifiutare o interpretare male il corpo della richiesta.
• -d —— il payload JSON. model seleziona flux-1-kontext-max, input.image punta all'immagine di origine, input.prompt descrive la modifica.

Leggere la risposta

Un invio riuscito restituisce un corpo JSON contenente l'identificatore dell'attività —— di solito un campo come id o job_id —— e uno stato che indica che l'attività è in coda o in elaborazione. Salva quell'identificatore; il polling dipende da esso.

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

Passare l'output attraverso python -m json.tool stampa la risposta in modo leggibile, rendendo facile individuare l'ID dell'attività. Mettere il corpo della richiesta in payload.json mantiene il comando leggibile anche quando il prompt si allunga.

A questo punto hai inviato l'attività ma non hai ancora ricevuto l'immagine. L'immagine arriva solo dopo il polling.

La stessa richiesta in Python

La versione Python porta la stessa logica, con un'ergonomia più confortevole per tutto ciò che pensi di conservare nel tempo.

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

Perché questo rispecchia la chiamata curl

Il parametro json=payload serializza il dizionario e imposta il tipo di contenuto per te —— i byte inviati sono gli stessi di curl. timeout=30 limita la richiesta di invio, facendo fallire rumorosamente le connessioni appese invece di bloccare il tuo processo. raise_for_status() trasforma una 4xx o 5xx in un'eccezione che puoi catturare, invece di restituire silenziosamente un documento di errore.

Piccole differenze importanti

• Dizionario invece di stringa —— costruisci il payload come oggetto Python, il che rende facile calcolare prompt o iniettare variabili.
• Gestione delle eccezioni —— avvolgi la chiamata in try/except requests.HTTPError per registrare il corpo della risposta; i payload di errore di solito spiegano cosa non va nell'input.
• Nessun escaping della shell —— i prompt lunghi con virgolette non richiedono più escaping, il che elimina un'intera classe di errori.

L'ID dell'attività restituito qui va direttamente al passo di polling seguente. L'attività non cambia in alcun modo per il fatto di essere stata inviata da Python invece che da curl.

Gestire le attività asincrone e fare polling dei risultati

L'invio ritorna rapidamente; il rendering no. Aspetti facendo polling dell'endpoint di stato con l'ID dell'attività.

Ciclo di polling

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

Interpretare gli stati dell'attività

Prevedi un piccolo vocabolario: uno stato queued/pending prima che il lavoro inizi, uno stato processing/running durante il rendering del modello, e uno stato terminale —— successo o fallimento. Alcune API usano grafie leggermente diverse; controlla la risposta che ricevi effettivamente e mappa quelle stringhe, non dare per scontato. Regola chiave: considera l'attività completata solo quando lo stato è terminale di successo e l'URL di output è presente. Un 200 dall'endpoint di stato non significa che l'immagine sia pronta.

Buone maniere nel polling

• Distanzia le richieste. Per le attività d'immagine bastano pochi secondi tra un polling e l'altro. Fare polling ogni 100 ms spreca rate limit senza alcun beneficio.
• Limita l'attesa. max_attempts limita il ciclo, facendo fallire il tuo codice su un'attività bloccata invece di appenderlo per sempre.
• Fai backoff sui 429. Se tocchi il rate limit, dormi più a lungo prima di riprovare invece di martellare.
• Preferisci i webhook se disponibili. Se la documentazione della piattaforma descrive un meccanismo di callback, questo elimina completamente il polling ed è più amichevole su larga scala.

Per i test da shell, un ciclo veloce funziona allo stesso modo:

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 volta uscito dal ciclo, la risposta terminale contiene il riferimento all'output. È il passo successivo.

Scaricare e salvare l'output

La risposta di un'attività completata punta all'immagine renderizzata, di solito un URL (a volte una lista). Recuperala con una seconda richiesta e scrivi i byte su 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

Aspetti da tenere a mente

• Non codificare la forma. Il campo di output varia da modello a modello —— a volte è una stringa URL, a volte un oggetto con chiave url, a volte una lista. Controlla la risposta reale e normalizza prima del download.
• Gli URL di output possono scadere. Tratta l'URL come di breve durata: scarica subito, oppure copia immediatamente nel tuo storage. Non conservare l'URL come record permanente dell'immagine.
• Verifica i byte. Controlla che il Content-Type inizi con image/ e che la dimensione del file sia ragionevole; una pagina di errore salvata come .png è un fallimento silenzioso.
• Riprova gli errori di download transitori. Avvolgi la GET in un retry con backoff modesto —— gli URL firmati a volte restituiscono 403 per deriva dell'orologio, e la rete oscilla.

Per un rapido controllo manuale, curl è sufficiente:

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

Il flag -L segue i redirect, che gli output ospitati su CDN usano spesso. Una volta che il file è su disco, aprilo e conferma che la modifica corrisponda davvero al prompt —— questa è la tua verifica end-to-end, che in una pipeline in cui le immagini devono essere rigenerate in caso di fallimento vale la pena automatizzare con un controllo visivo o di hash.

Punti di fallimento comuni e correzioni

La maggior parte dei fallimenti di prima integrazione si verifica in uno di pochi modi.

Autenticazione e header

• 401/403 —— chiave errata, scaduta o con spazi superflui. Riesporta la variabile d'ambiente e conferma che la forma dell'header corrisponda alla documentazione.
• Content-Type mancante —— inviare JSON senza quell'header può produrre confusi errori 4xx. Includilo, oppure usa json= in Python.

Problemi di input

• Immagine non raggiungibile —— l'API recupera la tua immagine lato server. URL localhost o link che richiedono autenticazione falliranno. Usa un URL HTTPS pubblico, oppure base64 inline quando supportato.
• Prompt ignorato o rifiutato —— istruzioni vaghe producono risultati vaghi. Descrivi cosa cambiare e cosa mantenere, come nell'esempio del tramonto. I prompt molto lunghi possono essere troncati.
• Parametri non supportati —— passare campi che il modello non accetta può causare errori di validazione. Parti da un payload minimo, poi aggiungi i controlli documentati uno alla volta.

Errori di elaborazione asincrona

• Leggere l'immagine dalla risposta di invio —— la chiamata di invio restituisce un handle dell'attività, non i pixel. Fai prima polling.
• Considerare completato qualsiasi 200 —— l'endpoint di stato restituisce 200 anche per attività ancora in esecuzione. Controlla il campo di stato e la presenza dell'URL di output.
• Polling troppo veloce o all'infinito —— aggiungi intervallo e un tetto, altrimenti esaurirai il rate limit e bloccherai il tuo worker.

Problemi di download

• Salvare HTML come immagine —— controlla sempre il tipo di contenuto e la dimensione.
• URL di output scaduto —— scarica immediatamente dopo il completamento; non mettere l'URL in coda per dopo.

Errori strutturali

• ID attività non corrispondenti —— se invii più attività, etichetta ogni ID con il tuo ID richiesta, così i risultati del polling non si incrociano.
• Nessun timeout —— ogni richiesta dovrebbe averne uno. Senza, una chiamata appesa può bloccare un intero lotto.

Trasformare l'esercitazione in una funzione helper riutilizzabile

I tre passi —— invio, polling, download —— appartengono a un'unica funzione, così chi chiama non deve ripetere l'orchestrazione.

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

Cosa la rende pronta per la produzione

• Un unico punto d'ingresso. Chi chiama passa immagine e prompt e riceve un percorso di file o un'eccezione. I dettagli del flusso restano nascosti.
• Fallimenti espliciti. Ogni richiesta lancia su errore HTTP, il poller lancia su fallimento o timeout dell'attività, così la tua coda o chi chiama può riprovare.
• Tempistiche configurabili. poll_interval e max_attempts sono parametri, quindi puoi allentare i limiti per modifiche più pesanti senza toccare la logica.
• Nomi di output prevedibili. Usare l'ID dell'attività come nome del file rende facile correlare file e log.

Dove andare dopo

Avvolgi la chiamata con preoccupazioni a livello applicativo: registra ID attività e latenza, riprova gli invii falliti con jitter, e invia i byte scaricati direttamente all'object storage invece che al disco locale. Se gestisci molte modifiche, invia prima un lotto di attività e poi fai polling insieme —— l'invio è economico, la concorrenza risiede nell'attesa, non nel caricamento.

La pagina di flux-1-kontext-max nella documentazione HiAPI è il riferimento per i campi accettati e i parametri correnti. Mantieni l'helper snello e lascia che la documentazione guidi lo schema: la stessa funzione può coprire sia script esplorativi che pipeline reali.