This document provides detailed reference documentation for the Python node API. It covers all available methods, properties, and classes that can be used when developing Python nodes for Workflows.

Overview

Python nodes in Workflows are implemented as Python files that define specific functions to handle initialization, readiness checking, and processing logic. Each node file must contain at least three required functions: init, isReady, and process.

Node Methods

All node methods are available through the self object within the node functions.

init(self)

The init function is called when the node is initialized. This is where you set up the node's properties, connectors, and other configurations.

def init(self):
    self.color = QColor(92, 184, 92)
    self.addConnector("input", Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.MODE_OPTIONAL)    
    self.addConnector("output", Cconnector.CONNECTOR_TYPE_OUTPUT, Cconnector.MODE_OPTIONAL)    
    self.addProperty("url")    
    self.addProperty("threads", 1)
    self.bypassSupported = True
    return True

isReady(self)

The isReady function checks if the node is ready to process. This is called before process.

def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True    
    # Add readiness checks here
    return True

process(self)

The process function contains the main logic of the node. This is where the actual work happens.

def process(self):
    # Process logic here
    return True

Properties

Properties are defined using self.addProperty(name, defaultValue) and can be accessed directly as self.property_name.

addProperty(self, name, defaultValue=None)

Add a property to the node.

Parameters

name	Property name. If the name starts with an underscore (_), it will be created as internal and will not be stored in the .mwf workflow file.
defaultValue	Default value for the property

self.addProperty("url")

self.addProperty("threads", 1)

addEncryptedProperty(self, name, defaultValue=None)

Add an encrypted property to the node.

Encrypted properties can be included in the Property Editor via the node XSD like any other property. When shown in the GUI, they are displayed as a password/masked field (the clear-text value is never revealed). The value is stored encrypted when persisted (node defaults, projects, etc.), but can be accessed programmatically by the node at runtime.

Parameters

name	Property name. If the name starts with an underscore (_), it will be created as internal and will not be stored in the .mwf workflow file.
defaultValue	Default value for the property

self.addEncryptedProperty("password")

self.addEncryptedProperty("api_key", "secret_key")

setAcceptConnectors(self, accept, pattern)

Set which connectors are accepted.

Parameters

accept	Boolean indicating whether to accept connectors
pattern	Pattern to match connector names

self.setAcceptConnectors(True, "input_%")

addConnector(self, name, type, mode)

Add a connector to the node.

Parameters

name	Connector name
type	Connector type (CONNECTOR_TYPE_INPUT or CONNECTOR_TYPE_OUTPUT)
mode	Connector mode (MODE_OPTIONAL or MODE_MANDATORY)

self.addConnector("input", Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.MODE_OPTIONAL)

self.addConnector("output", Cconnector.CONNECTOR_TYPE_OUTPUT, Cconnector.MODE_OPTIONAL)

Connector Types

CONNECTOR_TYPE_INPUT

Input connector type.

CONNECTOR_TYPE_OUTPUT

Output connector type.

MODE_OPTIONAL

Optional connector mode.

MODE_MANDATORY

Mandatory connector mode.

Detailed Node Methods

getConnectorsByType(self, type, specialType)

Get connectors by type.

Parameters

type	Connector type to filter by
specialType	Special type to filter by

Returns: List of connectors matching the criteria

inputs = self.getConnectorsByType(Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.CONNECTOR_SPECIALTYPE_NORMAL)

getFirstConnectorByName(self, name)

Get first connector by name.

Parameters

name	Connector name to find

Returns: First connector matching the name, or None if not found

output = self.getFirstConnectorByName("output")

evaluate(self, property_name)

Evaluate a property value.

Parameters

property_name Name of the property to evaluate

Returns: Evaluated property value

url = self.evaluate(self.url)

isCancelled(self)

Check if node execution was cancelled.

Returns: True if cancelled, False otherwise

if self.isCancelled():

return False

progressUpdated(self, value)

Update progress.

Parameters

value Progress value (0-100)

self.progressUpdated(50)

setComplexity(self, value)

Set node complexity for progress tracking.

Parameters

value Complexity value

self.setComplexity(totalNumberOfUPs(self)*100)

info(self, category, message)

Log information.

Parameters

category	Category of the information
message	Information message

self.info("ffmpeg:process:up","Preparing {}".format(fp))

critical(self, category, message)

Log critical error.

Parameters

category	Category of the error
message	Error message

return self.critical("ffmpeg:process:error","Unable to process {}. exitCode={}".format(fp,exitCode))

addFailedUP(self, up, message)

Add failed universal path.

Parameters

up	Universal path that failed
message	Error message

