ComfyUI工作流自动化服务器

# Run directly with uvx
uvx comfyui-easy-mcp

# Or install globally
uv pip install comfyui-easy-mcp

git clone https://github.com/IO-AtelierTech/comfyui-mcp.git
cd comfyui-mcp
uv sync

export COMFY_URL=http://localhost:8188
export COMFY_WORKFLOWS_DIR=/path/to/workflows-api      # API format for execution
export COMFY_WORKFLOWS_UI_DIR=/path/to/workflows-ui    # UI format for editor
export COMFY_WORKFLOW_JSON_FILE=/path/to/workflows-api/default.json
export PROMPT_NODE_ID=6
export OUTPUT_NODE_ID=9
export OUTPUT_MODE=file

{
  "mcpServers": {
    "ComfyUI": {
      "command": "uvx",
      "args": ["comfyui-easy-mcp"],
      "env": {
        "COMFY_URL": "http://localhost:8188",
        "COMFY_WORKFLOWS_DIR": "/path/to/workflows-api",
        "COMFY_WORKFLOWS_UI_DIR": "/path/to/workflows-ui",
        "COMFY_WORKFLOW_JSON_FILE": "/path/to/workflows-api/default.json",
        "PROMPT_NODE_ID": "6",
        "OUTPUT_NODE_ID": "9"
      }
    }
  }
}

# 1. CREATE: Start with empty workflow or template
wf = create_workflow()
# Or use a template:
wf = get_workflow_template("fal-flux-dev")

# 2. BUILD: Add nodes with explicit connections
wf = add_node(wf, "1", "LoadImage", {"image": "input.jpg"})
wf = add_node(wf, "2", "LumaImageToVideoNode", {
    "prompt": "smooth camera motion",
    "model": "ray-2",
    "first_image": ["1", 0],  # Connect to node "1", output index 0
    "resolution": "1080p",
    "duration": "5s"
})
wf = add_node(wf, "3", "SaveVideo_fal", {
    "videos": ["2", 0],
    "filename_prefix": "output"
})

# 3. VALIDATE: Check structure and node types before saving
validation = validate_workflow(wf)
# Returns: {"valid": true/false, "errors": [...], "warnings": [...]}

# 4. SAVE: Choose format based on purpose
# For MCP execution (API format):
save_workflow(wf, "my-workflow", format="api")  # → workflows-api/my-workflow.json

# For ComfyUI editor (UI format):
save_workflow(wf, "my-workflow", format="ui")   # → workflows-ui/my-workflow.json

# Connect node "1" output slot 0 to this input
wf = add_node(wf, "2", "CLIPTextEncode", {
    "text": "my prompt",
    "clip": ["1", 0]  # ← Connection to node "1", first output
})

# Find available nodes
nodes = list_nodes(filter="Luma")  # → ["LumaImageToVideoNode", ...]

# Get node details
info = get_node_info("LumaImageToVideoNode")
# Returns:
# {
#   "input": {
#     "required": {
#       "prompt": ["STRING", {...}],
#       "model": [["ray-2", "ray-flash-2"], {...}],
#       "first_image": ["IMAGE"],
#       ...
#     },
#     "optional": {...}
#   },
#   "output": ["VIDEO"],
#   ...
# }

# Using default workflow configuration
result = generate_image("a cyberpunk city at sunset")

# Execute saved workflow with custom inputs
result = run_workflow(
    "flux-dev.json",
    inputs={"6": {"text": "a forest landscape"}},
    output_node_id="9"
)

# 1. Create empty workflow
wf = create_workflow()

# 2. Add nodes
wf = add_node(wf, "1", "CheckpointLoaderSimple", {
    "ckpt_name": "flux-dev.safetensors"
})
wf = add_node(wf, "2", "CLIPTextEncode", {
    "text": "beautiful sunset over mountains",
    "clip": ["1", 1]  # Connect to checkpoint's CLIP output
})
wf = add_node(wf, "3", "KSampler", {
    "model": ["1", 0],
    "positive": ["2", 0],
    # ... other inputs
})

# 3. Validate before execution
validation = validate_workflow(wf)
if not validation.get("valid"):
    print(f"Errors: {validation.get('errors')}")

# 4. Execute
result = execute_workflow(wf, output_node_id="9")

# Submit without waiting
submission = submit_workflow(workflow)
prompt_id = submission["prompt_id"]

# Check status later
status = get_prompt_status(prompt_id)

# Get result when ready
if status["completed"]:
    image = get_result_image(prompt_id, "9")

# List all fal.ai nodes
nodes = list_nodes(filter="fal")

# Get details about a specific node
info = get_node_info("RemoteCheckpointLoader_fal")

# 1. Verify fal.ai extension is loaded
extensions = list_extensions()
# Should include "ComfyUI-fal-Connector"

# 2. List available fal.ai nodes
fal_nodes = list_nodes(filter="fal")
# Returns: ["RemoteCheckpointLoader_fal", "StringInput_fal", "SaveImage_fal", ...]

# 3. Get node details
info = get_node_info("RemoteCheckpointLoader_fal")
# Shows available checkpoints: flux-dev, flux-schnell, sd3.5-large, etc.

# 4. Build a fal.ai workflow
wf = create_workflow()
wf = add_node(wf, "1", "RemoteCheckpointLoader_fal", {
    "ckpt_name": "flux-dev"
})
wf = add_node(wf, "2", "StringInput_fal", {
    "text": "a futuristic city at sunset, cyberpunk style"
})
wf = add_node(wf, "3", "CLIPTextEncode", {
    "text": ["2", 0],
    "clip": ["1", 1]
})
# ... add sampler, VAE decode, save image nodes

# 5. Execute on fal.ai cloud GPUs
result = execute_workflow(wf, output_node_id="9")

export FAL_KEY=your-fal-api-key

comfy_mcp_server/
├── __init__.py      # Entry point, FastMCP server
├── api.py           # HTTP helpers, ComfyUI API functions
├── models.py        # Pydantic models for type safety
├── settings.py      # Configuration with pydantic-settings
├── compat.py        # Version compatibility layer
└── tools/           # MCP tool implementations
    ├── system.py    # System monitoring tools
    ├── discovery.py # Node/model discovery
    ├── workflow.py  # Workflow management
    └── execution.py # Workflow execution

# Install dev dependencies
uv sync --all-extras

# Run linting
uv run ruff check src/

# Run tests (requires running ComfyUI)
uv run pytest tests/

# Format code
uv run ruff format src/