Zum Inhalt springen

cat posts/comfyui-api-erstes-bild.md

ComfyUI per API: das erste Bild mit Python erzeugen

Einen ComfyUI-Workflow im API-Format exportieren, Prompt und Seed in Python setzen, den Auftrag per /prompt starten und das fertige Bild über /history und /view herunterladen.

ComfyUI besteht nicht nur aus dem Node-Editor im Browser. Hinter der Oberfläche läuft ein Server, der Workflows entgegennimmt, in eine Queue stellt und auf der GPU ausführt.

Wenn Du in der Oberfläche auf Run klickst, schickt das Frontend den aktuellen Workflow an diesen Server. Dasselbe können wir mit einem Python-Skript erledigen:

Python-Skript
    |
    | POST /prompt
    v
ComfyUI-Queue
    |
    | Workflow ausführen
    v
History
    |
    | GET /history/{prompt_id}
    v
Bildinformationen
    |
    | GET /view
    v
Lokale Bilddatei

In diesem Artikel verwenden wir bewusst nur normale HTTP-Requests. Das ist nicht die eleganteste Lösung für Live-Fortschrittsanzeigen, aber der einfachste nachvollziehbare Weg zum ersten per Skript erzeugten Bild.

Voraussetzungen

Bevor wir Python ins Spiel bringen, sollte die Bilderzeugung direkt in ComfyUI funktionieren.

Du benötigst:

  • eine laufende lokale ComfyUI-Instanz;
  • einen funktionierenden Text-to-Image-Workflow;
  • ein lokal verfügbares Checkpoint-Modell;
  • einen Save Image-Node am Ende des Workflows;
  • Python und das Paket requests.

Standardmäßig ist ComfyUI unter folgender Adresse erreichbar:

http://127.0.0.1:8188

Öffnet sich dort die ComfyUI-Oberfläche, läuft auch der Server, den unser Skript ansprechen wird.

Der Workflow hinter dem Skript

Für das Beispiel genügt der normale Text-to-Image-Workflow von ComfyUI. Er besteht vereinfacht aus folgenden Nodes:

Load Checkpoint
    ├──> CLIP Text Encode (positiver Prompt)
    ├──> CLIP Text Encode (negativer Prompt)
    └──> KSampler <── Empty Latent Image
             |
             v
         VAE Decode
             |
             v
         Save Image

Das Python-Skript baut diesen Graphen nicht selbst auf. Stattdessen erstellen und testen wir den Workflow einmal in der Oberfläche und exportieren ihn anschließend als JSON-Datei.

Das hat einen entscheidenden Vorteil: Alle modellabhängigen Einstellungen bleiben in ComfyUI. Das Skript muss später nur noch einzelne Werte wie Prompt und Seed ändern.

Führe den Workflow deshalb zunächst einmal über die Oberfläche aus. Erst wenn dort ein Bild erzeugt und vom Save Image-Node gespeichert wird, lohnt sich der nächste Schritt.

Workflow im API-Format exportieren

ComfyUI kann Workflows in unterschiedlichen JSON-Formaten speichern. Für einen API-Aufruf benötigen wir ausdrücklich das API-Format.

Wähle in der aktuellen ComfyUI-Oberfläche:

File → Export Workflow (API)

Speichere die Datei als:

workflow_api.json

Das normale Speichern eines Workflows ist für diesen Zweck nicht geeignet. Dieses Format enthält zusätzliche Informationen für die Oberfläche, beispielsweise Positionen, Größen und Gruppen der Nodes.

Das API-Format beschreibt dagegen vor allem:

  • welche Nodes ausgeführt werden;
  • welche Inputs sie erhalten;
  • wie ihre Ein- und Ausgänge verbunden sind.

Ein Ausschnitt kann beispielsweise so aussehen:

{
  "3": {
    "inputs": {
      "seed": 42,
      "steps": 20,
      "cfg": 7.0,
      "sampler_name": "euler",
      "scheduler": "normal",
      "denoise": 1,
      "model": ["4", 0],
      "positive": ["6", 0],
      "negative": ["7", 0],
      "latent_image": ["5", 0]
    },
    "class_type": "KSampler"
  },
  "6": {
    "inputs": {
      "text": "a small robot repairing a computer",
      "clip": ["4", 1]
    },
    "class_type": "CLIPTextEncode"
  },
  "7": {
    "inputs": {
      "text": "blurry, text, watermark",
      "clip": ["4", 1]
    },
    "class_type": "CLIPTextEncode"
  }
}