self.addFailedUP(up,"Unable to process {}. exitCode={}".format(fp,exitCode))

addPermanentCritical(self, id, description, group)

Add a permanent critical error that will be reported when evaluating the isReady status. Permanent errors are stored and remain active until removed with clearAllPermanentErrors(), clearPermanentErrorById(), or clearPermanentErrorsByGroup(). Parameters are equivalent to those of the normal critical-error function.

Important: For errors in init, isReady, and process functions, use the regular critical and error methods instead. Permanent errors should only be used in functions that are not part of the core validation/execution process.

Parameters

id	Error ID
description	Error description
group	Optional parameter used to group errors, so they can be removed later as a group

Returns: Same return value as the normal critical-error function

self.addPermanentCritical("auth:timeout", "Authentication timeout")

addPermanentError(self, id, description, group)

Add a permanent error that will be reported when evaluating the isReady status. Permanent errors are stored and remain active until removed with clearAllPermanentErrors(), clearPermanentErrorById(), or clearPermanentErrorsByGroup(). Parameters are equivalent to those of the normal error function.

Important: For errors in init, isReady, and process functions, use the regular critical and error methods instead. Permanent errors should only be used in functions that are not part of the core validation/execution process.

Parameters

id	Error ID
description	Error description
group	Optional parameter used to group errors, so they can be removed later as a group

Returns: Same return value as the normal error function

self.addPermanentError("auth:invalid", "Invalid authentication credentials")

clearPermanentErrors(self)

Clear all permanent errors stored in the node.

self.clearPermanentErrors()

deletePermanentErrorById(self, id)

Delete a permanent error by its ID.

Parameters

id	Error ID to delete

self.deletePermanentErrorById("auth:timeout")

deletePermanentErrorsByGroup(self, group)

Delete all permanent errors belonging to a specific group.

Parameters

group Error group to delete

self.deletePermanentErrorsByGroup("auth")

External Application Integration

createExternalApp(self)

Create an external application node.

Returns: External application node

app = self.createExternalApp()

createExecutionQueue(self, threads)

Create an execution queue.

Parameters

threads Number of threads to use

Returns: Execution queue

queue = self.createExecutionQueue(threads)

setParams(self, binary_path, arguments)

Set parameters for external application.

Parameters

binary_path	Path to the binary
arguments	Arguments to pass to the binary

app.setParams(binfp,args)

addJob(self, app)

Add a job to the execution queue.

Parameters

app	External application to add

queue.addJob(app)

wait(self, timeout)

Wait for jobs to complete.

Parameters

timeout Timeout in milliseconds

Returns: True if jobs completed, False if timeout occurred

while queue.wait(1000):
    if self.isCancelled():
        queue.cancelAll()

Complete Node Examples

Simple Node Example

# This example creates simple Task with no inputs nor outputs
# On execution, it simple prints a "Hello World" message in console
 
def init(self):
    return True
    
def isReady(self):
    return True
    
def process(self):
    self.info("HelloWorld","Hello World")
    return True

Complex Node Example

from Mistika.Qt import QColor
from Mistika.classes import Cconnector,CbaseItem,CexternalAppNodeLink
from Mistika import sgoPaths
import os
import sys
from baseItemTools import totalNumberOfUPs
import Mistika
 
def ffmpegPath():
    ext=".exe" if sys.platform == "win32" else ""
    fldr = "/MacOS" if sys.platform == "darwin" else "/bin"
    fp=sgoPaths.apps()+ fldr + "/ffmpeg" + ext
    return fp
    
def init(self):
    self.setClassName("FFmpeg") 
    self.color=QColor(92, 184, 92)
    self.addConnector("input",Cconnector.CONNECTOR_TYPE_INPUT,Cconnector.MODE_OPTIONAL)    
    self.addConnector("output",Cconnector.CONNECTOR_TYPE_OUTPUT,Cconnector.MODE_OPTIONAL)    
    self.setAcceptConnectors(True,"input_%") 
    self.addProperty("url")    
    self.addProperty("threads",1)
    self.addProperty("dstExtension","mp4")
    self.addProperty("cmdLineParams","-i [input] -c:v [param1] -b:v [param2] -c:a [param3] [param4] [param5] [param6] [param7] [param8] [output]")
    self.addProperty("param1","h264_nvenc")
    self.addProperty("param2","5M")
    self.addProperty("param3","aac")
    self.addProperty("param4","")
    self.addProperty("param5","")
    self.addProperty("param6","")
    self.addProperty("param7","")
    self.addProperty("param8","")
    
    self.bypassSupported=True
    self.setDropToProperty("url")
    #1=Directories, 2=Files, 3=both
    self.setDropSupportedTypes(1) 
    return True
        
