fabricatio_comfyui.capabilities.comfyui

ComfyUI capability mixin.

Mix into a Role to gain ComfyUI interaction methods. Clients are shared via ComfyuiHTTPClient.create() (@lru_cache) — no per-instance state.

Predicate-verb methods (acomfyui_*) follow the same naming convention as fabricatio_core.capabilities.usages.UseLLM — a prefix + domain verb.

Classes

ComfyUI capability mixin — delegates to a pooled ComfyuiClient.

class fabricatio_comfyui.capabilities.comfyui.Comfyui

ComfyUI capability mixin — delegates to a pooled ComfyuiClient.

async acomfyui_generate(workflow: Dict[str, Any] | Workflow, *, download_dir: str | Path | None = None, timeout: float | None = None, base_url: str | None = None) → fabricatio_comfyui.models.comfyui.ComfyuiExecutionResult
async acomfyui_generate(workflow: List[Dict[str, Any] | Workflow], *, download_dir: list[str | Path | None] | None = None, timeout: float | None = None, base_url: str | None = None) → List[ComfyuiExecutionResult]: Execute one or more workflows: queue all, then poll all, then download.

async acomfyui_queue(workflow: Dict[str, Any] | Workflow, **kwargs: Unpack[QueueKwargs]) → fabricatio_comfyui.models.comfyui.PromptResponse
async acomfyui_queue(workflow: List[Dict[str, Any] | Workflow], **kwargs: Unpack[QueueKwargs]) → List[PromptResponse]: Submit one or more workflows for execution without waiting.

async acomfyui_inspect_queue() → fabricatio_comfyui.models.comfyui.QueueInfo: Fetch the current execution queue state.

async acomfyui_history(prompt_id: str) → HistoryEntry | None: Retrieve execution history for prompt_id.

async acomfyui_retrieve(prompt_id: str, **kwargs: Unpack[PollKwargs]) → fabricatio_comfyui.models.comfyui.ComfyuiExecutionResult
async acomfyui_retrieve(prompt_id: List[str], **kwargs: Unpack[PollKwargs]) → List[ComfyuiExecutionResult]: Poll until one or more prompt_ids complete.

async acomfyui_retrieve_image(filename: str, **kwargs: Unpack[ViewImageKwargs]) → bytes: Download a single generated image by filename.

async acomfyui_upload(image_path: str | Path, **kwargs: Unpack[UploadKwargs]) → fabricatio_comfyui.models.comfyui.UploadResponse: Upload an image to the server.

async acomfyui_interrupt() → None: Interrupt the currently running workflow.

async comfyui_generate(workflow: Dict[str, Any] | Workflow | List[Dict[str, Any] | Workflow], **kwargs: Any) → ComfyuiExecutionResult | List[ComfyuiExecutionResult]: Alias for acomfyui_generate().

async comfyui_queue_prompt(workflow: Dict[str, Any] | Workflow | List[Dict[str, Any] | Workflow], **kwargs: Unpack[QueueKwargs]) → PromptResponse | List[PromptResponse]: Alias for acomfyui_queue().

async comfyui_get_queue_info() → fabricatio_comfyui.models.comfyui.QueueInfo: Alias for acomfyui_inspect_queue().

async comfyui_get_history(prompt_id: str) → HistoryEntry | None: Alias for acomfyui_history().

async comfyui_wait_for_completion(prompt_id: str | List[str], **kwargs: Unpack[PollKwargs]) → ComfyuiExecutionResult | List[ComfyuiExecutionResult]: Alias for acomfyui_retrieve().

async comfyui_get_image(filename: str, **kwargs: Unpack[ViewImageKwargs]) → bytes: Alias for acomfyui_retrieve_image().

async comfyui_upload_image(image_path: str | Path, **kwargs: Unpack[UploadKwargs]) → fabricatio_comfyui.models.comfyui.UploadResponse: Alias for acomfyui_upload().