Adapt Comfyui¶

Adapt ComfyUI -- mvp.adapt_comfyui

Cluster: Creative & Media | Type: component | MCP Tools: 61

Overview¶

ComfyUI workflow engine bridging G6's AI pipeline with a running ComfyUI server. Two-tier architecture: Tier 1 (AdaptComfyUIBlock) covers txt2img, img2img, inpainting, upscaling, ControlNet, LoRA/SDXL, raw workflow execution, readiness checks, capability discovery, object-info/node-schema inspection, and constrained backend-native operations. Tier 2 (AdaptComfyUIMCPBlock) adds SQLite persistence (7 tables), a professional design & illustration pipeline with brand kits, format presets (20 built-in), asset management, style consistency, project tracking, and MCP-facing degradation envelopes. Requires a running ComfyUI server (default http://127.0.0.1:8188) for generation.

Production caveat

adapt_comfyui is an orchestration adapter, not a standalone image generator. Production image generation requires a reachable ComfyUI server plus the checkpoint models, LoRAs, upscale models, and custom nodes referenced by the selected workflow. Without that runtime, health checks return offline/degraded status and MCP-facing generation operations surface completion_state, warning_card, evidence, and recovery actions rather than cosmetic success. Treat the design pipeline as production-ready only after validating the target server, installed models, and workflow-specific nodes in the deployment environment.

When to use:

Generating images from text prompts through a local or remote ComfyUI server
Professional design pipeline: brand kits, format presets, style-consistent generation
Submitting custom node-graph workflows (JSON) for advanced diffusion pipelines
Inspecting backend-native ComfyUI object info and node schemas before execution
Running constrained, audited backend-native operations when curated tools are insufficient
Running batch generation jobs and polling their status asynchronously
Managing design assets, projects, and style references across sessions

Example:

from mvp.adapt_comfyui import AdaptComfyUIBlock, ComfyUIInput

block = AdaptComfyUIBlock(name="comfyui")
result = block.infer(ComfyUIInput(
    op="txt2img",
    prompt_text="a photorealistic mountain landscape at golden hour",
    width=768,
    height=512,
    steps=25,
    cfg_scale=7.0,
))
# result.value.images_base64 → list of base64 PNG strings

Works well with: adapt_creative_api, adapt_image, adapt_blender

Public API¶

`AdaptComfyUIBlock(AIBlock[ComfyUIInput, ComfyUIOutput, None])`¶

ComfyUI adapter block.

