comfyui-node-lifecycle

class RandomNode(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="RandomNode",
            display_name="Random Value",
            category="utils",
            inputs=[
                io.Float.Input("min_val", default=0.0),
                io.Float.Input("max_val", default=1.0),
            ],
            outputs=[io.Float.Output("FLOAT")],
        )
 
    @classmethod
    def fingerprint_inputs(cls, min_val, max_val):
        """Return value compared to last run. Different value = re-execute."""
        # Return unique value each time to always re-execute
        import time
        return time.time()
 
    @classmethod
    def execute(cls, min_val, max_val):
        import random
        return io.NodeOutput(random.uniform(min_val, max_val))

@classmethod
def IS_CHANGED(s, min_val, max_val):
    return time.time()  # always re-execute

io.Schema(
    node_id="AlwaysRunNode",
    not_idempotent=True,  # prevents cache sharing between instances of the same node
    # ...
)

io.Schema(
    node_id="InteractiveCropNode",
    has_intermediate_output=True,
    # ...
)

from comfy_api.latest import Caching
 
class MyCacheProvider(Caching.CacheProvider):
    async def on_lookup(self, context):   # context: node_id, class_type, cache_key_hash
        ...  # return Caching.CacheValue(outputs=[...], ui={...}) or None on miss
 
    async def on_store(self, context, value):
        ...  # store to external storage (dispatched as a background task)
 
    def should_cache(self, context, value=None) -> bool:
        return True  # return False to skip external caching for a node
 
    def on_prompt_start(self, prompt_id): ...
    def on_prompt_end(self, prompt_id): ...
 
# Register in ComfyExtension.on_load():
api = ComfyAPI()
await api.caching.register_provider(MyCacheProvider())

class ValidatedNode(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="ValidatedNode",
            display_name="Validated Node",
            category="utils",
            inputs=[
                io.Int.Input("width", default=512, min=1, max=8192),
                io.Int.Input("height", default=512, min=1, max=8192),
            ],
            outputs=[io.Image.Output("IMAGE")],
        )
 
    @classmethod
    def validate_inputs(cls, width, height):
        """Return True if valid, or error string if invalid."""
        if width % 8 != 0 or height % 8 != 0:
            return "Width and height must be multiples of 8"
        if width * height > 4096 * 4096:
            return "Total pixels exceed maximum (4096x4096)"
        return True
 
    @classmethod
    def execute(cls, width, height):
        import torch
        return io.NodeOutput(torch.zeros(1, height, width, 3))

@classmethod
def VALIDATE_INPUTS(s, width, height):
    if width % 8 != 0:
        return "Width must be a multiple of 8"
    return True

@classmethod
def validate_inputs(cls, input_types: dict = None, **kwargs):
    # input_types contains the actual types of connected inputs
    # Returning True skips the default type checking
    return True

@classmethod
def check_lazy_status(cls, condition, value_a=None, value_b=None):
    """Called before execute. Return names of inputs that need evaluation."""
    if condition and value_a is None:
        return ["value_a"]
    if not condition and value_b is None:
        return ["value_b"]
    return []

class SaveMyData(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="SaveMyData",
            display_name="Save Data",
            category="output",
            is_output_node=True,  # marks as output node
            inputs=[
                io.String.Input("data"),
                io.String.Input("filename", default="output.txt"),
            ],
            outputs=[],  # output nodes may have no outputs
            hidden=[io.Hidden.prompt, io.Hidden.extra_pnginfo],
        )
 
    @classmethod
    def execute(cls, data, filename):
        import folder_paths, os
        output_dir = folder_paths.get_output_directory()
        with open(os.path.join(output_dir, filename), 'w') as f:
            f.write(data)
        return io.NodeOutput()

# V3: is_input_list=True in Schema (same as V1 INPUT_IS_LIST)
# All inputs arrive as lists — including widget values like batch_size
# Widget values: use widget_value[0] to get the scalar
# Shorter lists are padded by repeating the last value
 
# V1: INPUT_IS_LIST = True to receive full lists
class ListNode:
    INPUT_IS_LIST = True
    # Now execute() receives lists instead of individual items

# V3
io.Image.Output("IMAGE", is_output_list=True)
 
# V1
OUTPUT_IS_LIST = (True,)  # tuple matching RETURN_TYPES

@classmethod
def execute(cls, image, model):
    try:
        result = model.process(image)
    except RuntimeError as e:
        if "out of memory" in str(e):
            import torch
            torch.cuda.empty_cache()
            # Try with smaller batch
            result = process_in_chunks(image, model)
        else:
            raise
    return io.NodeOutput(result)

from server import PromptServer
 
@classmethod
def execute(cls, data):
    PromptServer.instance.send_sync(
        "my_extension.status",
        {"message": "Processing complete", "progress": 100}
    )
    return io.NodeOutput(data)

import time
import torch
from comfy_api.latest import ComfyExtension, io, ComfyAPISync
 
class FullLifecycleNode(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="FullLifecycleNode",
            display_name="Full Lifecycle Demo",
            category="example",
            inputs=[
                io.Image.Input("image"),
                io.Float.Input("threshold", default=0.5, min=0.0, max=1.0),
                io.Image.Input("optional_ref", optional=True, lazy=True),
            ],
            outputs=[
                io.Image.Output("IMAGE"),
                io.Mask.Output("MASK"),
            ],
            hidden=[io.Hidden.unique_id],
        )
 
    @classmethod
    def validate_inputs(cls, image, threshold, optional_ref=None):
        if threshold == 0.0:
            return "Threshold cannot be exactly 0"
        return True
 
    @classmethod
    def fingerprint_inputs(cls, image, threshold, optional_ref=None):
        # Re-execute if threshold changed; cache otherwise
        return threshold
 
    @classmethod
    def check_lazy_status(cls, image, threshold, optional_ref=None):
        # Only request optional_ref if threshold is high
        if threshold > 0.8 and optional_ref is None:
            return ["optional_ref"]
        return []
 
    @classmethod
    def execute(cls, image, threshold, optional_ref=None):
        node_id = cls.hidden.unique_id
 
        api = ComfyAPISync()  # use ComfyAPISync in sync execute; ComfyAPI in async
        api.execution.set_progress(0, 2)
 
        # Generate mask from threshold
        gray = image[:, :, :, 0] * 0.299 + image[:, :, :, 1] * 0.587 + image[:, :, :, 2] * 0.114
        mask = (gray > threshold).float()
 
        api.execution.set_progress(1, 2)
 
        # Apply mask
        result = image * mask.unsqueeze(-1)
        if optional_ref is not None:
            result = result + optional_ref * (1 - mask.unsqueeze(-1))
 
        api.execution.set_progress(2, 2)
        return io.NodeOutput(result, mask)