Die Schlüssel "3", "6" und "7" sind die IDs der Nodes.

Ein Wert wie:

"positive": ["6", 0]

bedeutet:

Verwende Ausgang 0 von Node 6 als positiven Prompt.

Dadurch können wir auch erkennen, welcher der beiden CLIPTextEncode-Nodes der positive und welcher der negative Prompt ist.

Die benötigten Node-IDs ermitteln

Für unser Skript benötigen wir zunächst drei IDs:

  • den KSampler;
  • den CLIPTextEncode-Node für den positiven Prompt;
  • den CLIPTextEncode-Node für den negativen Prompt.

Suche in workflow_api.json nach:

"class_type": "KSampler"

Die dazugehörige ID steht direkt darüber. Im Beispiel ist es Node "3".

Innerhalb dieses Nodes findest Du außerdem die Verbindungen:

"positive": ["6", 0],
"negative": ["7", 0]

Damit wissen wir:

KSampler:          3
Positiver Prompt:  6
Negativer Prompt:  7

Diese IDs sind nicht universell. Wenn Du Nodes löschst, neu anlegst oder einen anderen Workflow verwendest, können sie anders lauten.

Python-Projekt vorbereiten

Lege für das Beispiel ein neues Verzeichnis an:

mkdir comfy-api
cd comfy-api
uv init
uv add requests

Kopiere anschließend workflow_api.json in dieses Verzeichnis und lege daneben eine Datei namens generate.py an:

comfy-api/
├── generate.py
└── workflow_api.json

Das Skript verwendet nur requests. Pillow oder andere Pakete zur Bildverarbeitung werden nicht benötigt, da wir die von ComfyUI gelieferten Bytes direkt in eine Datei schreiben.

Das vollständige Skript

Kopiere folgenden Code nach generate.py:

from __future__ import annotations

import argparse
import json
import random
import sys
import time
from pathlib import Path
from typing import Any

import requests


SERVER_URL = "http://127.0.0.1:8188"
WORKFLOW_FILE = Path("workflow_api.json")
OUTPUT_DIR = Path("output")

# Diese IDs stammen aus dem exportierten Workflow.
# Passe sie an Deinen eigenen Workflow an.
SAMPLER_NODE_ID = "3"
POSITIVE_PROMPT_NODE_ID = "6"
NEGATIVE_PROMPT_NODE_ID = "7"

HTTP_TIMEOUT = 30
GENERATION_TIMEOUT = 600
POLL_INTERVAL = 0.5


def load_workflow(path: Path) -> dict[str, Any]:
    """Lädt einen Workflow im ComfyUI-API-Format."""

    with path.open(encoding="utf-8") as file:
        workflow = json.load(file)

    if not isinstance(workflow, dict):
        raise ValueError("Der Workflow muss ein JSON-Objekt sein.")

    return workflow


def set_node_input(
    workflow: dict[str, Any],
    node_id: str,
    input_name: str,
    value: Any,
) -> None:
    """Setzt einen Input eines bestimmten Nodes."""

    try:
        workflow[node_id]["inputs"][input_name] = value
    except KeyError as error:
        raise KeyError(
            f"Node {node_id!r} oder Input {input_name!r} "
            "wurde im Workflow nicht gefunden."
        ) from error


def parse_json_response(response: requests.Response) -> Any:
    """Prüft eine HTTP-Antwort und liest das enthaltene JSON."""

    if not response.ok:
        try:
            details = json.dumps(
                response.json(),
                ensure_ascii=False,
                indent=2,
            )
        except ValueError:
            details = response.text

        raise RuntimeError(
            f"ComfyUI antwortete mit HTTP {response.status_code}:\n"
            f"{details}"
        )

    try:
        return response.json()
    except ValueError as error:
        raise RuntimeError(
            "ComfyUI lieferte keine gültige JSON-Antwort."
        ) from error