Field	Type	Default
`name`	`str`	'adapt_comfyui'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: ComfyUIInput) -> Result[ComfyUIOutput]`¶

`ComfyUIInput(BaseModel)`¶

Input schema for AdaptComfyUIBlock.

Field	Type	Default
`op`	`str`	'get_status'
`workflow_json`	`str`	''
`prompt_text`	`str`	''
`negative_prompt`	`str`	''
`image_base64`	`str`	''
`mask_base64`	`str`	''
`model_name`	`str`	''
`width`	`int`	512
`height`	`int`	512
`steps`	`int`	20
`cfg_scale`	`float`	7.0
`seed`	`int`	-1
`sampler`	`str`	'euler'
`scheduler`	`str`	'normal'
`denoise`	`float`	1.0
`controlnet_model`	`str`	''
`controlnet_image_base64`	`str`	''
`workflow_name`	`str`	''
`preset_name`	`str`	''
`batch_size`	`int`	1
`prompt_id`	`str`	''
`timeout_seconds`	`int`	120
`wait_for_result`	`bool`	False
`poll_timeout`	`float`	120.0
`poll_interval`	`float`	2.0
`upload_filename`	`str`	''
`lora_names`	`list[str]`	[]
`lora_strengths`	`list[float]`	[]
`vae_name`	`str`	''
`node_class`	`str`	''
`allow_unvalidated`	`bool`	False
`safety_context`	`dict[str, Any]`	Field(default_factory=dict)
`backend_operation`	`str`	''
`backend_params`	`dict[str, Any]`	Field(default_factory=dict)

`ComfyUIOutput(BaseModel)`¶

Output schema for AdaptComfyUIBlock.

Field	Type	Default
`op`	`str`	required
`images_base64`	`list[str]`	Field(default_factory=list)
`output_path`	`str`	''
`prompt_id`	`str`	''
`status`	`str`	''
`queue_remaining`	`int`	0
`models`	`list[str]`	Field(default_factory=list)
`history`	`list[dict]`	Field(default_factory=list)
`metadata`	`dict`	Field(default_factory=dict)
`seed_used`	`int`	-1
`progress`	`dict`	Field(default_factory=dict)
`backend`	`str`	'comfyui'
`readiness`	`ReadinessReport \\| None`	None
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`degradation`	`ComfyUIDegradationNotice \\| None`	None
`completion_state`	`CompletionState`	'complete'
`warning_card`	`ComfyUIWarningCard \\| dict[str, Any] \\| None`	None
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`available_with`	`str`	''

`AdaptComfyUIMCPBlock(AIBlock[MCPComfyUIInput, MCPComfyUIOutput, dict])`¶

54-op ComfyUI MCP block with SQLite persistence.

Field	Type	Default
`name`	`str`	'adapt_comfyui_mcp'
`state`	`dict`	field(default_factory=dict)
`db_path`	`str`	field(default_factory=lambda: os.environ.get('COMFYUI_DB_PATH', _DEFAULT_DB))
`resource_bounds`	`ResourceBounds`	field(default_factory=ResourceBounds)
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: MCPComfyUIInput) -> Result[MCPComfyUIOutput]`¶

`MCPComfyUIRecord(BaseModel)`¶

A single metadata record returned from ComfyUIStore queries.

Field	Type	Default
`id`	`str`	required
`record_type`	`str`	required
`name`	`str`	required
`content`	`str`	required
`tags`	`str`	''
`timestamp`	`str`	''
`metadata`	`dict`	{}

`MCPComfyUIInput(BaseModel)`¶

Input to AdaptComfyUIMCPBlock -- 54 ops.