def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True    
    fp=ffmpegPath()
    if not os.path.exists(fp):
        return self.critical("ffmpeg:isReady","ffmpeg binary not found: ".format(fp))
    return True
 
 
def process(self):      
        
    res=True
    inputs=self.getConnectorsByType(Cconnector.CONNECTOR_TYPE_INPUT,Cconnector.CONNECTOR_SPECIALTYPE_NORMAL)
    out = self.getFirstConnectorByName("output")
    out.clearUniversalPaths()
    dstUrl=self.evaluate(self.url).strip()
    self.setComplexity(totalNumberOfUPs(self)*100)
    if self.bypassEnabled:
        return True
        
    self.progressUpdated(0)
    binfp=ffmpegPath()
    apps=[]
    threads=int(self.threads)    
# First, we create a queue of threads to execute the binary
    queue=self.createExecutionQueue(threads)
    cancelled=False
    totalFrames=0
    for c in inputs:
        if cancelled:
            break
        for up in c.getUniversalPaths():
            mfid=up.getMediaFileInfoData()
            frames=mfid.getToken("Frames")
            fp=up.getFilePath()
            self.info("ffmpeg:process:up","Preparing {}".format(fp))
            dst=self.composeDstFilePath(dstUrl,up,False,0)
            ext=self.evaluate(self.dstExtension.strip())
            if ext:
                dst.setExtension(ext)     
# For each UP to process, we create an instance of the app to execute it
            app=self.createExternalApp()
# Add the new app to the apps list. this list is used at the end to delete them
            apps.append((app,up,dst))
            if frames>0:
                totalFrames+=frames
                app.setMaxProgress(frames,totalFrames)       
# Register the Hooks to parse the command output
            app.addHookRX("errorRX",QRegularExpression("^(Error .*)$"))
            app.addHookRX("progressRX",QRegularExpression("^frame=\\s*(\\d+).*"))
            app.setOutputUP(dst)
# select if we want the app to print its output or not (printing it is not needed for the books to work, it is just for log/debug proposes            
            app.setPrintOutputs(app.EA_STD_BOTH)
# define the default tokens dictionary ([input] [input-firstFile] [output])            
            defTokens=app.createDefaultDict(up)
#parse the cmdLIne tokens            
            args=app.createArgumentsList(self.evaluate(self.cmdLineParams.strip()),up,defTokens)
            args=('-y',)+args # adds -y to cmdline to avoid ffmpeg blocking waiting an overwrite answer
# add " to the elements with spaces
            quoted=app.quoteList(args)
            if self.isCancelled():
                res=False
                cancelled=True
                queue.cancelAll()
                break
            self.info("ffmpeg:process:args","\"{}\" {}".format(binfp,' '.join(quoted)))
#execute the job           
            app.setParams(binfp,args)                   
            queue.addJob(app)
#wait for the apps to finish       
    while queue.wait(1000):
        if self.isCancelled():
            res=False
            queue.cancelAll()
#build the output connector and delete all the apps and queue
    for app,up,dst in apps:   
        exitCode=app.exitCode()
        fp=up.getFilePath()
        if exitCode: 
            res=self.critical("ffmpeg:process:error","Unable to process {}. exitCode={}".format(fp,exitCode)) and res
            self.addFailedUP(up,"Unable to process {}. exitCode={}".format(fp,exitCode))
        else:
            out.addUniversalPath(dst)
        app.deleteLater()
    queue.deleteLater()
    self.progressUpdated(self.complexity())
    return res
 
# appId ExternalApp emiting the signal
# output can be EA_STD_OUT or EA_STD_ERR
# name is the hook name
# rxMatch is the matching QRegularExpressionMatch 
# line is the matching line
def onExternalAppHook(self,appId,output,name,rxMatch,line):
    # you can manage your custom hooks here
    print("Signal",name,"received from std",output,"(Ignored)")

Development Workflow

Create a new Python file in /workspace/project/src/data/resources.workflows.nodes.library/shared/workflowsLibrary/
Implement the required functions (init, isReady, process)
Create corresponding XSD schema file in /workspace/project/src/data/resources.workflows.nodes.library/shared/config/schemas/
Test the node in the workflow editor
Debug and refine as needed

Troubleshooting

Common issues:

Missing required functions (init, isReady, process)
Incorrect connector setup
Property access errors
Progress tracking issues
External application path problems
Schema validation errors

Always check the workflow logs for error messages when debugging node issues.