def queue_prompt(workflow: dict[str, Any]) -> str:
    """Stellt einen Workflow in die ComfyUI-Queue."""

    response = requests.post(
        f"{SERVER_URL}/prompt",
        json={"prompt": workflow},
        timeout=HTTP_TIMEOUT,
    )
    data = parse_json_response(response)

    if not isinstance(data, dict) or "prompt_id" not in data:
        raise RuntimeError(
            "Die Antwort von /prompt enthält keine prompt_id:\n"
            f"{json.dumps(data, ensure_ascii=False, indent=2)}"
        )

    return str(data["prompt_id"])


def wait_for_result(prompt_id: str) -> dict[str, Any]:
    """Wartet, bis ComfyUI den Auftrag in die History geschrieben hat."""

    deadline = time.monotonic() + GENERATION_TIMEOUT

    while time.monotonic() < deadline:
        response = requests.get(
            f"{SERVER_URL}/history/{prompt_id}",
            timeout=HTTP_TIMEOUT,
        )
        history = parse_json_response(response)

        if isinstance(history, dict) and prompt_id in history:
            result = history[prompt_id]

            if not isinstance(result, dict):
                raise RuntimeError(
                    "Der History-Eintrag hat ein unerwartetes Format."
                )

            return result

        time.sleep(POLL_INTERVAL)

    raise TimeoutError(
        f"Nach {GENERATION_TIMEOUT} Sekunden liegt noch kein Ergebnis vor."
    )


def collect_images(
    result: dict[str, Any],
) -> list[dict[str, str]]:
    """Sammelt alle Bildinformationen aus einem History-Eintrag."""

    status = result.get("status", {})

    if (
        isinstance(status, dict)
        and status.get("status_str") == "error"
    ):
        messages = status.get("messages", [])

        raise RuntimeError(
            "ComfyUI konnte den Workflow nicht ausführen:\n"
            f"{json.dumps(messages, ensure_ascii=False, indent=2)}"
        )

    images: list[dict[str, str]] = []

    for node_output in result.get("outputs", {}).values():
        for image in node_output.get("images", []):
            required_fields = ("filename", "subfolder", "type")

            if all(field in image for field in required_fields):
                images.append(image)

    if not images:
        raise RuntimeError(
            "Der Workflow wurde beendet, enthält aber keine abrufbaren "
            "Bilder. Prüfe, ob er einen 'Save Image'-Node besitzt."
        )

    return images


def download_image(
    image: dict[str, str],
    index: int,
) -> Path:
    """Lädt ein von ComfyUI erzeugtes Bild herunter."""

    response = requests.get(
        f"{SERVER_URL}/view",
        params={
            "filename": image["filename"],
            "subfolder": image["subfolder"],
            "type": image["type"],
        },
        timeout=HTTP_TIMEOUT,
    )
    response.raise_for_status()

    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

    suffix = Path(image["filename"]).suffix or ".png"
    target = OUTPUT_DIR / f"bild_{index:02d}{suffix}"
    target.write_bytes(response.content)

    return target


def parse_arguments() -> argparse.Namespace:
    """Liest Prompt, negativen Prompt und Seed ein."""

    parser = argparse.ArgumentParser(
        description="Erzeugt über die lokale ComfyUI-API ein Bild."
    )
    parser.add_argument(
        "--prompt",
        default=(
            "a small friendly robot repairing a computer, "
            "detailed illustration"
        ),
        help="Positiver Prompt",
    )
    parser.add_argument(
        "--negative",
        default="blurry, text, watermark, low quality",
        help="Negativer Prompt",
    )
    parser.add_argument(
        "--seed",
        type=int,
        help=(
            "Fester Seed; ohne Angabe wird ein zufälliger Seed "
            "verwendet."
        ),
    )

    return parser.parse_args()