Field	Type	Default
`op`	Literal['submit_workflow', 'queue_prompt', 'get_history', 'get_image', 'list_models', 'get_status', 'interrupt', 'upload_image', 'txt2img', 'img2img', 'inpaint', 'upscale', 'controlnet', 'lora_txt2img', 'sdxl_txt2img', 'workflow_save', 'workflow_load', 'workflow_list', 'workflow_delete', 'preset_list', 'preset_apply', 'model_download', 'batch_generate', 'queue_list', 'queue_clear', 'stats', 'get_info', 'generate', 'wait_and_get', 'save_images', 'design_generate', 'design_variations', 'design_refine', 'design_from_reference', 'design_composite', 'brand_kit_create', 'brand_kit_get', 'brand_kit_list', 'brand_kit_update', 'format_preset_list', 'format_preset_get', 'format_preset_create', 'style_reference_save', 'style_reference_list', 'style_apply', 'project_create', 'project_get', 'project_list', 'project_add_asset', 'asset_save', 'asset_search', 'asset_export', 'design_recommend', 'list_patterns', 'readiness', 'precheck', 'capabilities', 'object_info', 'node_schema', 'submit_raw_workflow', 'backend_operation']	required
`workflow_json`	`str`	''
`prompt_text`	`str`	''
`negative_prompt`	`str`	''
`image_base64`	`str`	''
`mask_base64`	`str`	''
`model_name`	`str`	''
`width`	`int`	512
`height`	`int`	512
`steps`	`int`	20
`cfg_scale`	`float`	7.0
`seed`	`int`	-1
`sampler`	`str`	'euler'
`scheduler`	`str`	'normal'
`denoise`	`float`	1.0
`controlnet_model`	`str`	''
`controlnet_image_base64`	`str`	''
`batch_size`	`int`	1
`prompt_id`	`str`	''
`timeout_seconds`	`int`	120
`workflow_name`	`str`	''
`description`	`str`	''
`tags_csv`	`str`	''
`preset_name`	`str`	''
`preset_params`	`dict`	{}
`model_url`	`str`	''
`model_dest`	`str`	''
`prompts`	`list[str]`	[]
`query`	`str`	''
`limit`	`int`	50
`wait_for_result`	`bool`	False
`poll_timeout`	`float`	120.0
`poll_interval`	`float`	2.0
`upload_filename`	`str`	''
`lora_names`	`list[str]`	[]
`lora_strengths`	`list[float]`	[]
`vae_name`	`str`	''
`images_base64`	`list[str]`	[]
`output_dir`	`str`	''
`filename_template`	`str`	''
`brand_kit_id`	`str`	''
`brand_kit_name`	`str`	''
`brand_colors`	`list[str]`	[]
`style_prompt`	`str`	''
`preferred_loras`	`list[dict]`	[]
`format_preset`	`str`	''
`target_dpi`	`int`	72
`refinement_prompt`	`str`	''
`denoise_strength`	`float`	0.5
`num_variations`	`int`	4
`reference_image_base64`	`str`	''
`reference_description`	`str`	''
`project_name`	`str`	''
`client_name`	`str`	''
`brief_text`	`str`	''
`deliverables`	`list[dict]`	[]
`asset_name`	`str`	''
`asset_tags`	`str`	''
`job_id`	`str`	''
`layout_type`	`str`	''
`image_ids`	`list[str]`	[]
`brief`	`str`	''
`intent`	`str`	''
`aspect`	`str`	''
`use_case`	`str`	''
`node_class`	`str`	''
`allow_unvalidated`	`bool`	False
`safety_context`	`dict[str, Any]`	{}
`backend_operation`	`str`	''
`backend_params`	`dict[str, Any]`	{}
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`MCPComfyUIOutput(BaseModel)`¶

Output from AdaptComfyUIMCPBlock.

Field	Type	Default
`op`	`str`	required
`images_base64`	`list[str]`	[]
`output_path`	`str`	''
`prompt_id`	`str`	''
`prompt_ids`	`list[str]`	[]
`status`	`str`	''
`queue_remaining`	`int`	0
`models`	`list[str]`	[]
`history`	`list[dict]`	[]
`brand_kits`	`list[dict]`	[]
`projects`	`list[dict]`	[]
`workflows`	`list[dict]`	[]
`presets`	`list[dict]`	[]
`metadata`	`dict`	{}
`message`	`str`	''
`found`	`bool`	False
`count`	`int`	0
`seed_used`	`int`	-1
`progress`	`dict`	{}
`brand_kit`	`dict`	{}
`format_presets`	`list[dict]`	[]
`project`	`dict`	{}
`assets`	`list[dict]`	[]
`style_references`	`list[dict]`	[]
`generation_params`	`dict`	{}
`backend`	`str`	'comfyui'
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`degradation`	`dict`	{}
`available_with`	`list[str]`	[]
`readiness`	`dict`	{}
`job_status`	`str`	'complete'
`completion_state`	`str`	'qualified-draft'
`reliability_label`	`str`	'qualified-draft'
`warning_card`	`dict \\| None`	None
`evidence`	`dict`	{}
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`recommendation`	`dict`	{}
`llm_used`	`bool`	False

`ComfyUIStore`¶

SQLite-backed store for the adapt_comfyui MCP sub-package.

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`save_workflow(name: str, workflow_json: str, description: str = '', tags: str = '') -> str`¶

`load_workflow(name: str) -> dict | None`¶

`list_workflows(limit: int = 50) -> list[dict]`¶

`delete_workflow(name: str) -> bool`¶

`save_preset(name: str, params: dict, description: str = '', tags: str = '') -> str`¶

`get_preset(name: str) -> dict | None`¶

`list_presets(limit: int = 50) -> list[dict]`¶

