fabricatio-comfyui

Async ComfyUI API integration for Fabricatio — generate images from typed knobs and download the results. Built on httpx with full Pydantic-typed API coverage.

Design: workflows are fully internal

The package owns its workflow graph: it is initialised entirely in Python code (an internal typed Graph model) and serialized to ComfyUI's API format only on submission. Callers never see, construct, or operate on a workflow — they supply high-level knobs (prompt, negative_prompt, width, height, seed, steps, cfg, checkpoint) and the package parameterises the built-in template internally. There is no dict[str, Any] workflow injection anywhere in the public signatures.

Naming follows fabricatio-skill: one Use* capability mixin (UseComfyUI), a module-level one-shot function (generate_image), and bare-noun response models (ExecutionResult, QueueInfo, …). workflows/ stays a docstring-only namespace, as in fabricatio-skill.

Architecture

| Layer | Module / Class | Purpose | | Graph | Graph (models/graph.py) | Typed graph initialised in Python, serialized to the ComfyUI wire format; internal only | | Transport | ComfyUIHttpClient / ComfyUIClientBase | Async REST client; shared per-URL via get_comfyui_client | | Capability | UseComfyUI (capabilities/comfyui.py) | Mixin: high-level generate (queue → poll → download) | | API | api.py | One-shot functions that run on the shared pooled client | | Actions | GenerateImage | Pluggable step for Fabricatio WorkFlow |

Installation

pip install fabricatio[comfyui]
# or
uv pip install fabricatio[comfyui]

Configuration

All options below are read through the fabricatio configuration chain (see the Configuration Guide). Set them under the [ext.comfyui] table in fabricatio.toml, equivalently under [tool.fabricatio.ext.comfyui] in pyproject.toml, or via FABRICATIO_EXT__COMFYUI__<FIELD_UPPER> environment variables.

[ext.comfyui]
base_url = "http://127.0.0.1:8188"
timeout = 300.0
download_dir = "./outputs"

Access at runtime: from fabricatio_comfyui.config import comfyui_config.

Usage

One-shot functions

The lowest-friction entry point — no Role, no client, no workflow:

import asyncio
from fabricatio_comfyui import generate_image


async def main() -> None:
    path = await generate_image(
        "masterpiece, best quality, a mountain landscape",
        negative_prompt="worst quality, blurry",
        width=1024,
        height=768,
    )
    print(path)  # Path to the generated image, or None on failure


asyncio.run(main())

Capability mixin (with a Role or Action)

Mix UseComfyUI into a Role to get the same typed-knob methods:

from fabricatio import Role
from fabricatio_comfyui import UseComfyUI


class ImageRole(Role, UseComfyUI):
    """Role with ComfyUI image generation capability."""


# then: path = await role.generate_image("a mountain landscape")  # Path | None

Action (in a WorkFlow)

Use GenerateImage as a composable step:

from fabricatio import WorkFlow
from fabricatio_comfyui import GenerateImage

GenerateImageWorkflow = WorkFlow(
    name="ComfyUI Generate",
    steps=(
        GenerateImage(
            prompt="masterpiece, best quality",
            download_dir="./outputs",
        ),
    ),
)

Standalone client (advanced)

The mixin and the one-shot functions share one pooled client per server URL, obtained from the cached factory fabricatio_comfyui.http_client.get_comfyui_client — never close the client it returns. When you need a private pool instead (tests, alternate backends), build a scoped one with async with:

import asyncio
from fabricatio_comfyui import ComfyUIHttpClient


async def main() -> None:
    async with ComfyUIHttpClient.create() as client:
        results = await client.generate("a mountain landscape", seed=42)
        if results[0].succeeded():
            await client.download_images(results[0], "./outputs")


asyncio.run(main())

API Reference

Capability methods (UseComfyUI)

generate_image keyword parameters: negative_prompt, width, height, seed, steps, cfg, checkpoint, download_dir, timeout. download_dir resolution: per-call argument → scoped config on the Role (UseComfyUI inherits ComfyUIScopedConfig; set it as a subclass default class ImageRole(Role, UseComfyUI): download_dir: str = "./outputs" or per instance) → [ext.comfyui] download_dir config.

On failure the methods above return None — a single prompt yields at most one image file path. A list of prompts yields a list with one entry per prompt (Path | None each, None marking a failed generation), generated sequentially:

paths = await role.generate_image(["a mountain", "a river"])
# paths: list[Path | None] — [path, None] when the second generation failed

The module-level function in fabricatio_comfyui.api (generate_image) shares the exact same keyword surface and hides the client lifecycle entirely. Server-side state inspection and control (queue, history, interrupt) stay on the client transport, which mirrors the REST API one-to-one — generate interrupts the running job automatically when its awaiting task is cancelled.

Client methods (ComfyUIHttpClient / ComfyUIClientBase)

Actions

Models

All API responses are deserialized into frozen Pydantic models:

License

MIT — see the LICENSE file.