Custom Nodes

nodes

Custom nodes management — additive-only manifest system.

Provides a simple, reliable engine for managing ComfyUI custom nodes:

• Install missing bundled nodes (git clone).
• Update existing bundled nodes (git pull --ff-only).
• Never remove user-installed nodes.

Uses uv exclusively for package installs.

The manifest is defined in custom_nodes.json with the schema described by :class:NodeManifest and :class:NodeEntry.

NodeEntry

Bases: BaseModel

A single custom node definition in the manifest.

Attributes:

Name	Type	Description
name	str	Directory name for the node (used as clone target).
url	str	Git repository URL.
tier	str	Bundle tier — "minimal", "umeairt", or "full" (default). Hierarchical: each tier includes nodes from all lower tiers.
requirements	str | None	Relative path to a requirements.txt inside the cloned directory (e.g. "requirements.txt").
subfolder	str | None	If set, the node lives inside another node's directory (e.g. Impact-Subpack inside Impact-Pack).
note	str | None	Human-readable comment (not used by the installer).

NodeManifest

Bases: BaseModel

The custom_nodes.json manifest.

filter_by_tier(manifest, tier)

Return a new manifest containing only nodes for the given tier.

Each tier includes all nodes from lower tiers::

minimal ⊂ umeairt ⊂ full

Parameters:

Name	Type	Description	Default
manifest	NodeManifest	The full manifest.	required
tier	str	One of "minimal", "umeairt", "full".	required

Returns:

Type	Description
NodeManifest	A filtered NodeManifest.

Source code in src/installer/nodes.py

def filter_by_tier(manifest: NodeManifest, tier: str) -> NodeManifest:
    """Return a new manifest containing only nodes for the given tier.

    Each tier includes all nodes from lower tiers::

        minimal ⊂ umeairt ⊂ full

    Args:
        manifest: The full manifest.
        tier: One of ``"minimal"``, ``"umeairt"``, ``"full"``.

    Returns:
        A filtered NodeManifest.
    """
    try:
        tier_key = NodeTier(tier) if isinstance(tier, str) else tier
    except ValueError:
        tier_key = NodeTier.FULL
    allowed = TIER_HIERARCHY.get(tier_key, TIER_HIERARCHY[NodeTier.FULL])
    filtered = [n for n in manifest.nodes if n.tier in allowed]
    return NodeManifest(nodes=filtered)

load_manifest(path)

Load and validate the custom_nodes.json manifest.

Parameters:

Name	Type	Description	Default
path	Path	Absolute path to the JSON manifest file.	required

Returns:

Name	Type	Description
Validated	NodeManifest	class:NodeManifest instance.

Raises:

Type	Description
FileNotFoundError	If path does not exist.

Source code in src/installer/nodes.py

def load_manifest(path: Path) -> NodeManifest:
    """Load and validate the ``custom_nodes.json`` manifest.

    Args:
        path: Absolute path to the JSON manifest file.

    Returns:
        Validated :class:`NodeManifest` instance.

    Raises:
        FileNotFoundError: If *path* does not exist.
    """
    if not path.exists():
        raise FileNotFoundError(f"Node manifest not found: {path}")

    with open(path, encoding="utf-8") as f:
        data = json.load(f)

    return NodeManifest.model_validate(data)

install_node(node, custom_nodes_dir, python_exe, log)

Install a single custom node via git clone.

Clones the repository into custom_nodes_dir/node.name (or node.subfolder if specified). On network failure, retries up to 3 times with shallow clone.

After cloning, installs pip requirements if node.requirements is set.

Parameters:

Name	Type	Description	Default
node	NodeEntry	Node definition from the manifest.	required
custom_nodes_dir	Path	ComfyUI/custom_nodes/ directory.	required
python_exe	Path	Path to the venv Python executable.	required
log	InstallerLogger	Installer logger for user-facing messages.	required

Returns:

Type	Description
bool	True if the node was installed (or already existed).

Source code in src/installer/nodes.py