`log_job(op: str, prompt_id: str = '', params: dict | None = None, status: str = 'queued') -> str`¶

`update_job_status(prompt_id: str, status: str, result: dict | None = None, error: str = '') -> None`¶

`get_job(prompt_id: str) -> dict | None`¶

`list_jobs(limit: int = 50) -> list[dict]`¶

`count_all() -> dict[str, int]`¶

`job_stats() -> dict[str, int]`¶

Count jobs by status.

`save_brand_kit(name: str, brand_colors: list[str] | None = None, style_prompt: str = '', negative_prompt: str = '', preferred_model: str = '', preferred_loras: list[dict] | None = None, metadata: dict | None = None) -> str`¶

`get_brand_kit(name: str = '', kit_id: str = '') -> dict | None`¶

`list_brand_kits(limit: int = 50) -> list[dict]`¶

`update_brand_kit(kit_id: str, **updates: Any) -> bool`¶

`save_style_reference(brand_kit_id: str, image_base64: str = '', description: str = '', tags: str = '') -> str`¶

`list_style_references(brand_kit_id: str, limit: int = 50) -> list[dict]`¶

`get_style_reference(ref_id: str) -> dict | None`¶

`save_project(name: str, client_name: str = '', brief_text: str = '', brand_kit_id: str = '', deliverables: list[dict] | None = None, metadata: dict | None = None) -> str`¶

`get_project(project_id: str) -> dict | None`¶

`list_projects(limit: int = 50) -> list[dict]`¶

`add_project_asset(project_id: str, asset_id: str) -> bool`¶

`save_asset(name: str = '', image_base64: str = '', job_id: str = '', project_id: str = '', brand_kit_id: str = '', format_preset: str = '', width: int = 0, height: int = 0, tags: str = '', metadata: dict | None = None) -> str`¶

`get_asset(asset_id: str) -> dict | None`¶

`search_assets(tags: str = '', project_id: str = '', brand_kit_id: str = '', format_preset: str = '', limit: int = 50) -> list[dict]`¶

MCP Tools¶

Operation	Source
`submit_workflow`	comfyui_mcp
`queue_prompt`	comfyui_mcp
`get_history`	comfyui_mcp
`get_image`	comfyui_mcp
`list_models`	comfyui_mcp
`get_status`	comfyui_mcp
`interrupt`	comfyui_mcp
`upload_image`	comfyui_mcp
`txt2img`	comfyui_mcp
`img2img`	comfyui_mcp
`inpaint`	comfyui_mcp
`upscale`	comfyui_mcp
`controlnet`	comfyui_mcp
`lora_txt2img`	comfyui_mcp
`sdxl_txt2img`	comfyui_mcp
`workflow_save`	comfyui_mcp
`workflow_load`	comfyui_mcp
`workflow_list`	comfyui_mcp
`workflow_delete`	comfyui_mcp
`preset_list`	comfyui_mcp
`preset_apply`	comfyui_mcp
`model_download`	comfyui_mcp
`batch_generate`	comfyui_mcp
`queue_list`	comfyui_mcp
`queue_clear`	comfyui_mcp
`stats`	comfyui_mcp
`get_info`	comfyui_mcp
`generate`	comfyui_mcp
`wait_and_get`	comfyui_mcp
`save_images`	comfyui_mcp
`design_generate`	comfyui_mcp
`design_variations`	comfyui_mcp
`design_refine`	comfyui_mcp
`design_from_reference`	comfyui_mcp
`design_composite`	comfyui_mcp
`brand_kit_create`	comfyui_mcp
`brand_kit_get`	comfyui_mcp
`brand_kit_list`	comfyui_mcp
`brand_kit_update`	comfyui_mcp
`format_preset_list`	comfyui_mcp
`format_preset_get`	comfyui_mcp
`format_preset_create`	comfyui_mcp
`style_reference_save`	comfyui_mcp
`style_reference_list`	comfyui_mcp
`style_apply`	comfyui_mcp
`project_create`	comfyui_mcp
`project_get`	comfyui_mcp
`project_list`	comfyui_mcp
`project_add_asset`	comfyui_mcp
`asset_save`	comfyui_mcp
`asset_search`	comfyui_mcp
`asset_export`	comfyui_mcp
`design_recommend`	comfyui_mcp
`list_patterns`	comfyui_mcp
`readiness`	comfyui_mcp
`precheck`	comfyui_mcp
`capabilities`	comfyui_mcp
`object_info`	comfyui_mcp
`node_schema`	comfyui_mcp
`submit_raw_workflow`	comfyui_mcp
`backend_operation`	comfyui_mcp

