Comfyui的官方示例：https://github.com/comfyanonymous/ComfyUI/blob/master/custom_nodes/example_node.py.example

一、代码

（ 逻辑：将图片对应掩膜覆盖的区域替换为黑色）
因为comfyui加载的图片后，转化为tensor

import torch
import numpy as np


class ImgCombMask:
    """
    自定义节点：将图像中的指定掩膜区域替换为黑色。

    功能说明：
    - 输入的图像和掩膜必须是张量格式（Tensor）。
    - 掩膜为二值化张量，白色区域（值 > 0.5）将被视为需要遮盖的区域。
    - 输出的图像会保持输入图像的形状。
    """

    @classmethod
    def INPUT_TYPES(cls):
        """
        定义节点的输入参数。
        """
        return {
            "required": {
                "image": ("IMAGE",),  # 图像输入 (Tensor)
                "mask": ("MASK",),    # 掩膜输入 (Tensor)
            }
        }

    CATEGORY = "Custom/Image Processing"
    RETURN_TYPES = ("IMAGE",)  # 返回值类型为图像
    FUNCTION = "apply_mask"    # 节点主函数

    def apply_mask(self, image, mask):
        """
        主逻辑：将掩膜覆盖的区域替换为黑色。

        参数:
        - image: 输入的图像张量，形状为 [Batch, Height, Width, Channels]。
        - mask: 输入的掩膜张量，形状为 [Batch, Height, Width] 或 [Batch, 1, Height, Width]。

        返回:
        - 处理后的图像张量，形状为 [Batch, Height, Width, Channels]。
        """
        # 确保图像和掩膜在同一个设备上（CPU/GPU）
        device = image.device
        mask = mask.to(device)

        # 调整图像维度为 [Batch, Channels, Height, Width]
        if len(image.shape) == 4:  # 如果图像格式为 [Batch, Height, Width, Channels]
            image = image.permute(0, 3, 1, 2)  # 转换为 [Batch, Channels, Height, Width]

        # 如果掩膜缺少通道维度，则添加通道维度
        if len(mask.shape) == 3:  # 掩膜格式为 [Batch, Height, Width]
            mask = mask.unsqueeze(1)  # 添加通道维度，变为 [Batch, 1, Height, Width]

        # 将掩膜二值化（值为 0 或 1）
        mask_binary = (mask > 0.5).float()  # 掩膜二值化：[Batch, 1, Height, Width]

        # 应用掩膜：将图像掩膜区域设置为黑色
        image = image * (1 - mask_binary)  # 被掩膜的区域会被设置为黑色

        # 恢复图像维度为 [Batch, Height, Width, Channels]
        image = image.permute(0, 2, 3, 1)  # 转换回 [Batch, Height, Width, Channels]

        return (image,)


# 注册自定义节点
NODE_CLASS_MAPPINGS = {
    "ImgCombMask": ImgCombMask
}

NODE_DISPLAY_NAME_MAPPINGS = {
    "ImgCombMask": "Image Combine with Mask to Black"
}

代码对应UI说明 (可改为任意自己喜欢)

Image Combine with Mask to Black 对应comfyui使用名

NODE_DISPLAY_NAME_MAPPINGS

= {
“ImgCombMask”: “Image Combine with Mask to Black”
}

CATEGORY

CATEGORY = "Custom/Image Processing"

附录 comfyui的官方示例代码