def main() -> None:
    args = parse_arguments()

    seed = (
        args.seed
        if args.seed is not None
        else random.randrange(0, 2**32)
    )

    workflow = load_workflow(WORKFLOW_FILE)

    set_node_input(
        workflow,
        POSITIVE_PROMPT_NODE_ID,
        "text",
        args.prompt,
    )
    set_node_input(
        workflow,
        NEGATIVE_PROMPT_NODE_ID,
        "text",
        args.negative,
    )
    set_node_input(
        workflow,
        SAMPLER_NODE_ID,
        "seed",
        seed,
    )

    print(f"Seed: {seed}")

    prompt_id = queue_prompt(workflow)
    print(f"Auftrag gestartet: {prompt_id}")

    result = wait_for_result(prompt_id)
    images = collect_images(result)

    for index, image in enumerate(images, start=1):
        target = download_image(image, index)
        print(f"Gespeichert: {target}")


if __name__ == "__main__":
    try:
        main()
    except (
        OSError,
        ValueError,
        KeyError,
        RuntimeError,
        TimeoutError,
        requests.RequestException,
    ) as error:
        print(f"Fehler: {error}", file=sys.stderr)
        raise SystemExit(1)

Prüfe vor dem ersten Start noch einmal diese drei Werte:

SAMPLER_NODE_ID = "3"
POSITIVE_PROMPT_NODE_ID = "6"
NEGATIVE_PROMPT_NODE_ID = "7"

Sie müssen zu den IDs in Deiner workflow_api.json passen.

Das erste Bild erzeugen

Starte das Skript zunächst ohne weitere Argumente:

uv run python generate.py

Das Skript verwendet dann den eingebauten Beispiel-Prompt und erzeugt bei jedem Aufruf einen zufälligen Seed.

Eine erfolgreiche Ausgabe sieht ungefähr so aus:

Seed: 3277080762
Auftrag gestartet: 9d9c6bd9-2caf-48ad-b988-f48b928ca88a
Gespeichert: output/bild_01.png

Die heruntergeladene Datei liegt anschließend im lokalen Verzeichnis output:

comfy-api/
├── output/
│   └── bild_01.png
├── generate.py
└── workflow_api.json

ComfyUI selbst speichert durch den Save Image-Node zusätzlich eine Kopie in seinem eigenen Output-Verzeichnis. Unser Skript lädt diese Datei über die API herunter und speichert sie noch einmal im Python-Projekt.

Einen eigenen Prompt übergeben

Den positiven Prompt kannst Du über --prompt setzen:

uv run python generate.py \
  --prompt "a corgi astronaut on the moon, cinematic lighting"

Auch der negative Prompt lässt sich überschreiben:

uv run python generate.py \
  --prompt "an old library inside a giant tree, fantasy illustration" \
  --negative "blurry, text, watermark, distorted architecture"

Die Anführungszeichen sind wichtig, damit die Shell den vollständigen Text als ein Argument behandelt.

Einen festen Seed verwenden

Ohne --seed erzeugt das Skript bei jedem Aufruf einen neuen zufälligen Seed:

random.randrange(0, 2**32)

Für reproduzierbare Ergebnisse kannst Du einen festen Wert angeben:

uv run python generate.py \
  --prompt "a corgi astronaut on the moon, cinematic lighting" \
  --seed 42

Solange Workflow, Modell, Prompt und weitere Einstellungen identisch bleiben, liefert derselbe Seed in der Regel auch dasselbe Ergebnis.

Änderst Du lediglich den Prompt, kann ComfyUI die unveränderten Teile des Workflows aus seinem Cache übernehmen und nur die davon abhängigen Nodes erneut ausführen.

Was beim API-Aufruf passiert

Das Skript verwendet drei Endpunkte.

POST /prompt

Der Workflow wird als JSON an ComfyUI gesendet:

response = requests.post(
    f"{SERVER_URL}/prompt",
    json={"prompt": workflow},
    timeout=HTTP_TIMEOUT,
)

ComfyUI validiert den Graphen und stellt ihn in seine Ausführungs-Queue. Bei Erfolg enthält die Antwort unter anderem eine prompt_id:

{
  "prompt_id": "9d9c6bd9-2caf-48ad-b988-f48b928ca88a",
  "number": 4,
  "node_errors": {}
}

Die prompt_id identifiziert genau diesen Auftrag.