Operational Checklist¶

Before using adapt_comfyui in a production workflow:

Start ComfyUI and confirm get_status reports online for the configured COMFYUI_URL or default http://127.0.0.1:8188.
Run comfyui_readiness or comfyui_capabilities before generation in a new environment.
Install the checkpoint, LoRA, ControlNet, VAE, and upscale models referenced by the workflow or brand kit.
Install any custom ComfyUI nodes required by generated workflows, especially LoadImageFromBase64 and workflow-specific ControlNet or upscale nodes.
Use comfyui_object_info or comfyui_node_schema to confirm node availability before submitting raw workflows.
Keep allow_unvalidated=False for raw workflows unless an advanced caller has an explicit safety context and accepts the degraded validation envelope.
Run a representative generate or design_generate job in the target environment and verify returned images, dimensions, and asset persistence.
Document model paths and server startup commands alongside any pilot or customer workflow so the setup can be reproduced without founder intervention.

Adapt Comfyui¶

Overview¶

Public API¶

AdaptComfyUIBlock(AIBlock[ComfyUIInput, ComfyUIOutput, None])¶

infer(data: ComfyUIInput) -> Result[ComfyUIOutput]¶

ComfyUIInput(BaseModel)¶

ComfyUIOutput(BaseModel)¶

AdaptComfyUIMCPBlock(AIBlock[MCPComfyUIInput, MCPComfyUIOutput, dict])¶

infer(data: MCPComfyUIInput) -> Result[MCPComfyUIOutput]¶

MCPComfyUIRecord(BaseModel)¶

MCPComfyUIInput(BaseModel)¶

MCPComfyUIOutput(BaseModel)¶

ComfyUIStore¶

save_workflow(name: str, workflow_json: str, description: str = '', tags: str = '') -> str¶

load_workflow(name: str) -> dict | None¶

list_workflows(limit: int = 50) -> list[dict]¶

delete_workflow(name: str) -> bool¶

save_preset(name: str, params: dict, description: str = '', tags: str = '') -> str¶

get_preset(name: str) -> dict | None¶

list_presets(limit: int = 50) -> list[dict]¶

log_job(op: str, prompt_id: str = '', params: dict | None = None, status: str = 'queued') -> str¶

update_job_status(prompt_id: str, status: str, result: dict | None = None, error: str = '') -> None¶

get_job(prompt_id: str) -> dict | None¶

list_jobs(limit: int = 50) -> list[dict]¶

count_all() -> dict[str, int]¶

job_stats() -> dict[str, int]¶

save_brand_kit(name: str, brand_colors: list[str] | None = None, style_prompt: str = '', negative_prompt: str = '', preferred_model: str = '', preferred_loras: list[dict] | None = None, metadata: dict | None = None) -> str¶

get_brand_kit(name: str = '', kit_id: str = '') -> dict | None¶

list_brand_kits(limit: int = 50) -> list[dict]¶

update_brand_kit(kit_id: str, **updates: Any) -> bool¶

save_style_reference(brand_kit_id: str, image_base64: str = '', description: str = '', tags: str = '') -> str¶

list_style_references(brand_kit_id: str, limit: int = 50) -> list[dict]¶

get_style_reference(ref_id: str) -> dict | None¶