def install_node(
    node: NodeEntry,
    custom_nodes_dir: Path,
    python_exe: Path,
    log: InstallerLogger,
) -> bool:
    """Install a single custom node via ``git clone``.

    Clones the repository into ``custom_nodes_dir/node.name``
    (or ``node.subfolder`` if specified). On network failure,
    retries up to 3 times with shallow clone.

    After cloning, installs pip requirements if ``node.requirements``
    is set.

    Args:
        node: Node definition from the manifest.
        custom_nodes_dir: ``ComfyUI/custom_nodes/`` directory.
        python_exe: Path to the venv Python executable.
        log: Installer logger for user-facing messages.

    Returns:
        ``True`` if the node was installed (or already existed).
    """
    # Handle subfolder nodes (e.g. Impact-Subpack inside Impact-Pack)
    node_dir = custom_nodes_dir / node.subfolder if node.subfolder else custom_nodes_dir / node.name

    if node_dir.exists():
        log.sub(f"  {node.name}: already installed", style="success")
        return True

    # Clone with retry (network can be flaky)
    log.sub(f"  {node.name}: cloning...", style="cyan")
    max_retries = 3
    for attempt in range(1, max_retries + 1):
        try:
            clone_args = ["clone", node.url, str(node_dir)]
            if attempt > 1:
                # Shallow clone on retry to reduce data
                clone_args = ["clone", "--depth", "1", node.url, str(node_dir)]
                log.sub(f"  {node.name}: retry {attempt}/{max_retries} (shallow)...", style="yellow")
            run_and_log("git", clone_args, timeout=300)
            break  # Success
        except CommandError as e:
            # Clean up partial clone before retry
            if node_dir.exists():
                import shutil
                shutil.rmtree(node_dir, ignore_errors=True)
            if attempt == max_retries:
                log.sub(f"  {node.name}: clone FAILED after {max_retries} attempts ({e})", style="red")
                return False

    # Install requirements
    if node.requirements:
        req_file = node_dir / node.requirements
        if req_file.exists():
            _pip_install_requirements(python_exe, req_file, log)

    return True

update_node(node, custom_nodes_dir, python_exe, log)

Update an existing custom node via git pull --ff-only.

If the node is not installed, delegates to :func:install_node. Re-installs pip requirements after pulling in case they changed.

Parameters:

Name	Type	Description	Default
node	NodeEntry	Node definition from the manifest.	required
custom_nodes_dir	Path	ComfyUI/custom_nodes/ directory.	required
python_exe	Path	Path to the venv Python executable.	required
log	InstallerLogger	Installer logger for user-facing messages.	required

Returns:

Type	Description
bool	True if updated successfully.

Source code in src/installer/nodes.py

def update_node(
    node: NodeEntry,
    custom_nodes_dir: Path,
    python_exe: Path,
    log: InstallerLogger,
) -> bool:
    """Update an existing custom node via ``git pull --ff-only``.

    If the node is not installed, delegates to :func:`install_node`.
    Re-installs pip requirements after pulling in case they changed.

    Args:
        node: Node definition from the manifest.
        custom_nodes_dir: ``ComfyUI/custom_nodes/`` directory.
        python_exe: Path to the venv Python executable.
        log: Installer logger for user-facing messages.

    Returns:
        ``True`` if updated successfully.
    """
    node_dir = custom_nodes_dir / node.subfolder if node.subfolder else custom_nodes_dir / node.name

    if not node_dir.exists():
        # Not installed — install it
        return install_node(node, custom_nodes_dir, python_exe, log)

    # Git pull
    log.sub(f"  {node.name}: updating...", style="cyan")
    try:
        run_and_log(
            "git", ["-C", str(node_dir), "pull", "--ff-only"],
            ignore_errors=True,
            timeout=60,
        )
    except CommandError:
        log.sub(f"  {node.name}: pull failed (may have local changes)", style="yellow")

    # Re-install requirements (in case they changed)
    if node.requirements:
        req_file = node_dir / node.requirements
        if req_file.exists():
            _pip_install_requirements(python_exe, req_file, log)

    return True

install_all_nodes(manifest, custom_nodes_dir, python_exe, log)

Install all nodes from the manifest (additive-only).

Only installs missing nodes — existing nodes and user-installed nodes are left untouched. Required nodes are processed first.