Kann der Workflow bereits vor der Ausführung nicht validiert werden, antwortet ComfyUI mit einem HTTP-Fehler und zusätzlichen Informationen zu den betroffenen Nodes. Unser Skript gibt diese Antwort vollständig aus.

GET /history/{prompt_id}

Die Ausführung erfolgt asynchron. Das bedeutet: Der Request an /prompt wartet nicht, bis das Bild fertig ist.

Das Skript fragt deshalb regelmäßig folgenden Endpunkt ab:

GET /history/9d9c6bd9-2caf-48ad-b988-f48b928ca88a

Solange der Auftrag noch nicht abgeschlossen wurde, ist der gesuchte Eintrag dort noch nicht vorhanden. Nach der Ausführung enthält er unter outputs die Ergebnisse der Output-Nodes.

Für einen Save Image-Node sieht der relevante Teil ungefähr so aus:

{
  "outputs": {
    "9": {
      "images": [
        {
          "filename": "ComfyUI_00001_.png",
          "subfolder": "",
          "type": "output"
        }
      ]
    }
  }
}

Diese Informationen beschreiben eine Datei auf dem ComfyUI-Server. Sie enthalten das Bild noch nicht selbst.

GET /view

Das eigentliche Bild wird über /view abgerufen. Dazu übergeben wir genau die drei Felder aus der History:

params = {
    "filename": image["filename"],
    "subfolder": image["subfolder"],
    "type": image["type"],
}

Der Server liefert daraufhin die Binärdaten der Datei. Diese schreibt das Skript mit write_bytes() in das lokale Output-Verzeichnis.

Warum wir keine client_id benötigen

In vielen ComfyUI-Beispielen taucht zusätzlich eine client_id auf:

{
  "prompt": {},
  "client_id": "eine-uuid"
}

Für die hier verwendete HTTP-Polling-Variante ist sie nicht notwendig. Die prompt_id genügt, um das fertige Ergebnis später aus der History abzurufen.

Eine client_id wird insbesondere dann relevant, wenn sich ein Client zusätzlich mit dem WebSocket-Endpunkt verbindet:

ws://127.0.0.1:8188/ws?clientId=<client_id>

Über diese Verbindung sendet ComfyUI unter anderem:

  • den Beginn einer Ausführung;
  • den aktuell ausgeführten Node;
  • den Fortschritt eines Samplers;
  • Fehlermeldungen;
  • das Ende der Ausführung.

Für eine Anwendung mit Fortschrittsanzeige ist die Kombination aus WebSocket und anschließendem Abruf der History die bessere Lösung. Für das erste Skript würde sie jedoch zusätzliche Abhängigkeiten und deutlich mehr Code bedeuten.

Mehrere Bilder aus einem Workflow

Das Skript sammelt nicht nur das erste gefundene Bild. Es durchläuft alle Output-Nodes und alle darin enthaltenen images-Einträge:

for node_output in result.get("outputs", {}).values():
    for image in node_output.get("images", []):
        images.append(image)

Dadurch funktioniert es auch, wenn:

  • der batch_size größer als 1 ist;
  • der Workflow mehrere Save Image-Nodes besitzt;
  • mehrere Bilder von unterschiedlichen Nodes zurückgegeben werden.

Die Dateien werden fortlaufend benannt:

bild_01.png
bild_02.png
bild_03.png

Häufige Fehler

Verbindung zu ComfyUI nicht möglich

Eine Meldung wie:

Failed to establish a new connection

bedeutet meistens, dass ComfyUI nicht läuft oder unter einer anderen Adresse erreichbar ist.

Prüfe zunächst im Browser:

http://127.0.0.1:8188

Verwendest Du einen anderen Port, passe SERVER_URL an:

SERVER_URL = "http://127.0.0.1:8288"

Node oder Input wurde nicht gefunden

Eine Meldung wie:

Node '6' oder Input 'text' wurde im Workflow nicht gefunden.

bedeutet, dass die im Skript eingetragene Node-ID nicht zum exportierten Workflow passt.

Öffne workflow_api.json und ermittle die IDs erneut über den KSampler und seine Inputs positive und negative.

HTTP 400 mit node_errors

