API Midjourney in Python: genera e salva immagini

Questo esempio chiama da Python l'API Midjourney di OmniaKey, un servizio di terze parti, in tre passaggi: creare l'attività, controllarne lo stato fino al completamento e scaricare il PNG dall'URL firmato dell'asset. Un test di integrazione del 12 novembre 2026 ha restituito un'immagine 2048x2048 in 197,1 secondi, e quell'immagine è una griglia 2x2 con quattro candidati. Lo script sotto aggiunge un file di stato, così un timeout o un download fallito non creano una seconda attività a pagamento.

Risposta breve

• Invia POST https://api.omniakey.com/v1/media/tasks con un Idempotency-Key, poi controlla GET /v1/media/tasks/{id} e scarica assets[0].url.
• Un'attività restituisce una griglia PNG con quattro immagini, non quattro file separati.
• Midjourney in questa API fa solo testo-immagine: niente modifica, immagine di riferimento, blend, upscale, vary, spostamento o selettore di versione.
• La chiave viene da OmniaKey ed è separata da un abbonamento Midjourney. Salva il corpo e la chiave di idempotenza prima del POST, aggiungi l'id dell'attività quando l'API accetta e conserva il file di stato per riprendere più tardi. Un timeout locale non annulla l'attività sul server.

Quale API usa questo esempio Python?

L'esempio chiama l'API di terze parti di OmniaKey per Midjourney. OmniaKey non è Midjourney, e una chiave OmniaKey è un account separato con saldo proprio. Un abbonamento sul sito Midjourney è un altro prodotto e non crea una chiave OmniaKey.

La pagina del modello Midjourney mostra il modello e la tariffa attuale lato OmniaKey. Il resto del percorso copre questa integrazione e il ciclo di vita delle sue attività.

Cosa serve

Servono tre cose:

1. Una chiave API OmniaKey, esportata come OMNIAKEY_API_KEY.
2. Python 3. Lo script usa solo la libreria standard, quindi non c'è un SDK da installare.
3. midjourney disponibile per quella chiave/account. Se il modello manca nell'elenco della chiave, controlla l'accesso dell'account prima di eseguire la chiamata di creazione.

export OMNIAKEY_API_KEY="your-key"
python3 midjourney_grid.py

Tieni la chiave fuori dallo script e dai log. Il codice la legge una volta dall'ambiente.

Corpo della richiesta e parametri

Midjourney riceve un prompt e tre parametri. Il tipo di bot è preimpostato lato OmniaKey, quindi il corpo può omettere la versione del modello e botType.

Definisci la forma con aspectRatio, non con una dimensione in pixel. Il campo è in camelCase: aspectRatio, mai aspect_ratio. Usa uno dei cinque valori elencati.

Il corpo per il prompt di esempio è così:

{
  "model": "midjourney",
  "type": "image",
  "operation": "generate",
  "input": {
    "prompt": "A small red paper boat floating on a quiet blue lake at sunrise, cinematic lighting, elegant composition, no text"
  },
  "parameters": {"aspectRatio": "1:1", "quality": "1", "n": 1}
}

Chiamare l'API Midjourney da Python

Salvalo come midjourney_grid.py ed eseguilo. La prima esecuzione scrive midjourney-task.json prima del POST, poi aggiunge l'id dell'attività quando OmniaKey accetta la richiesta.

import hashlib
import json
import os
import sys
import time
import urllib.error
import urllib.request
import uuid
from pathlib import Path

API = "https://api.omniakey.com/v1"
KEY = os.environ["OMNIAKEY_API_KEY"]
STATE = Path("midjourney-task.json")
OUT = Path("midjourney-grid.png")
PROMPT = ("A small red paper boat floating on a quiet blue lake at sunrise, "
          "cinematic lighting, elegant composition, no text")


def save(state):
    tmp = STATE.with_suffix(".tmp")
    tmp.write_text(json.dumps(state, indent=2))
    tmp.replace(STATE)


def read_json(response):
    raw = response.read()
    try:
        return json.loads(raw)
    except ValueError:
        return {"raw": raw[:300].decode("utf-8", "replace")}


def call(method, path, body=None, headers=None):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(
        API + path, data=data, method=method,
        headers={"Authorization": f"Bearer {KEY}", **(headers or {})})
    if body is not None:
        req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=60) as response:
            return response.status, read_json(response)
    except urllib.error.HTTPError as error:
        return error.code, read_json(error)
    except (urllib.error.URLError, TimeoutError) as error:
        return None, {"error": str(error)}