Parameters:

Name	Type	Description	Default
manifest	NodeManifest	Validated node manifest.	required
custom_nodes_dir	Path	ComfyUI/custom_nodes/ directory.	required
python_exe	Path	Path to the venv Python executable.	required
log	InstallerLogger	Installer logger for user-facing messages.	required

Returns:

Type	Description
tuple[int, int]	Tuple of (success_count, fail_count).

Source code in src/installer/nodes.py

def install_all_nodes(
    manifest: NodeManifest,
    custom_nodes_dir: Path,
    python_exe: Path,
    log: InstallerLogger,
) -> tuple[int, int]:
    """Install all nodes from the manifest (additive-only).

    Only installs missing nodes — existing nodes and user-installed
    nodes are left untouched. Required nodes are processed first.

    Args:
        manifest: Validated node manifest.
        custom_nodes_dir: ``ComfyUI/custom_nodes/`` directory.
        python_exe: Path to the venv Python executable.
        log: Installer logger for user-facing messages.

    Returns:
        Tuple of ``(success_count, fail_count)``.
    """
    log.item(f"{len(manifest.nodes)} nodes in manifest")

    custom_nodes_dir.mkdir(parents=True, exist_ok=True)
    success = 0
    fail = 0

    # Install required nodes first
    required = [n for n in manifest.nodes if n.required]
    optional = [n for n in manifest.nodes if not n.required]

    for node in required + optional:
        if install_node(node, custom_nodes_dir, python_exe, log):
            success += 1
        else:
            fail += 1

    log.item(f"Nodes: {success} OK, {fail} failed", style="success" if fail == 0 else "yellow")
    return success, fail

update_all_nodes(manifest, custom_nodes_dir, python_exe, log)

Update all bundled nodes. User-installed nodes are NEVER touched.

Identifies user-installed nodes (present on disk but not in the manifest) and preserves them. Then updates each manifest node via :func:update_node.

Parameters:

Name	Type	Description	Default
manifest	NodeManifest	Validated node manifest.	required
custom_nodes_dir	Path	ComfyUI/custom_nodes/ directory.	required
python_exe	Path	Path to the venv Python executable.	required
log	InstallerLogger	Installer logger for user-facing messages.	required

Returns:

Type	Description
tuple[int, int]	Tuple of (success_count, fail_count).

Source code in src/installer/nodes.py

def update_all_nodes(
    manifest: NodeManifest,
    custom_nodes_dir: Path,
    python_exe: Path,
    log: InstallerLogger,
) -> tuple[int, int]:
    """Update all bundled nodes. User-installed nodes are NEVER touched.

    Identifies user-installed nodes (present on disk but not in
    the manifest) and preserves them. Then updates each manifest
    node via :func:`update_node`.

    Args:
        manifest: Validated node manifest.
        custom_nodes_dir: ``ComfyUI/custom_nodes/`` directory.
        python_exe: Path to the venv Python executable.
        log: Installer logger for user-facing messages.

    Returns:
        Tuple of ``(success_count, fail_count)``.
    """
    log.step("Updating Custom Nodes")

    # List all directories in custom_nodes (ignore __pycache__ and hidden dirs)
    installed_names = {
        d.name for d in custom_nodes_dir.iterdir()
        if d.is_dir() and d.name != "__pycache__" and not d.name.startswith(".")
    } if custom_nodes_dir.exists() else set()
    manifest_names = {n.name for n in manifest.nodes}

    # User-installed nodes = installed but not in manifest
    user_nodes = installed_names - manifest_names
    if user_nodes:
        log.item(f"{len(user_nodes)} user-installed node(s) preserved:", style="success")
        for name in sorted(user_nodes):
            log.sub(f"  {name}", style="dim")

    # Update bundled nodes
    log.item(f"Updating {len(manifest.nodes)} bundled nodes...")
    success = 0
    fail = 0

    for node in manifest.nodes:
        if update_node(node, custom_nodes_dir, python_exe, log):
            success += 1
        else:
            fail += 1

    log.item(f"Updated: {success} OK, {fail} failed", style="success" if fail == 0 else "yellow")
    return success, fail