In diesem Fall konnte ComfyUI den Workflow nicht validieren. Häufige Ursachen sind:

  • ein im Workflow ausgewähltes Modell ist nicht vorhanden;
  • ein Custom Node ist nicht installiert;
  • ein Node besitzt einen ungültigen Wert;
  • der Workflow wurde nicht im API-Format exportiert;
  • eine Verbindung zwischen zwei Nodes verweist auf eine nicht vorhandene ID.

Das Skript gibt die vollständige Antwort von ComfyUI aus. Besonders hilfreich ist der Abschnitt node_errors, weil dort die betroffenen Node-IDs genannt werden.

Der Workflow liefert keine Bilder

Wird der Auftrag erfolgreich beendet, aber das Skript meldet:

Der Workflow wurde beendet, enthält aber keine abrufbaren Bilder.

fehlt meistens ein geeigneter Output-Node.

Verbinde den Ausgang von VAE Decode mit einem Save Image-Node, führe den Workflow erneut in der Oberfläche aus und exportiere ihn danach noch einmal im API-Format.

Das Skript läuft in einen Timeout

Standardmäßig wartet das Skript höchstens zehn Minuten:

GENERATION_TIMEOUT = 600

Bei großen Modellen, sehr hohen Auflösungen oder einer bereits gefüllten Queue kann das zu kurz sein. Der Wert lässt sich in Sekunden erhöhen:

GENERATION_TIMEOUT = 1800

Die einzelnen HTTP-Requests besitzen zusätzlich einen eigenen Timeout:

HTTP_TIMEOUT = 30

Dadurch hängt das Skript nicht unbegrenzt, wenn der Server während der Generierung nicht mehr antwortet.

Trotz mehrerer Aufrufe entsteht dasselbe Bild

Verwendest Du denselben Workflow, denselben Prompt und denselben Seed, ist ein identisches Ergebnis beabsichtigt.

Lasse --seed weg oder übergib einen anderen Wert:

uv run python generate.py --seed 123456

ComfyUI auf einem anderen Rechner

Standardmäßig lauscht ComfyUI nur auf der lokalen Adresse 127.0.0.1. Ein anderer Rechner kann den Server deshalb nicht direkt erreichen.

Für einen Server im lokalen Netzwerk kann ComfyUI beispielsweise mit folgender Option gestartet werden:

python main.py --listen 0.0.0.0

Oder kürzer:

python main.py --listen

Danach muss im Skript die Adresse des Servers eingetragen werden:

SERVER_URL = "http://192.168.1.50:8188"

Ein so gestarteter Server sollte nicht ungeschützt aus dem Internet erreichbar sein. Für einen dauerhaften Remote-Betrieb gehören mindestens Netzwerkfilter, TLS und eine Authentifizierung vor den Dienst, beispielsweise über einen Reverse Proxy.

Für rein lokale Skripte bleibt 127.0.0.1 die einfachste Variante.

Wie es weitergehen kann

Mit dem vollständigen Ablauf haben wir nun alle Bausteine, die auch eine größere Anwendung benötigt:

  1. Workflow laden;
  2. Inputs verändern;
  3. Workflow in die Queue stellen;
  4. auf das Ergebnis warten;
  5. Output-Dateien herunterladen.

Darauf lassen sich unter anderem aufbauen:

  • Batch-Generierungen aus einer Prompt-Liste;
  • wechselnde Seeds und Bildgrößen;
  • ein Web-Frontend für eigene Workflows;
  • Telegram- oder Discord-Bots;
  • automatisierte Upscaling-Pipelines;
  • mehrere nacheinander geschaltete ComfyUI-Workflows;
  • Live-Fortschritt über WebSockets.

Der wichtigste Schritt ist dabei nicht der HTTP-Request selbst. Entscheidend ist, dass Du einen funktionierenden Workflow in der Oberfläche erstellst, ihn im API-Format exportierst und anschließend gezielt seine Node-Inputs veränderst.

Weiterlesen

  • uv: pip und venv in schnellgeplant – Python-Projekte und Abhängigkeiten mit uv verwalten.

Quellen

0 Kommentare

Noch keine Kommentare. Sei der/die Erste!