fabricatio_comfyui.models.workflow

Typed models for ComfyUI workflow graphs.

Provides Node (a single workflow node) and Workflow (a full workflow graph) with proper typing, validation, and clean API-format serialization.

Classes

`NodeRef`	A typed reference to another node's output.
`Node`	A single node in a ComfyUI workflow graph.
`Workflow`	Programmatic builder for ComfyUI workflow graphs.

Module Contents

class fabricatio_comfyui.models.workflow.NodeRef(/, **data: Any)

Bases: pydantic.BaseModel

A typed reference to another node’s output.

In the ComfyUI API format, connections are stored as [node_id, output_index]. This model makes that explicit.

model_config: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

node_id: str: The source node ID.

output_index: int = 0: The output index on the source node (default 0).

to_api() → list[str | int]: Serialize to ComfyUI API format: [node_id, output_index].

classmethod from_api(raw: list[Any]) → Self: Parse from [node_id, output_index].

static is_ref(value: Any) → bool: Return True if value looks like a node reference.

class fabricatio_comfyui.models.workflow.Node(/, **data: Any)

Bases: pydantic.BaseModel

A single node in a ComfyUI workflow graph.

Inputs are stored in their API-format representation. Literal values keep their native types. Node connections are stored as [node_id, output_index] lists — use connect() / get_ref() for typed access.

model_config: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str: Node identifier (e.g. "42").

type: str: ComfyUI node class (e.g. "KSampler", "CLIPTextEncode").

inputs: Dict[str, Any] = None: Input values. Literals are native types; node refs are [node_id, output_index].

title: str = '': Human-readable title (stored in _meta.title).

set_input(name: str, value: Any) → Self: Set a literal input value.

connect(input_name: str, source: Node, output_index: int = 0) → Self: Wire input_name to source’s output at output_index.

get_ref(input_name: str) → NodeRef | None: Return a NodeRef if the input is a connection, else None.

to_api() → Dict[str, Any]: Serialize to ComfyUI API format.

__repr__() → str: Return a developer-friendly representation of this node.

class fabricatio_comfyui.models.workflow.Workflow

Programmatic builder for ComfyUI workflow graphs.

Load from exported API-format JSON or build from scratch:

wf = Workflow.from_file("demo.json")
wf.set_positive_prompt("masterpiece, best quality")
wf.set_checkpoint("v1-5-pruned-emaonly.safetensors")

# Serialize for client.queue_prompt()
data = wf.to_api()

# Build from scratch
wf = Workflow.new()
ckpt = wf.add("CheckpointLoaderSimple", inputs={"ckpt_name": "model.safetensors"})
empty = wf.add("EmptyLatentImage", inputs={"width": 512, "height": 512, "batch_size": 1})
pos = wf.add("CLIPTextEncode", inputs={"text": "a cat"})
pos.connect("clip", ckpt, 1)

node_map: Dict[str, Node]

counter: int = 1

classmethod new() → Self: Create an empty workflow.

classmethod from_api(data: Dict[str, Any]) → Self: Load from a ComfyUI API-format JSON dict.

classmethod from_file(path: str | pathlib.Path) → Self: Load from a .json file.

classmethod default() → Self: Load the bundled demo workflow shipped with the package.

add(type: str, *, title: str = '', inputs: Dict[str, Any] | None = None) → Node: Add a new node, auto-assigning the next available ID.

get(node_id: str) → Node: Get a node by ID. Raises KeyError if not found.

by_type(type: str) → List[Node]: Find all nodes with the given type.

remove(node_id: str) → None: Remove a node and disconnect all references to it.

property node_ids: list[str]: All node IDs in this workflow.

property nodes: list[Node]: All nodes in this workflow.

to_api() → Dict[str, Any]: Serialize to ComfyUI API format (pass to client.queue_prompt).

set_checkpoint(ckpt_name: str, *, node_id: str | None = None) → Node: Set the checkpoint on a CheckpointLoaderSimple node.

set_vae(vae_name: str, *, node_id: str | None = None) → Node: Set the VAE on a VAELoader node.

set_positive_prompt(text: str, *, node_id: str | None = None) → Node: Set the positive prompt text on a CLIPTextEncode node.

set_negative_prompt(text: str, *, node_id: str | None = None) → Node: Set the negative prompt text on a CLIPTextEncode node (second one).

set_sampler(*, seed: int | None = None, steps: int | None = None, cfg: float | None = None, sampler_name: str | None = None, scheduler: str | None = None, denoise: float | None = None, node_id: str | None = None) → Node: Update sampler parameters on a KSampler or KSamplerAdvanced node.

set_resolution(*, width: int | None = None, height: int | None = None, node_id: str | None = None) → Node: Set width/height on an EmptyLatentImage node.

__repr__() → str: Return a developer-friendly representation of this workflow.