def main():
    state = json.loads(STATE.read_text()) if STATE.exists() else None
    if state is None:
        body = {
            "model": "midjourney", "type": "image", "operation": "generate",
            "input": {"prompt": PROMPT},
            "parameters": {"aspectRatio": "1:1", "quality": "1", "n": 1},
        }
        state = {"idempotency_key": str(uuid.uuid4()), "body": body,
                 "task_id": None, "status": "creating"}
        save(state)  # no API key here; write before POST
    elif state["body"]["input"]["prompt"] != PROMPT:
        sys.exit("The state file holds a different prompt. Move it aside to start a new job.")

    if OUT.exists() and state.get("status") == "downloaded":
        saved_sha = state.get("download_sha")
        if not saved_sha or hashlib.sha256(OUT.read_bytes()).hexdigest() == saved_sha:
            print("already saved", OUT)
            return

    if not state["task_id"]:
        code, task = call("POST", "/media/tasks", state["body"],
                          {"Idempotency-Key": state["idempotency_key"]})
        if code is None:
            sys.exit(f"Create not confirmed: {task['error']}. Rerun to retry the same key and body.")
        if code == 409:
            sys.exit("409 idempotency_key_reused: same key, different body. Restore the saved body.")
        if code == 503:
            sys.exit("503 no_available_channel: check model access for this key/account; "
                     "contact support with the request id if it persists.")
        if code not in (200, 202) or "id" not in task:
            sys.exit(f"Create failed: HTTP {code} {task}")
        state.update(task_id=task["id"], status=task["status"])
        save(state)

    deadline = time.monotonic() + 900
    while True:
        code, task = call("GET", f"/media/tasks/{state['task_id']}")
        if code is None:
            sys.exit(f"Polling stopped: {task['error']}. The job is still saved; rerun this script.")
        if code != 200 or "status" not in task:
            sys.exit(f"Poll failed: HTTP {code} {task}")
        state["status"] = task["status"]
        save(state)
        if task["status"] in ("failed", "canceled"):
            sys.exit(f"Job {task['status']}: {task.get('error')}")
        if task["status"] == "succeeded":
            break
        if time.monotonic() > deadline:
            sys.exit(f"Still {task['status']} after 900 s. Rerun with the same state file.")
        time.sleep(min(8, 2 + task.get("progress", {}).get("percent", 0) // 20))

    asset = task["assets"][0]
    if asset.get("mime_type") != "image/png":
        sys.exit(f"Expected a PNG grid, got {asset.get('mime_type')}")
    try:
        with urllib.request.urlopen(asset["url"], timeout=120) as response:
            data = response.read()  # signed URL: no Authorization header
    except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError) as error:
        sys.exit(f"Download failed: {error}. Rerun to get a fresh signed URL and retry.")
    digest = hashlib.sha256(data).hexdigest()
    if asset.get("sha256") and digest != asset["sha256"]:
        sys.exit("SHA-256 mismatch. Rerun to download the asset again.")
    part = OUT.with_suffix(".part")
    part.write_bytes(data)
    part.replace(OUT)
    state.update(status="downloaded", download_sha=digest)
    save(state)
    print("saved", OUT, len(data), "bytes")


if __name__ == "__main__":
    main()

Le forme di richiesta e risposta vengono dal test di integrazione del 12 novembre; lo stato di ripresa è l'aggiunta qui, e i suoi rami del ciclo di vita sono stati verificati offline con risposte simulate.

Il file di stato conserva ciò che serve per riprendere: il corpo JSON e l'UUID usato come chiave di idempotenza. Non contiene chiavi API. Viene scritto prima del POST, quindi una connessione interrotta lascia entrambi i valori su disco per il recupero.

La chiamata di creazione può restituire due codici di successo. Un 202 accetta una nuova attività. Un 200 significa che OmniaKey ha già visto la stessa chiave e lo stesso corpo e restituisce l'attività originale; continua a controllare quell'attività. Invia di nuovo lo stesso corpo e la stessa chiave solo quando l'esito della creazione è sconosciuto dopo un errore di connessione. La documentazione indica 409 idempotency_key_reused per un corpo modificato.

Guida il ciclo di controllo con status. La percentuale è indicativa e può restare a 0 mentre l'attività procede. Fermati su succeeded, failed o canceled. La documentazione del ciclo di vita delle attività media di OmniaKey (verificata il 12 novembre 2026) descrive questi stati.

Il limite di controllo è locale. Quando scade, l'attività è ancora in esecuzione o in coda; rieseguire lo stesso script legge l'id salvato e continua a controllare. Un timeout locale non è un annullamento. L'annullamento usa un endpoint separato.

La griglia di immagini Midjourney restituita

Un'attività riuscita ha un solo asset: una griglia PNG. Questo è il campione del test di integrazione del 12 novembre 2026:

La risposta sotto è abbreviata: l'URL è oscurato e expires_at è storico. Aggiunge i campi dell'asset che servono:

{
  "id": "task_DbCWJPfPL1jyhSD38IihNHiiFQao8hjV",
  "status": "succeeded",
  "progress": {"percent": 100, "stage": "completed"},
  "assets": [
    {
      "kind": "image",
      "mime_type": "image/png",
      "width": 2048,
      "height": 2048,
      "byte_size": 5834301,
      "sha256": "f503cef038d15fa150b36b5fe7c41874febad3aa1167698d790781e47a4a10b9",
      "expires_at": "2026-11-12T04:16:41Z",
      "url": "https://…redacted-signed-url…"
    }
  ]
}

Leggi width e height dall'asset invece di fissare 2048. Il quadrato 2048x2048 e i 197,1 secondi sono un campione; trattali come tale. I quattro candidati condividono un PNG, quindi dividi la griglia dopo il download se l'app vuole immagini singole.

L'url è firmata e di breve durata, circa un'ora. Il record dell'attività e l'immagine sono conservati 24 ore dopo la creazione, secondo la documentazione dell'API immagini di OmniaKey (verificata il 12 novembre 2026). Scarica prima che l'URL scada. Se è scaduto, controlla di nuovo l'id dell'attività per ottenere un URL firmato nuovo.

Scarica l'URL dell'asset con un GET normale e senza header Authorization. La firma è nell'URL. Lo script confronta anche i byte scaricati con lo sha256 dell'attività, così un file troncato emerge prima di sovrascrivere la copia buona.

Rieseguire dopo un timeout o un download fallito

Riesegui lo stesso comando e mantieni l'attività. Lo script legge midjourney-task.json e prende il percorso sicuro più breve:

• Creazione non confermata: invia di nuovo POST dello stesso corpo con la stessa chiave di idempotenza.
• L'id esiste e l'attività è ancora in esecuzione: controlla l'id esistente.
• L'attività è riuscita ma manca il PNG: controlla una volta per un URL firmato nuovo e scarica di nuovo.
• Il PNG è già salvato e corrisponde allo SHA-256 registrato: esce senza altre richieste. Un file corrotto viene controllato e scaricato di nuovo.

Funziona nella finestra di conservazione di 24 ore. La stessa chiave con un prompt o parametri diversi produce un 409. Lo script rifiuta un file di stato il cui prompt non corrisponde più; sposta il file se cambi i parametri.

Errori e soluzioni

I guasti sotto derivano dalle regole di attività e idempotenza dell'API. Correggi la causa, poi riesegui con lo stesso file di stato dove serve.

Un 503 no_available_channel non dimostra che a una chiave manchi il permesso per il modello. Due invii con una chiave usata in questo lotto hanno restituito 503 e non hanno creato attività; un'altra chiave ha funzionato in un test di integrazione precedente nello stesso giorno. La causa è sconosciuta: può essere la disponibilità della chiave/account o del canale upstream. Verifica che midjourney sia disponibile per la tua chiave/account attuale. Se il 503 persiste, invia al supporto il corpo della risposta e l'id della richiesta.

Un 200 alla creazione è la ripetizione dell'attività originale. Solo failed e canceled sono guasti terminali; fermare lo script lascia l'attività in esecuzione.

Costo e quando usare un'altra API immagini

OmniaKey addebita una volta per attività riuscita, e le attività fallite o annullate vengono rimborsate automaticamente. La tariffa attuale è sulla pagina del modello Midjourney; controllala prima di stimare un lotto.

Midjourney è adatto per scegliere tra quattro varianti in una griglia. Questa integrazione restituisce quella griglia testo-immagine; per un'immagine per chiamata o una modifica che conserva una scena di origine, vedi GPT Image 2.5 API o Nano Banana Pro API.

Se preferisci digitare un prompt invece di scrivere codice, AI Image Generator esegue modelli di immagine nel browser.

Domande frequenti

Serve un abbonamento Midjourney per usare questa API?

No. La chiave viene da OmniaKey e le chiamate vanno all'API di terze parti di OmniaKey. OmniaKey non è Midjourney; un abbonamento sul sito Midjourney è un altro prodotto e non crea una chiave OmniaKey.

Perché un'attività Midjourney restituisce quattro immagini in un solo file?

Midjourney restituisce una griglia 2x2 con quattro candidati come un unico PNG. Tutti e quattro condividono quel file, e upscale e vary non sono disponibili in questa integrazione, quindi dividi la griglia da solo dopo il download se servono file separati.

Cosa significa 409 idempotency_key_reused?

Hai inviato una chiave di idempotenza che OmniaKey conosce già, ma il corpo della richiesta è cambiato. Invia il corpo originale o avvia una nuova attività con una nuova chiave. Riusare una chiave con un prompt diverso è la causa abituale.

Cosa succede se lo script va in timeout durante il controllo?

L'attività continua. L'id è in midjourney-task.json, quindi rieseguire lo stesso script controllerà quell'id o scaricherà l'asset completato. Un timeout locale lascia viva l'attività e mantiene l'addebito.