save_project(name: str, client_name: str = '', brief_text: str = '', brand_kit_id: str = '', deliverables: list[dict] | None = None, metadata: dict | None = None) -> str¶

get_project(project_id: str) -> dict | None¶

list_projects(limit: int = 50) -> list[dict]¶

add_project_asset(project_id: str, asset_id: str) -> bool¶

save_asset(name: str = '', image_base64: str = '', job_id: str = '', project_id: str = '', brand_kit_id: str = '', format_preset: str = '', width: int = 0, height: int = 0, tags: str = '', metadata: dict | None = None) -> str¶

get_asset(asset_id: str) -> dict | None¶

search_assets(tags: str = '', project_id: str = '', brand_kit_id: str = '', format_preset: str = '', limit: int = 50) -> list[dict]¶

MCP Tools¶

Operational Checklist¶

`AdaptComfyUIBlock(AIBlock[ComfyUIInput, ComfyUIOutput, None])`¶

`infer(data: ComfyUIInput) -> Result[ComfyUIOutput]`¶

`ComfyUIInput(BaseModel)`¶

`ComfyUIOutput(BaseModel)`¶

`AdaptComfyUIMCPBlock(AIBlock[MCPComfyUIInput, MCPComfyUIOutput, dict])`¶

`infer(data: MCPComfyUIInput) -> Result[MCPComfyUIOutput]`¶

`MCPComfyUIRecord(BaseModel)`¶

`MCPComfyUIInput(BaseModel)`¶

`MCPComfyUIOutput(BaseModel)`¶

`ComfyUIStore`¶

`save_workflow(name: str, workflow_json: str, description: str = '', tags: str = '') -> str`¶

`load_workflow(name: str) -> dict | None`¶

`list_workflows(limit: int = 50) -> list[dict]`¶

`delete_workflow(name: str) -> bool`¶

`save_preset(name: str, params: dict, description: str = '', tags: str = '') -> str`¶

`get_preset(name: str) -> dict | None`¶

`list_presets(limit: int = 50) -> list[dict]`¶

`log_job(op: str, prompt_id: str = '', params: dict | None = None, status: str = 'queued') -> str`¶

`update_job_status(prompt_id: str, status: str, result: dict | None = None, error: str = '') -> None`¶

`get_job(prompt_id: str) -> dict | None`¶

`list_jobs(limit: int = 50) -> list[dict]`¶

`count_all() -> dict[str, int]`¶

`job_stats() -> dict[str, int]`¶

`save_brand_kit(name: str, brand_colors: list[str] | None = None, style_prompt: str = '', negative_prompt: str = '', preferred_model: str = '', preferred_loras: list[dict] | None = None, metadata: dict | None = None) -> str`¶

`get_brand_kit(name: str = '', kit_id: str = '') -> dict | None`¶

`list_brand_kits(limit: int = 50) -> list[dict]`¶

`update_brand_kit(kit_id: str, **updates: Any) -> bool`¶

`save_style_reference(brand_kit_id: str, image_base64: str = '', description: str = '', tags: str = '') -> str`¶

`list_style_references(brand_kit_id: str, limit: int = 50) -> list[dict]`¶

`get_style_reference(ref_id: str) -> dict | None`¶

`save_project(name: str, client_name: str = '', brief_text: str = '', brand_kit_id: str = '', deliverables: list[dict] | None = None, metadata: dict | None = None) -> str`¶

`get_project(project_id: str) -> dict | None`¶

`list_projects(limit: int = 50) -> list[dict]`¶

`add_project_asset(project_id: str, asset_id: str) -> bool`¶

`save_asset(name: str = '', image_base64: str = '', job_id: str = '', project_id: str = '', brand_kit_id: str = '', format_preset: str = '', width: int = 0, height: int = 0, tags: str = '', metadata: dict | None = None) -> str`¶

`get_asset(asset_id: str) -> dict | None`¶

`search_assets(tags: str = '', project_id: str = '', brand_kit_id: str = '', format_preset: str = '', limit: int = 50) -> list[dict]`¶