"""The published ComfyUI V2 Python API — the complete conversion surface. Generated from the live public ComfyUI API modules. If a member is not here it does not exist: do not call it, and report the gap naming what is missing. Reached from a converted pack as: from comfy_api.latest import io, sdk """ from typing import Any # ─── comfy_api.latest.io ────────────────────────────────────────────────────────── class Accumulation(ComfyTypeIO): io_type: Any = ... class AnyType(ComfyTypeIO): io_type: Any = ... class Array(ComfyTypeIO): io_type: Any = ... class Audio(ComfyTypeIO): io_type: Any = ... class AudioEncoder(ComfyTypeIO): io_type: Any = ... class AudioEncoderOutput(ComfyTypeIO): io_type: Any = ... class Autogrow(ComfyTypeI): io_type: Any = ... class BBOX(ComfyTypeIO): io_type: Any = ... class BackgroundRemoval(ComfyTypeIO): io_type: Any = ... class Boolean(ComfyTypeIO): io_type: Any = ... class BoundingBox(ComfyTypeIO): io_type: Any = ... class BoundingBoxes(ComfyTypeIO): io_type: Any = ... class Clip(ComfyTypeIO): io_type: Any = ... class ClipVision(ComfyTypeIO): io_type: Any = ... class ClipVisionOutput(ComfyTypeIO): io_type: Any = ... class Color(ComfyTypeIO): io_type: Any = ... class Colors(ComfyTypeIO): io_type: Any = ... class Combo(ComfyTypeIO): io_type: Any = ... class ComfyNode(_ComfyNodeBaseInternal): """Common base class for all V3 nodes.""" @classmethod # DO NOT override this class. Will break things in execution.py. def GET_BASE_CLASS(cls): ... @classmethod # Optionally, define this function to return a list of input names that should be evaluated. def check_lazy_status(cls, **kwargs) -> list[str]: ... @classmethod # Override this function with one that returns a Schema instance. def define_schema(cls) -> Schema: ... @classmethod # Override this function with one that performs node's actions. def execute(cls, **kwargs) -> NodeOutput: ... @classmethod # Optionally, define this function to fingerprint inputs; equivalent to V1's IS_CHANGED. def fingerprint_inputs(cls, **kwargs) -> Any: ... @classmethod # Optionally, define this function to validate inputs; equivalent to V1's VALIDATE_INPUTS. def validate_inputs(cls, **kwargs) -> bool | str: ... class ComfyTypeI(_ComfyType): """ComfyType subclass that only has a default Input class - intended for types that only have Inputs.""" ... class ComfyTypeIO(ComfyTypeI): """ComfyType subclass that has default Input and Output classes; useful for types with both Inputs and Outputs.""" ... class Conditioning(ComfyTypeIO): io_type: Any = ... class ControlAfterGenerate(str, Enum): decrement: Any = ... fixed: Any = ... increment: Any = ... randomize: Any = ... class ControlNet(ComfyTypeIO): io_type: Any = ... class Curve(ComfyTypeIO): io_type: Any = ... # Create a ComfyType for a custom io_type. def Custom(io_type: str) -> type[ComfyTypeIO]: ... class Dict(ComfyTypeIO): io_type: Any = ... class DynamicCombo(ComfyTypeI): io_type: Any = ... class FaceAnalysis(ComfyTypeIO): io_type: Any = ... class File3DAny(ComfyTypeIO): """General 3D file type - accepts any supported 3D format.""" io_type: Any = ... class File3DFBX(ComfyTypeIO): """FBX format 3D file - best for game engines and animation.""" io_type: Any = ... class File3DGLB(ComfyTypeIO): """GLB format 3D file - binary glTF, best for web and cross-platform.""" io_type: Any = ... class File3DGLTF(ComfyTypeIO): """GLTF format 3D file - JSON-based glTF with external resources.""" io_type: Any = ... class File3DKSPLAT(ComfyTypeIO): """KSPLAT format 3D file - 3D Gaussian splat.""" io_type: Any = ... class File3DOBJ(ComfyTypeIO): """OBJ format 3D file - simple geometry format.""" io_type: Any = ... class File3DPLY(ComfyTypeIO): """PLY format 3D file - point cloud or Gaussian splat.""" io_type: Any = ... class File3DPointCloudAny(ComfyTypeIO): """General point cloud file type - accepts any supported point cloud container (currently .ply).""" io_type: Any = ... class File3DSPLAT(ComfyTypeIO): """SPLAT format 3D file - 3D Gaussian splat.""" io_type: Any = ... class File3DSPZ(ComfyTypeIO): """SPZ format 3D file - compressed 3D Gaussian splat.""" io_type: Any = ... class File3DSTL(ComfyTypeIO): """STL format 3D file - best for 3D printing.""" io_type: Any = ... class File3DSplatAny(ComfyTypeIO): """General 3D Gaussian splat file type - accepts any supported splat container (.ply / .spz / .splat / .ksplat).""" io_type: Any = ... class File3DUSDZ(ComfyTypeIO): """USDZ format 3D file - Apple AR format.""" io_type: Any = ... class Float(ComfyTypeIO): io_type: Any = ... class FlowControl(ComfyTypeIO): io_type: Any = ... class FolderType(str, Enum): input: Any = ... output: Any = ... temp: Any = ... class Gligen(ComfyTypeIO): """ModelPatcher that wraps around a 'Gligen' model.""" io_type: Any = ... class Guider(ComfyTypeIO): io_type: Any = ... class Hidden(str, Enum): """Enumerator for requesting hidden variables in nodes.""" api_key_comfy_org: Any = ... auth_token_comfy_org: Any = ... comfy_usage_source: Any = ... dynprompt: Any = ... extra_pnginfo: Any = ... prompt: Any = ... unique_id: Any = ... class HiddenHolder: def __init__(self, unique_id: str, prompt: Any, extra_pnginfo: Any, dynprompt: Any, auth_token_comfy_org: str, api_key_comfy_org: str, comfy_usage_source: str = None, **kwargs): ... @classmethod def from_dict(cls, d: dict | None): ... @classmethod def from_v3_data(cls, v3_data: V3Data | None) -> HiddenHolder: ... class Histogram(ComfyTypeIO): """A histogram represented as a list of bin counts.""" io_type: Any = ... class HookKeyframes(ComfyTypeIO): io_type: Any = ... class Hooks(ComfyTypeIO): io_type: Any = ... class Image(ComfyTypeIO): io_type: Any = ... class ImageCompare(ComfyTypeI): io_type: Any = ... class Input(_IO_V3): """Base class for a V3 Input.""" def __init__(self, id: str, display_name: str = None, optional=False, tooltip: str = None, lazy: bool = None, extra_dict=None, raw_link: bool = None, advanced: bool = None): ... def as_dict(self): ... def get_all(self) -> list[Input]: ... def get_io_type(self): ... class Int(ComfyTypeIO): io_type: Any = ... class Latent(ComfyTypeIO): """Latents are stored as a dictionary.""" io_type: Any = ... class LatentOperation(ComfyTypeIO): io_type: Any = ... class LatentUpscaleModel(ComfyTypeIO): io_type: Any = ... class Load3D(ComfyTypeIO): """3D models are stored as a dictionary.""" io_type: Any = ... class Load3DAnimation(Load3D): io_type: Any = ... class Load3DCamera(ComfyTypeIO): io_type: Any = ... class Load3DModelInfo(ComfyTypeIO): io_type: Any = ... class LoraModel(ComfyTypeIO): io_type: Any = ... class LossMap(ComfyTypeIO): io_type: Any = ... MAX_RESOLUTION: Any = ... # = 16384 class Mask(ComfyTypeIO): io_type: Any = ... class MatchType(ComfyTypeIO): io_type: Any = ... class Mesh(ComfyTypeIO): io_type: Any = ... class Model(ComfyTypeIO): io_type: Any = ... class ModelPatch(ComfyTypeIO): io_type: Any = ... class MultiCombo(ComfyTypeI): """Multiselect Combo input (dropdown for selecting potentially more than one value).""" io_type: Any = ... class MultiType(MultiType, ComfyTypeIO): io_type: Any = ... class NodeInfoV1: """NodeInfoV1(input: 'dict' = None, input_order: 'dict[str, list[str]]' = None, is_input_list: 'bool' = None, output: 'list[str]' = None, output_is_list: 'list[bool]' = None, output_name: 'list[str]' = None, output_tooltips: 'list[str]' = None, output_matchtypes: 'list[str]' = None, name: 'str' = None, display_name: 'str' = None, description: 'str' = None, python_module: 'Any' = None, category: 'str' = None, output_node: 'bool' = None, deprecated: 'bool' = None, experimental: 'bool' = None, dev_only: 'bool' = None, api_node: 'bool' = None, price_badge: 'dict | None' = None, search_aliases: 'list[str]' = None, essentials_category: 'str' = None, has_intermediate_output: 'bool' = None)""" def __init__(self, input: dict = None, input_order: dict[str, list[str]] = None, is_input_list: bool = None, output: list[str] = None, output_is_list: list[bool] = None, output_name: list[str] = None, output_tooltips: list[str] = None, output_matchtypes: list[str] = None, name: str = None, display_name: str = None, description: str = None, python_module: Any = None, category: str = None, output_node: bool = None, deprecated: bool = None, experimental: bool = None, dev_only: bool = None, api_node: bool = None, price_badge: dict | None = None, search_aliases: list[str] = None, essentials_category: str = None, has_intermediate_output: bool = None) -> None: ... api_node: Any = ... category: Any = ... deprecated: Any = ... description: Any = ... dev_only: Any = ... display_name: Any = ... essentials_category: Any = ... experimental: Any = ... has_intermediate_output: Any = ... input: Any = ... input_order: Any = ... is_input_list: Any = ... name: Any = ... output: Any = ... output_is_list: Any = ... output_matchtypes: Any = ... output_name: Any = ... output_node: Any = ... output_tooltips: Any = ... price_badge: Any = ... python_module: Any = ... search_aliases: Any = ... class NodeOutput(_NodeOutputInternal): """Standardized output of a node; can pass in any number of args and/or a UIOutput into 'ui' kwarg.""" def __init__(self, *args: Any, ui: _UIOutput | dict = None, expand: dict = None, block_execution: str = None): ... @classmethod def from_dict(cls, data: dict[str, Any]) -> NodeOutput: ... @property def result(self) -> Any: ... class NodeReplace: """Defines a possible node replacement, mapping inputs and outputs of the old node to the new node.""" def __init__(self, new_node_id: str, old_node_id: str, old_widget_ids: list[str] | None = None, input_mapping: list[InputMap] | None = None, output_mapping: list[OutputMap] | None = None): ... # Create serializable representation of the node replacement. def as_dict(self): ... class Noise(ComfyTypeIO): io_type: Any = ... class NumberDisplay(str, Enum): gradient_slider: Any = ... number: Any = ... slider: Any = ... class Output(_IO_V3): def __init__(self, id: str = None, display_name: str = None, tooltip: str = None, is_output_list=False): ... def as_dict(self): ... def get_io_type(self): ... class Photomaker(ComfyTypeIO): io_type: Any = ... class Point(ComfyTypeIO): io_type: Any = ... class PriceBadge: """PriceBadge(expr: 'str', depends_on: 'PriceBadgeDepends' = , engine: 'str' = 'jsonata')""" def __init__(self, expr: str, depends_on: PriceBadgeDepends = ..., engine: str = 'jsonata') -> None: ... def as_dict(self, schema_inputs: list['Input']) -> dict[str, Any]: ... engine: Any = ... def validate(self) -> None: ... class PriceBadgeDepends: """PriceBadgeDepends(widgets: 'list[str]' = , inputs: 'list[str]' = , input_groups: 'list[str]' = )""" def __init__(self, widgets: list[str] = ..., inputs: list[str] = ..., input_groups: list[str] = ...) -> None: ... def as_dict(self, schema_inputs: list['Input']) -> dict[str, Any]: ... def validate(self) -> None: ... class Range(ComfyTypeIO): io_type: Any = ... class RemoteOptions: def __init__(self, route: str, refresh_button: bool, control_after_refresh: Literal['first', 'last'] = 'first', timeout: int = None, max_retries: int = None, refresh: int = None): ... def as_dict(self): ... class SEGS(ComfyTypeIO): io_type: Any = ... class SVG(ComfyTypeIO): io_type: Any = ... class Sampler(ComfyTypeIO): io_type: Any = ... class Schema: """Definition of V3 node properties.""" def __init__(self, node_id: str, display_name: str = None, category: str = 'sd', inputs: list[Input] = ..., outputs: list[Output] = ..., hidden: list[Hidden] = ..., description: str = '', search_aliases: list[str] = ..., is_input_list: bool = False, is_output_node: bool = False, is_deprecated: bool = False, is_experimental: bool = False, is_dev_only: bool = False, is_api_node: bool = False, price_badge: PriceBadge | None = None, not_idempotent: bool = False, enable_expand: bool = False, accept_all_inputs: bool = False, essentials_category: str | None = None, has_intermediate_output: bool = False) -> None: ... accept_all_inputs: Any = ... category: Any = ... description: Any = ... display_name: Any = ... enable_expand: Any = ... essentials_category: Any = ... # Add hidden based on selected schema options, and give outputs without ids default ids. def finalize(self): ... def get_v1_info(self, cls) -> NodeInfoV1: ... has_intermediate_output: Any = ... is_api_node: Any = ... is_deprecated: Any = ... is_dev_only: Any = ... is_experimental: Any = ... is_input_list: Any = ... is_output_node: Any = ... not_idempotent: Any = ... price_badge: Any = ... # Validate the schema: def validate(self): ... class Sigmas(ComfyTypeIO): io_type: Any = ... class Splat(ComfyTypeIO): io_type: Any = ... class String(ComfyTypeIO): io_type: Any = ... class StyleModel(ComfyTypeIO): io_type: Any = ... class TimestepsRange(ComfyTypeIO): """Range defined by start and endpoint, between 0.0 and 1.0.""" io_type: Any = ... class Tracks(ComfyTypeIO): io_type: Any = ... class UploadType(str, Enum): audio: Any = ... image: Any = ... model: Any = ... video: Any = ... class UpscaleModel(ComfyTypeIO): io_type: Any = ... class V3Data(dict): ... class Vae(ComfyTypeIO): io_type: Any = ... class Video(ComfyTypeIO): io_type: Any = ... class Voxel(ComfyTypeIO): io_type: Any = ... class WanCameraEmbedding(ComfyTypeIO): io_type: Any = ... class Webcam(ComfyTypeIO): io_type: Any = ... class WidgetInput(Input): """Base class for a V3 Input with widget.""" def __init__(self, id: str, display_name: str = None, optional=False, tooltip: str = None, lazy: bool = None, default: Any = None, socketless: bool = None, widget_type: str = None, force_input: bool = None, extra_dict=None, raw_link: bool = None, advanced: bool = None): ... def as_dict(self): ... def get_io_type(self): ... def add_to_dict_v1(i: Input, d: dict): ... # Decorator to mark nested classes as ComfyType; io_type will be bound to the class. def comfytype(io_type: str, **kwargs): ... # ─── comfy_api.latest.sdk ───────────────────────────────────────────────────────── class Ref: """Base opaque resource handle. ``kind`` is the io_type (IMAGE, LATENT...).""" def __init__(self, kind: str, id: str) -> None: ... # Return a bounded, inert description of this opaque value. async def describe(self, max_value_chars: int = 32768) -> dict[str, Any]: ... # Optional early free. **A node never has to call this.** async def release(self) -> None: ... class ClosureRef(_TypedRef): """Handle to a retained node closure (D21).""" KIND: Any = ... # Expose a ``latent_operation`` closure as LATENT_OPERATION. async def as_latent_operation(self) -> 'LatentOperationRef': ... # Expose a ``custom_sampler`` closure as a host-owned SAMPLER. async def as_sampler(self) -> 'SamplerRef': ... # Attach this closure to its declared canonical model phase. async def attach_model(self, model: 'ModelRef') -> 'ModelRef': ... # Wrap a sampler's model calls with a ``model_sigma`` closure. async def wrap_sampler(self, sampler: 'SamplerRef', *, start_percent: Optional[float] = None, end_percent: Optional[float] = None) -> 'SamplerRef': ... class ClosuresDomain: def __init__(self, *args, **kwargs): ... async def attach_model(self, closure: ClosureRef, model: ModelRef) -> ModelRef: ... async def attach_sampler(self, closure: ClosureRef, sampler: SamplerRef, *, start_percent: Optional[float] = None, end_percent: Optional[float] = None) -> SamplerRef: ... async def create_latent_operation(self, closure: ClosureRef) -> LatentOperationRef: ... async def create_sampler(self, closure: ClosureRef) -> SamplerRef: ... async def retain(self, kind: str, fn: Callable, *, captures: Optional[dict] = None) -> ClosureRef: ... class AnimaDomain: """Vendor-specific Anima model adapters.""" def __init__(self, *args, **kwargs): ... async def apply_lllite(self, model: ModelRef, weights: AssetRef, image: ImageRef, *, strength: float = 1.0, start_percent: float = 0.0, end_percent: float = 1.0, preserve_wrapper: bool = True) -> ModelRef: ... class TensorRef(_TypedRef): KIND: Any = ... async def raw(self) -> Any: ... class UpscaleModelRef(_TypedRef): KIND: Any = ... async def upscale(self, images: ImageRef, per_batch: int = 16, downscale_ratio: float = 1.0, downscale_method: str = 'lanczos', precision: str = 'float32', tile_size: Optional[int] = None, channels_last: bool = False) -> ImageRef: ... class ValueRef(_TypedRef): """A handle whose value is buffer-safe structured data.""" KIND: Any = ... @classmethod def from_value(cls, v: Any) -> 'Ref': ... # Read a dict/list of tensors and JSON scalars. async def value(self) -> Any: ... class ImageRef(TensorRef): KIND: Any = ... # Return the number of images without exposing their pixel buffers. async def batch_size(self) -> int: ... async def invert(self) -> 'ImageRef': ... async def op(self, name: str, **params: Any) -> 'ImageRef': ... # Return the first three channels of an image as an opaque ref. async def rgb(self) -> 'ImageRef': ... async def scale(self, factor: float) -> 'ImageRef': ... # Select an ordered, bounded set of images from a BHWC batch. async def select_batch(self, indices: list[int]) -> 'ImageRef': ... # Return image height and width without exposing its pixel buffer. async def spatial_shape(self) -> tuple[int, int]: ... # Clone this image onto a named ComfyUI-managed device. async def to_device(self, device: str = 'auto') -> 'ImageRef': ... class ImageClassifierRef(_TypedRef): KIND: Any = ... # Classify a host-side image batch and return bounded label scores. async def classify(self, images: ImageRef, use_accelerator: bool = True, top_k: int = 5) -> list[list[dict[str, Any]]]: ... # Run a multi-label classifier and retain its score matrix host-side. async def predict_scores(self, images: ImageRef) -> 'ClassifierScoresRef': ... class ClassifierScoresRef(_TypedRef): """Opaque bounded batch-by-class scores from an image classifier.""" KIND: Any = ... # Page score/index pairs above a threshold in one class range. async def select_above(self, batch_index: int, start: int, end: int, threshold: float, offset: int = 0, limit: int = 512) -> dict[str, Any]: ... async def shape(self) -> tuple[int, int]: ... class InpaintModelRef(_TypedRef): """Opaque prompt-free image inpainting model.""" KIND: Any = ... # Fill the masked image region while keeping model weights host-side. async def inpaint(self, image: ImageRef, mask: MaskRef) -> ImageRef: ... class ImagePreprocessorRef(_TypedRef): """Opaque host-created image preprocessor with one bounded operation.""" KIND: Any = ... async def apply(self, image: ImageRef, mask: Optional[MaskRef] = None) -> ImageRef: ... class IpAdapterEmbedsRef(_TypedRef): """Opaque image embeddings produced by a host IP-Adapter encoder.""" KIND: Any = ... async def combine(self, others: list['IpAdapterEmbedsRef'], method: str = 'concat') -> 'IpAdapterEmbedsRef': ... class IpAdapterRef(_TypedRef): """Opaque, host-created IP-Adapter pipeline.""" KIND: Any = ... # Apply this pipeline to a model using bounded image inputs. async def apply(self, model: ModelRef, image: ImageRef, negative_image: Optional[ImageRef] = None, attn_mask: Optional[MaskRef] = None, style_image: Optional[ImageRef] = None, composition_image: Optional[ImageRef] = None, weight: float = 0.7, weight_type: str = 'channel penalty', start_percent: float = 0.0, end_percent: float = 1.0, combine_embeds: str = 'concat', weight_faceidv2: float = 1.0, embeds_scaling: str = 'V only', unfold_batch: bool = False, layer_weights: Optional[str] = None, weight_style: float = 1.0, weight_composition: float = 1.0, expand_style: bool = False) -> ModelRef: ... # Apply already encoded image embeddings to a model. async def apply_embeds(self, model: ModelRef, positive: IpAdapterEmbedsRef, negative: Optional[IpAdapterEmbedsRef] = None, attn_mask: Optional[MaskRef] = None, weight: float = 1.0, weight_type: str = 'linear', start_percent: float = 0.0, end_percent: float = 1.0, embeds_scaling: str = 'V only') -> ModelRef: ... # Apply the canonical tiled IP-Adapter operation. async def apply_tiled(self, model: ModelRef, image: ImageRef, negative_image: Optional[ImageRef] = None, attn_mask: Optional[MaskRef] = None, weight: float = 0.7, weight_type: str = 'linear', start_percent: float = 0.0, end_percent: float = 1.0, combine_embeds: str = 'concat', embeds_scaling: str = 'V only', sharpening: float = 0.0, unfold_batch: bool = False) -> tuple[ModelRef, ImageRef, MaskRef]: ... # Encode one image into positive and negative IP-Adapter embeddings. async def encode(self, image: ImageRef, weight: float = 1.0, mask: Optional[MaskRef] = None) -> tuple[IpAdapterEmbedsRef, IpAdapterEmbedsRef]: ... class InteractionDomain: def __init__(self, *args, **kwargs): ... async def request(self, kind: str, payload: Any, *, reuse_last: bool = False, remember: bool = False, timeout: float = 540.0) -> Any: ... class InterpolationStatesRef(_TypedRef): """A frame-interpolation skip policy projected from another pack.""" KIND: Any = ... # Return which source-frame pairs must not be interpolated. async def skip_mask(self, pair_count: int) -> list[bool]: ... class MaskRef(TensorRef): KIND: Any = ... # Dilate or erode a mask through core's canonical morphology node. async def grow(self, amount: int, tapered_corners: bool = False) -> 'MaskRef': ... class MattingModelRef(_TypedRef): """Opaque fixed-architecture image matting model.""" KIND: Any = ... # Refine a coarse trimap into an alpha mask. async def refine(self, image: ImageRef, trimap: MaskRef, max_megapixels: float = 2.0) -> MaskRef: ... class LatentOperationRef(_TypedRef): """Handle to a host-owned LATENT_OPERATION callable.""" KIND: Any = ... class LatentRef(ValueRef): KIND: Any = ... # Composite a source latent using core's bounded latent operation. async def composite(self, source: 'LatentRef', *, x: int = 0, y: int = 0, resize_source: bool = False, mask: Optional['MaskRef'] = None) -> 'LatentRef': ... @classmethod # Create a bounded zero latent without granting raw tensor access. def empty(cls, width: int, height: int, batch_size: int = 1, channels: int = 4, spatial_downscale_ratio: Optional[int] = None) -> 'LatentRef': ... async def minimax_h3_token_count(self, conditioning: 'CondRef') -> dict[str, Any]: ... # Return the latent's optional noise mask as an opaque mask ref. async def noise_mask(self) -> Optional['MaskRef']: ... # Generate latent-shaped noise with a host-owned CPU/GPU RNG. async def random_noise(self, seed: int, source: str = 'cpu', batch_size: Optional[int] = None) -> TensorRef: ... # Repeat a latent through core's canonical batch operation. async def repeat_batch(self, amount: int) -> 'LatentRef': ... # Resize latent spatial cells with ComfyUI's canonical interpolators. async def resize(self, width: int, height: int, method: str = 'bilinear') -> 'LatentRef': ... # Return the latent sample height and width without exposing buffers. async def spatial_shape(self) -> tuple[int, int]: ... class LlmDomain: """Provider-neutral bounded chat and function-tool contract.""" def __init__(self, *args, **kwargs): ... async def chat(self, provider: str, profile: str, model: str, messages: list[dict[str, Any]], *, tools: Optional[list[dict[str, Any]]] = None, temperature: float = 0.8, max_tokens: int = 512, thinking: bool = False, response_format: str | dict[str, Any] = '', timeout_seconds: float = 600.0, vendor_options: Optional[dict[str, Any]] = None) -> dict[str, Any]: ... class LlamaCppDomain: """Bounded llama.cpp vendor adapter over managed GGUF weights.""" def __init__(self, *args, **kwargs): ... async def generate(self, model: LlamaCppModelRef, system: str, prompt: str, image: Optional[ImageRef] = None, video: Optional[ImageRef] = None, max_tokens: int = 512, temperature: float = 0.7, top_p: float = 0.9, repetition_penalty: float = 1.0, seed: int = 1) -> str: ... async def load_chat_model(self, model_weight: str, mmproj_weight: Optional[str] = None, *, family: str = 'qwen3_vl', device: str = 'auto', context_length: int = 8192, batch_size: int = 512, gpu_layers: int = -1, image_max_tokens: int = 4096, top_k: int = 0, pool_size: int = 4194304, cache: bool = True) -> LlamaCppModelRef: ... class LlamaCppModelRef(_TypedRef): """Opaque vendor-owned llama.cpp chat/VLM session.""" KIND: Any = ... async def generate(self, system: str, prompt: str, image: Optional[ImageRef] = None, video: Optional[ImageRef] = None, max_tokens: int = 512, temperature: float = 0.7, top_p: float = 0.9, repetition_penalty: float = 1.0, seed: int = 1) -> str: ... class CondRef(ValueRef): KIND: Any = ... async def combine(self, other: 'CondRef') -> 'CondRef': ... async def concat(self, other: 'CondRef') -> 'CondRef': ... # Whether tile-relative conditioning must be cropped per image. async def has_spatial_metadata(self) -> bool: ... async def sequence_length(self) -> int: ... # Crop 2D spatial conditioning to a latent-space window. async def spatial_crop(self, *, x: int, y: int, width: int, height: int, source_width: int, source_height: int, target_width: Optional[int] = None, target_height: Optional[int] = None) -> 'CondRef': ... # Attach one opaque CLIP-vision result to every conditioning row. async def with_clip_vision_output(self, output: 'ClipVisionOutputRef') -> 'CondRef': ... # Attach model-formatted latent ``c_concat`` conditioning. async def with_concat_latent(self, model: 'ModelRef', latent: 'LatentRef', extra_latent: Optional['LatentRef'] = None) -> 'CondRef': ... # Attach core conditioning-mask metadata without exposing tensors. async def with_mask(self, mask: MaskRef, strength: float = 1.0, set_area_to_bounds: bool = False) -> 'CondRef': ... # Attach closed, scalar micro-conditioning metadata. async def with_metadata(self, *, width: Optional[int] = None, height: Optional[int] = None, crop_w: Optional[int] = None, crop_h: Optional[int] = None, target_width: Optional[int] = None, target_height: Optional[int] = None) -> 'CondRef': ... # Clone conditioning with a normalized sampling-percent range. async def with_timestep_range(self, start: float, end: float) -> 'CondRef': ... # Zero embeddings while preserving the conditioning structure. async def zero_out(self) -> 'CondRef': ... class GligenRef(_TypedRef): KIND: Any = ... async def apply_batched(self, conditioning: 'CondRef', clip: 'ClipRef', text: str, boxes: list[tuple[int, int, int | float, int | float]]) -> 'CondRef': ... class GuiderRef(_TypedRef): """Opaque handle to a host-owned sampling guider.""" KIND: Any = ... # Clone this guider and crop model-owned spatial inputs for tiles. async def spatial_crop_inputs(self, *, regions: list[tuple[int, int, int, int]], source_width: int, source_height: int, target_width: int, target_height: int) -> 'GuiderRef': ... class SamplerRef(_TypedRef): """Opaque handle to a host-owned sampler.""" KIND: Any = ... @classmethod # Select a core sampler with its small, validated option set. def named(cls, name: str, *, eta: Optional[float] = None, ge_gamma: Optional[float] = None) -> 'SamplerRef': ... @classmethod def self_refine_video(cls, stochastic_steps: list[dict[str, int]], certain_percentage: float, uncertainty_threshold: float, seed: int, verbose: bool = False, latent: Optional['LatentRef'] = None) -> 'SamplerRef': ... class SigmasRef(_TypedRef): """Opaque one-dimensional host-owned sampling schedule.""" KIND: Any = ... # Return the number of sampling intervals in this schedule. async def steps(self) -> int: ... # Return one finite scalar from a bounded sampling schedule. async def value_at(self, index: int) -> float: ... class SamModelRef(_TypedRef): KIND: Any = ... # Segment one host-side image from bounded boxes and point hints. async def segment(self, image: ImageRef, boxes: list[Optional[list[float]]], point_coords: Optional[list[list[list[float]]]] = None, point_labels: Optional[list[list[int]]] = None, multimask_output: bool = True) -> tuple[MaskRef, list[list[float]]]: ... # Propagate frame-zero boxes through a host-side SAM2 video batch. async def segment_video(self, frames: ImageRef, boxes: list[list[float]]) -> MaskRef: ... class SemanticSegmentationRef(_TypedRef): """Opaque fixed-architecture semantic segmentation model.""" KIND: Any = ... # Return the union of selected semantic class IDs as a mask. async def mask(self, image: ImageRef, classes: list[int]) -> MaskRef: ... class HuggingFaceWeight: """A public Hugging Face weight file required by a node.""" def __init__(self, repo_id: str, filename: str, folder: str, revision: str = 'main', sha256: Optional[str] = None, on_demand: bool = False) -> None: ... @property def catalogue_name(self) -> Any: ... on_demand: Any = ... revision: Any = ... sha256: Any = ... class ModelRef(_TypedRef): KIND: Any = ... async def apply_dit_block_lora(self, asset: 'AssetRef', strength_model: float, block_weights: list[dict[str, Any]]) -> tuple['ModelRef', str]: ... # Apply one resolved LoRA without exposing model weights or paths. async def apply_lora(self, asset: 'AssetRef', clip: Optional['ClipRef'], strength_model: float, strength_clip: float) -> tuple['ModelRef', Optional['ClipRef']]: ... async def apply_ltx2_lora(self, asset: 'AssetRef', strength_model: float, block_weights: list[dict[str, Any]], video: float, video_to_audio: float, audio: float, audio_to_video: float, other: float) -> tuple['ModelRef', str, str]: ... # Return the model's canonical base family, or ``unknown``. async def family(self) -> str: ... # Text-ground objects in an image with a compatible vision MODEL. async def ground_image(self, image: ImageRef, conditioning: CondRef, *, threshold: float = 0.5, refine_iterations: int = 2, individual_masks: bool = True, max_detections: int = 64) -> tuple[MaskRef, list[list[dict[str, float]]]]: ... # Whether the model uses ComfyUI's FLOW model family. async def is_flow(self) -> bool: ... # Whether the model's sampling schedule uses zero terminal SNR. async def is_zero_terminal_snr(self) -> bool: ... async def latent_scale_factor(self) -> float: ... async def lora_weight_differences(self, original: 'ModelRef', include_bias: bool = False) -> 'WeightDiffCursorRef': ... # Apply a named transform, returning a NEW model ref. async def patch(self, transform: str, **params: Any) -> 'ModelRef': ... # Return one bounded scheduler delta in latent-value units. async def sampling_sigma_delta(self, *, steps: int, sampler_name: str, scheduler: str, start_step: int, end_step: int, denoise: float = 1.0, sigma_schedule: Optional[dict] = None) -> float: ... async def scheduled_cfg_guider(self, positive: 'CondRef', negative: 'CondRef', cfg: float, start_percent: float = 0.0, end_percent: float = 1.0, *, bounds: Optional[dict] = None) -> 'GuiderRef': ... # Project one sampling percentage through the model's schedule. async def sigma_for_percent(self, percent: float, actual_endpoints: bool = False) -> float: ... # Clone the model with model-owned spatial inputs cropped to tiles. async def spatial_crop_inputs(self, *, regions: list[tuple[int, int, int, int]], source_width: int, source_height: int, target_width: int, target_height: int) -> 'ModelRef': ... # Return the transforms supported by the active host. async def transforms(self) -> list[dict]: ... # Return a model's scalar UNet context dimension when published. async def unet_context_dim(self) -> Optional[int]: ... class ModelsDomain: def __init__(self, *args, **kwargs): ... async def download_huggingface_weights(self, repo_id: str, filename: str, folder: str, revision: str = 'main', sha256: Optional[str] = None) -> str: ... async def generate_text(self, generator: str, input_text: str, max_new_tokens: int = 128, weight: Optional[str] = None) -> str: ... async def list_controlnet(self) -> list[str]: ... async def list_diffusion_models(self, include_connectors: bool = False) -> list[str]: ... async def list_vae(self) -> list[str]: ... async def load_advanced_controlnet(self, name: str, model: Optional[ModelRef] = None, timestep_keyframe: Optional[TimestepKeyframeRef] = None) -> ControlNetRef: ... async def load_background_removal_model(self, model: str) -> BackgroundRemovalModelRef: ... async def load_brushnet(self, model: str, dtype: str = 'float16') -> BrushNetRef: ... async def load_checkpoint(self, name: str, weight_dtype: str = 'default', compute_dtype: str = 'default', cublas_linear: bool = False) -> tuple[ModelRef, Optional[ClipRef], Optional[VaeRef]]: ... async def load_clip_vision(self, model: str) -> ClipVisionRef: ... async def load_clipseg(self, model: str) -> ClipSegRef: ... async def load_controlnet(self, name: str, model: Optional[ModelRef] = None) -> ControlNetRef: ... async def load_controlnet_plusplus(self, name: str, control_type: str = 'none') -> ControlNetRef: ... async def load_diffusion_model(self, name: str, extra_name: Optional[str] = None, weight_dtype: str = 'default', compute_dtype: str = 'default', cublas_linear: bool = False) -> ModelRef: ... async def load_gguf_model(self, name: str, extra_name: Optional[str] = None, dequant_dtype: str = 'default', patch_dtype: str = 'default', patch_on_device: bool = False) -> ModelRef: ... async def load_gguf_text_encoders(self, names: Sequence[str], clip_type: str) -> ClipRef: ... async def load_image_classifier(self, model: str, architecture: str, labels: list[str]) -> ImageClassifierRef: ... async def load_inpaint_model(self, model: str, architecture: str = 'big-lama') -> InpaintModelRef: ... async def load_ipadapter(self, model: str, clip_vision: ClipVisionRef) -> IpAdapterRef: ... async def load_language_model(self, weights: list[str], family: str, device: str = 'default', cache: bool = True) -> ClipRef: ... async def load_object_detector(self, model: str) -> ObjectDetectorRef: ... async def load_onnx_detector(self, model: str) -> OnnxDetectorRef: ... async def load_onnx_image_classifier(self, model: str, input_layout: str = 'NHWC', channel_order: str = 'BGR', resize_mode: str = 'fit_pad', input_scale: float = 255.0, pad_color: tuple[float, float, float] = (1.0, 1.0, 1.0), mean: tuple[float, float, float] = (0.0, 0.0, 0.0), std: tuple[float, float, float] = (1.0, 1.0, 1.0), activation: str = 'identity', resize_filter: str = 'lanczos') -> ImageClassifierRef: ... async def load_powerpaint(self, model: str, base_clip: str, powerpaint_clip: str, dtype: str = 'float16') -> PowerPaintRef: ... async def load_sam(self, model: str, architecture: str = 'vit_b', device_mode: str = 'AUTO') -> SamModelRef: ... async def load_segformer(self, model: str, variant: str, num_labels: int) -> SemanticSegmentationRef: ... async def load_text_encoder(self, model: str, model_type: str, device: str = 'default') -> ClipRef: ... async def load_transparent_vae_decoder(self, model: str, family: str) -> TransparentVaeDecoderRef: ... async def load_upscale_model(self, name: str) -> UpscaleModelRef: ... async def load_vae(self, name: str, device: str = 'default', weight_dtype: str = 'default') -> VaeRef: ... async def load_vitmatte(self, model: str, variant: str) -> MattingModelRef: ... async def load_vqa(self, model: str, architecture: str, precision: str = 'fp16', device: str = 'cuda') -> VqaModelRef: ... async def memory_cleanup(self, empty_cache: bool = True, collect_cycles: bool = True, unload_all_models: bool = False) -> tuple[int, int]: ... class OnnxDetectorRef(_TypedRef): KIND: Any = ... # Run a catalogued ONNX object detector on one host-side image. async def detect(self, image: ImageRef) -> list[dict[str, Any]]: ... class ObjectDetectorRef(_TypedRef): KIND: Any = ... async def detect(self, image: ImageRef, threshold: float = 0.5, class_name: str = 'all', max_detections: int = 100) -> list[list[dict[str, Any]]]: ... class PowerPaintRef(_TypedRef): """Opaque PowerPaint model and token-extended CLIP pipeline.""" KIND: Any = ... async def apply(self, model: ModelRef, vae: VaeRef, image: ImageRef, mask: MaskRef, positive: CondRef, negative: CondRef, fitting: float = 1.0, function: str = 'text guided', scale: float = 1.0, start_step: int = 0, end_step: int = 10000, save_memory: str = 'none') -> tuple[ModelRef, CondRef, CondRef, LatentRef]: ... class ClipRef(_TypedRef): KIND: Any = ... # Describe bounded token IDs without exposing tokenizer objects. async def describe_tokens(self, tokens: dict) -> dict: ... # The two steps above in one call, for the common case. async def encode(self, text: str) -> 'CondRef': ... # Mirrors ``clip.encode_from_tokens_scheduled``, ``add_dict`` included. async def encode_from_tokens_scheduled(self, tokens: dict, add_dict: dict | None = None) -> 'CondRef': ... # Encode token-weight pairs with one named CLIP component. async def encode_token_weights_component(self, component: str, tokens: list) -> tuple[TensorRef, Optional[TensorRef]]: ... # Generate bounded text with a canonical Comfy text encoder. async def generate_text(self, prompt: str, image: Optional[ImageRef] = None, video: Optional[ImageRef] = None, max_length: int = 256, do_sample: bool = False, temperature: float = 1.0, top_k: Optional[int] = 50, top_p: float = 0.95, min_p: float = 0.0, repetition_penalty: float = 1.0, seed: Optional[int] = None, presence_penalty: float = 0.0, thinking: bool = False, use_default_template: bool = True, num_beams: int = 1) -> str: ... async def lora_weight_differences(self, original: 'ClipRef', include_bias: bool = False) -> 'WeightDiffCursorRef': ... # Scale selected CLIP/T5 attention projection weights. async def scale_attention_weights(self, *, clip_l: Optional[list[float]] = None, clip_g: Optional[list[float]] = None, t5xxl: Optional[list[float]] = None, query: bool = True, key: bool = True, value: bool = True, output: bool = True) -> 'ClipRef': ... # Clone this CLIP and stop encoding at a bounded hidden layer. async def set_last_layer(self, stop_at_clip_layer: int) -> 'ClipRef': ... # Mirrors ``clip.tokenize``, including its per-model kwargs. async def tokenize(self, text: str, **kwargs: Any) -> dict: ... # Clone this encoder with one host-registered attention function. async def with_attention_impl(self, mode: str) -> 'ClipRef': ... class ClipSegRef(_TypedRef): KIND: Any = ... # Return native-resolution sigmoid CLIPSeg predictions. async def predict_mask(self, images: ImageRef, text: str, use_accelerator: bool = True) -> MaskRef: ... async def segment(self, images: ImageRef, text: str, threshold: float = 0.5, binary_mask: bool = True, combine_mask: bool = False, use_accelerator: bool = True, blur_sigma: float = 0.0, previous_mask: Optional[MaskRef] = None, invert: bool = False, image_background_level: float = 0.5) -> tuple[MaskRef, ImageRef]: ... class ClipVisionRef(_TypedRef): KIND: Any = ... async def encode_image(self, image: ImageRef, crop: bool = True) -> ClipVisionOutputRef: ... class ClipVisionOutputRef(_TypedRef): KIND: Any = ... # Concatenate opaque penultimate vision tokens along their token axis. async def concat(self, other: 'ClipVisionOutputRef') -> 'ClipVisionOutputRef': ... async def image_embeds(self) -> TensorRef: ... class ControlNetRef(_TypedRef): KIND: Any = ... # Apply this ControlNet while all referenced data stays host-owned. async def apply(self, positive: CondRef, negative: CondRef, image: ImageRef, strength: float = 1.0, start_percent: float = 0.0, end_percent: float = 1.0, vae: Optional[VaeRef] = None) -> tuple[CondRef, CondRef]: ... # Apply Advanced-ControlNet scheduling, weights, and effect masks. async def apply_advanced(self, positive: CondRef, negative: CondRef, image: ImageRef, strength: float = 1.0, start_percent: float = 0.0, end_percent: float = 1.0, vae: Optional[VaeRef] = None, mask: Optional[MaskRef] = None, timestep_keyframe: Optional[TimestepKeyframeRef] = None, weights: Optional[ControlNetWeightsRef] = None) -> tuple[CondRef, CondRef]: ... async def compile(self, *, backend: str = 'inductor', mode: str = 'default', fullgraph: bool = False) -> 'ControlNetRef': ... async def with_union_type(self, type_number: Optional[int]) -> 'ControlNetRef': ... class ControlNetWeightsRef(_TypedRef): """Opaque Advanced-ControlNet weight policy.""" KIND: Any = ... @classmethod def from_list(cls, weights: list[float], uncond_multiplier: float = 1.0, extras: Any = None) -> tuple['ControlNetWeightsRef', 'TimestepKeyframeRef']: ... @classmethod def scaled_soft(cls, base_multiplier: float = 0.825, uncond_multiplier: float = 1.0) -> tuple['ControlNetWeightsRef', 'TimestepKeyframeRef']: ... class StyleModelRef(_TypedRef): KIND: Any = ... async def apply(self, clip_vision_output: ClipVisionOutputRef, conditioning: CondRef, strength: float = 1.0) -> CondRef: ... class SystemDomain: def __init__(self, *args, **kwargs): ... async def monitor(self) -> dict[str, Any]: ... async def stats(self) -> dict[str, Any]: ... class VaeRef(_TypedRef): KIND: Any = ... async def compile(self, *, backend: str = 'inductor', mode: str = 'default', fullgraph: bool = False, encoder: bool = True, decoder: bool = True) -> 'VaeRef': ... async def decode(self, latent: 'LatentRef') -> 'ImageRef': ... # Decode an audio latent and attach the VAE's output sample rate. async def decode_audio(self, latent: 'LatentRef') -> 'AudioRef': ... # Decode to an opaque BHWC tensor without assuming RGB channels. async def decode_tensor(self, latent: 'LatentRef') -> 'TensorRef': ... # Tiled :meth:`decode_tensor` with the canonical VAE tile options. async def decode_tensor_tiled(self, latent: 'LatentRef', tile_size: int = 512, overlap: int = 64, temporal_size: int = 64, temporal_overlap: int = 8) -> 'TensorRef': ... # Decode through ComfyUI's bounded, pixel-sized tiled operation. async def decode_tiled(self, latent: 'LatentRef', tile_size: int = 512, overlap: int = 64, temporal_size: int = 64, temporal_overlap: int = 8) -> 'ImageRef': ... # Decode image/video latents and flatten video batches to frames. async def decode_video(self, latent: 'LatentRef', *, tiled: bool = False, tile_size: int = 512, overlap: int = 64, temporal_size: int = 4096, temporal_overlap: int = 16) -> 'ImageRef': ... async def downscale_index_formula(self) -> Optional[tuple[int, int, int]]: ... # Mirrors ``vae.encode`` exactly. The caller owns any channel slicing. async def encode(self, image: 'ImageRef') -> 'LatentRef': ... # Run core VAEEncodeForInpaint with a bounded mask grow amount. async def encode_for_inpaint(self, image: 'ImageRef', mask: 'MaskRef', grow_mask_by: int = 6) -> 'LatentRef': ... # Run core InpaintModelConditioning without exposing model tensors. async def encode_inpaint_conditioning(self, image: 'ImageRef', mask: 'MaskRef', positive: 'CondRef', negative: 'CondRef', noise_mask: bool = True) -> tuple['CondRef', 'CondRef', 'LatentRef']: ... async def encode_tiled(self, image: 'ImageRef', tile_x: Optional[int] = None, tile_y: Optional[int] = None, overlap: Optional[int] = None, tile_t: Optional[int] = None, overlap_t: Optional[int] = None) -> 'LatentRef': ... # Encode frames after applying this VAE's temporal frame constraint. async def encode_video(self, image: 'ImageRef') -> tuple['LatentRef', int]: ... # The CPU dtype a node should use before VAE-specific preprocessing. async def input_dtype(self) -> str: ... # Return the VAE's bounded latent channel/compression metadata. async def latent_layout(self) -> dict[str, Optional[int]]: ... async def merge(self, other: 'VaeRef', ratio: float = 0.5) -> 'VaeRef': ... async def patch_triton(self, *, fuse_norm_silu: bool = True, channels_last: bool = True, int8_conv: bool = False, autotune: bool = False) -> 'VaeRef': ... class AudioRef(ValueRef): KIND: Any = ... class BackgroundRemovalModelRef(_TypedRef): """Opaque ComfyUI background-removal model.""" KIND: Any = ... # Generate a foreground alpha mask through core's canonical model. async def mask(self, image: ImageRef) -> MaskRef: ... class BrushNetRef(_TypedRef): """Opaque BrushNet weights loaded by the canonical host extension.""" KIND: Any = ... async def apply(self, model: ModelRef, vae: VaeRef, image: ImageRef, mask: MaskRef, positive: CondRef, negative: CondRef, scale: float = 1.0, start_step: int = 0, end_step: int = 10000) -> tuple[ModelRef, CondRef, CondRef, LatentRef]: ... class VideoRef(_TypedRef): KIND: Any = ... # Return encoded bytes and trim metadata, never the source path. async def encoded_source(self) -> ValueRef: ... class VqaModelRef(_TypedRef): """Opaque visual question-answering model.""" KIND: Any = ... async def answer(self, image: ImageRef, question: str, max_new_tokens: int = 32) -> str: ... class TimestepKeyframeRef(_TypedRef): """Opaque Advanced-ControlNet timestep-keyframe schedule.""" KIND: Any = ... class TransparentVaeDecoderRef(_TypedRef): """Opaque canonical decoder for Layer Diffusion transparency weights.""" KIND: Any = ... # Return a consistent RGBA batch and decoded alpha masks. async def decode(self, latent: LatentRef, image: ImageRef, frames: int = 1, sub_batch_size: int = 16) -> tuple[ImageRef, MaskRef]: ... class WeightDiffCursorRef(_TypedRef): """Execution-scoped iterator over host-owned model weight differences.""" KIND: Any = ... async def next(self) -> Optional[dict[str, Any]]: ... class AssetRef(_TypedRef): """A file/model resolved by name+hash, tenant-scoped. Never a raw path.""" KIND: Any = ... class AssetsDomain: def __init__(self, *args, **kwargs): ... async def delete_input(self, name: str) -> bool: ... async def digest(self, ref: AssetRef, algorithm: str = 'sha256') -> str: ... async def exists(self, folder: str, name: str) -> bool: ... async def latest(self, folder: str, prefix: str = '', suffix: str = '') -> Optional[str]: ... async def list(self, folder: str, prefix: str = '', recursive: bool = True) -> list[str]: ... async def load_image(self, ref: AssetRef) -> ImageRef: ... async def load_latent(self, ref: AssetRef) -> LatentRef: ... async def load_state_dict(self, ref: AssetRef, return_metadata: bool = False) -> Any: ... async def path(self, ref: AssetRef) -> str: ... async def read_bytes(self, ref: AssetRef) -> bytes: ... async def read_range(self, ref: AssetRef, offset: int = 0, length: int = 8388608) -> bytes: ... async def resolve(self, folder: str, name: str) -> AssetRef: ... async def size(self, ref: AssetRef) -> int: ... class OutputDomain: def __init__(self, *args, **kwargs): ... async def save_animation(self, images: ImageRef, fps: float = 8.0, filename_prefix: str = 'animation/ComfyUI', format: str = 'webp', loop_count: int = 0, lossless: bool = True, quality: int = 90, save_output: bool = True) -> dict: ... async def save_image_sequence(self, images: ImageRef, filename_prefix: str = 'sequence/ComfyUI', format: str = 'png', bit_depth: int = 8, save_output: bool = True) -> dict: ... async def save_images(self, images: ImageRef, filename_prefix: str = 'ComfyUI', subfolder: str = '', compress_level: int = 4, caption: Optional[str] = None, caption_extension: str = '.txt', save_metadata: bool = True, extra_metadata: Optional[dict[str, Any]] = None, image_format: str = 'png', quality: int = 95, filenames: Optional[list[str]] = None, lossless: bool = False, optimize: bool = False) -> dict: ... async def save_images_with_alpha(self, images: ImageRef, mask: MaskRef, filename_prefix: str = 'ComfyUI', subfolder: str = '', compress_level: int = 4) -> dict: ... async def save_latent(self, latent: LatentRef, filename_prefix: str = 'latents/LatentSender', preview_method: str = 'Latent2RGB-SDXL') -> dict: ... async def save_model(self, model: ModelRef, filename_prefix: str, model_key_prefix: str = 'model.diffusion_model.') -> str: ... async def save_state_dict(self, state_dict: ValueRef, filename_prefix: str, metadata: Optional[dict[str, str]] = None) -> str: ... async def save_text(self, text: str, filename_prefix: str = 'text', subfolder: str = '', extension: str = '.txt') -> str: ... async def save_video(self, images: ImageRef, audio: Optional[AudioRef] = None, fps: float = 25.0, filename_prefix: str = 'video/ComfyUI', format: str = 'auto', codec: str = 'auto', encoder_options: Optional[dict[str, Any]] = None, loop_count: int = 0, bit_depth: int = 8, save_output: bool = True, save_metadata: bool = True) -> dict: ... async def save_workflow_json(self, filename: str, mode: str = 'new_only') -> str: ... async def write_text(self, text: str, filename: str, folder: str = 'output', mode: str = 'overwrite', insert_newline: bool = False) -> str: ... class GraphDomain: def __init__(self, *args, **kwargs): ... async def block(self, reason: Optional[str] = None) -> Any: ... async def current_node_id(self) -> str: ... async def expand_loop(self, flow: Any, values: list[Any]) -> dict[str, Any]: ... async def expand_nodes(self, nodes: list[dict[str, Any]], outputs: list[dict[str, Any]]) -> dict[str, Any]: ... async def input_label(self, input_name: str, default: str = '') -> str: ... async def widget_values(self, node_id: int | str = 0, node_title: str = '', node_name: str = '', linked_input: str = 'any_input') -> dict[str, Any]: ... class Context: def __init__(self, *args, **kwargs): ... class CivitaiDomain: """Bounded read-only projection of the Civitai public model API.""" def __init__(self, *args, **kwargs): ... async def model_version(self, model_version_id: int) -> dict[str, Any]: ... async def model_version_by_hash(self, hash_value: str, refresh: bool = False) -> dict[str, Any]: ... async def search_models(self, username: str, query: Optional[str] = None, limit: int = 20, nsfw: bool = False) -> dict[str, Any]: ... class OllamaDomain: """Bounded Ollama vendor API; endpoint is loopback or an admin profile.""" def __init__(self, *args, **kwargs): ... async def chat(self, endpoint: str, model: str, messages: list[dict[str, Any]], images: Optional[ImageRef] = None, think: bool = False, options: Optional[dict[str, Any]] = None, keep_alive: int = 5, keep_alive_unit: str = 'minutes', format: str | dict[str, Any] = '', timeout_seconds: float = 600.0, tools: Optional[list[dict[str, Any]]] = None) -> dict[str, Any]: ... async def generate(self, endpoint: str, model: str, system: str, prompt: str, images: Optional[ImageRef] = None, context: Optional[list[int]] = None, think: bool = False, options: Optional[dict[str, Any]] = None, keep_alive: int = 5, keep_alive_unit: str = 'minutes', format: str | dict[str, Any] = '', timeout_seconds: float = 600.0) -> dict[str, Any]: ... async def list_models(self, endpoint: str) -> list[str]: ... class WanVideoDomain: """Bounded metadata for WanVideo vendor-owned opaque model handles.""" def __init__(self, *args, **kwargs): ... async def transformer_dim(self, model: Ref) -> int: ... class WebSearchDomain: """Fixed-profile web search with bounded normalized results.""" def __init__(self, *args, **kwargs): ... async def search(self, query: str, *, provider_profile: str = 'duckduckgo', limit: int = 5, vendor_options: Optional[dict[str, Any]] = None) -> list[dict[str, str]]: ... class IntegrationsDomain: """Vendor pass-throughs with vendor-shaped, less-stable contracts.""" def __init__(self, *args, **kwargs): ... class ExecutionDomain: def __init__(self, *args, **kwargs): ... async def interrupt(self) -> bool: ... def ctx() -> 'Context': ... def current_context() -> 'Context': ... class RefResolver: def __init__(self, *args, **kwargs): ... async def create(self, kind: str, obj: Any) -> Ref: ... async def release(self, ref: Ref) -> None: ... async def resolve(self, ref: Ref) -> Any: ... class ExecutionBackend: def __init__(self, *args, **kwargs): ... # Run the node. Default just awaits ``local_call`` (in-process). The async def dispatch(self, plan: ExecutionPlan, local_call: Callable[[], Awaitable[Any]], runtime: Optional[Runtime] = None) -> Any: ... class ExecutionPlan: """What the execution seam hands the backend to decide placement.""" def __init__(self, prompt_id: str, node_id: str, node_type: str, tier: str = 'default', permissions: tuple[str, ...] = (), required_weights: tuple[HuggingFaceWeight, ...] = (), node_module: str = '', inputs: Optional[dict] = None, input_mode: str = 'refs', prompt: Any = None, extra_pnginfo: Any = None, dynamic_prompt: Any = None, method: str = 'execute') -> None: ... dynamic_prompt: Any = ... extra_pnginfo: Any = ... input_mode: Any = ... inputs: Any = ... method: Any = ... node_module: Any = ... permissions: Any = ... prompt: Any = ... required_weights: Any = ... tier: Any = ... class OpNotSupported(NotImplementedError): """Raised by ``apply`` for an op this provider does not implement. Carries""" def __init__(self, op: str) -> None: ... providers: Providers = ...