class Example:
    """
    示例节点

    类方法说明
    -------------
    INPUT_TYPES (dict):
        定义了主程序中节点的输入参数。
    IS_CHANGED:
        （可选）控制节点在输入参数未变化时是否仍然需要重新执行。

    属性说明
    ----------
    RETURN_TYPES (`tuple`):
        返回值类型的元组，定义节点输出的每个元素的类型。
    RETURN_NAMES (`tuple`):
        （可选）定义返回值中每个元素的名称，方便后续引用。
    FUNCTION (`str`):
        节点的入口方法名称。例如，如果 `FUNCTION = "execute"`，则会调用 `Example().execute()`。
    OUTPUT_NODE ([`bool`] 可选):
        表示该节点是否是输出节点。例如 SaveImage 节点就是一个输出节点。
        如果是输出节点，后端会遍历执行所有连接到此节点的父节点。
        如果未定义此属性，则默认为 False。
    CATEGORY (`str`):
        定义节点在 UI 中所属的分类。
    DEPRECATED (`bool`):
        指示该节点是否已弃用。弃用的节点会在 UI 中隐藏，但在已有的工作流中仍然可以使用。
    EXPERIMENTAL (`bool`):
        指示该节点是否为实验性节点。实验性节点在 UI 中会被标记，并且可能在未来版本中发生重大变化或被移除。
    execute(s) -> tuple || None:
        节点的入口方法名称。此方法名称必须与属性 `FUNCTION` 的值一致。
        例如，如果 `FUNCTION = "execute"`，则此方法必须命名为 `execute`；如果 `FUNCTION = "foo"`，则必须命名为 `foo`。
    """

    def __init__(self):
        pass

    @classmethod
    def INPUT_TYPES(s):
        """
        返回一个包含所有输入字段配置的字典。

        输入字段的类型（字符串）可以是以下几种：
        "MODEL", "VAE", "CLIP", "CONDITIONING", "LATENT", "IMAGE", "INT", "STRING", "FLOAT"。
        其中，"INT"、"STRING" 或 "FLOAT" 是特殊的值类型，通常用于节点的字段。

        返回值：`dict`:
            - 键 `input_fields_group`（`string`）: 可以是 required（必需）、hidden（隐藏）或 optional（可选）。
              节点类中必须包含 `required` 属性。
            - 值 `input_fields` (`dict`): 包含输入字段的具体配置：
                * 键 field_name (`string`): 入口方法参数的名称。
                * 值 field_config (`tuple`):
                    + 第一个值为字符串，表示字段类型，或者是一个供选择的列表。
                    + 第二个值是配置字典，用于字段类型 "INT"、"STRING" 或 "FLOAT" 的具体设置。
        """
        return {
            "required": {
                "image": ("IMAGE",),  # 输入图像类型
                "int_field": ("INT", {  # 整数类型字段
                    "default": 0,  # 默认值
                    "min": 0,  # 最小值
                    "max": 4096,  # 最大值
                    "step": 64,  # 滑块的步长
                    "display": "number",  # 外观显示：数字输入框或滑块
                    "lazy": True  # 只有在 check_lazy_status 方法中需要时才会被调用
                }),
                "float_field": ("FLOAT", {  # 浮点类型字段
                    "default": 1.0,
                    "min": 0.0,
                    "max": 10.0,
                    "step": 0.01,
                    "round": 0.001,  # 指定四舍五入的精度，可以设置为 False 以禁用
                    "display": "number",
                    "lazy": True
                }),
                "print_to_screen": (["enable", "disable"],),  # 下拉选项
                "string_field": ("STRING", {  # 字符串字段
                    "multiline": False,  # 如果为 True，则输入框支持多行文本
                    "default": "Hello World!",  # 默认值
                    "lazy": True
                }),
            },
        }

    RETURN_TYPES = ("IMAGE",)  # 返回值类型为图像
    # RETURN_NAMES = ("image_output_name",)  # （可选）为返回值命名

    FUNCTION = "test"  # 入口方法的名称

    # OUTPUT_NODE = False  # 是否为输出节点，默认为 False

    CATEGORY = "Example"  # 节点在 UI 中的分类

    def check_lazy_status(self, image, string_field, int_field, float_field, print_to_screen):
        """
        返回需要评估的输入字段名称列表。

        如果存在尚未评估的惰性输入字段，会调用此函数。
        只要返回的字段列表中包含尚未评估的字段，该函数就会再次被调用，直到所有字段评估完成。

        已评估的输入会作为参数传递给此函数；未评估的输入值将为 None。
        """
        if print_to_screen == "enable":
            return ["int_field", "float_field", "string_field"]
        else:
            return []

    def test(self, image, string_field, int_field, float_field, print_to_screen):
        """
        主逻辑方法，根据 FUNCTION 属性定义的入口方法。

        在此示例中，如果 print_to_screen 启用，则打印输入字段的信息。
        """
        if print_to_screen == "enable":
            print(f"""Your input contains:
                string_field aka input text: {string_field}
                int_field: {int_field}
                float_field: {float_field}
            """)
        # 示例逻辑：反转图像的像素值
        image = 1.0 - image
        return (image,)

    """
    节点会在任何输入更改时重新执行，但可以使用此方法强制节点即使输入未更改时也重新执行。
    通过返回一个数字或字符串，比较当前值和上一次执行返回的值，如果不同则节点会重新执行。
    此方法在核心代码中用于 LoadImage 节点，其中返回的是图像的哈希值。
    如果图像哈希在两次执行之间发生变化，LoadImage 节点会重新执行。
    """
    # @classmethod
    # def IS_CHANGED(s, image, string_field, int_field, float_field, print_to_screen):
    #     return ""

# 设置 Web 目录，任何在该目录中的 .js 文件都会被前端加载为扩展
# WEB_DIRECTORY = "./somejs"


# 添加自定义 API 路由，使用 router
from aiohttp import web
from server import PromptServer

@PromptServer.instance.routes.get("/hello")
async def get_hello(request):
    return web.json_response("hello")


# 导出节点的映射字典
# 注意：节点名称必须全局唯一
NODE_CLASS_MAPPINGS = {
    "Example": Example
}

# 映射节点的友好名称（在 UI 中显示）
NODE_DISPLAY_NAME_MAPPINGS = {
    "Example": "示例节点"
}