Custom Script Node - Synthreo Pylon

Custom Script node for Pylon - write and execute custom Python code inline within a workflow to implement bespoke logic, transformations, or integrations not covered by built-in nodes.

Purpose

The Custom Script node allows you to execute custom code inside Synthreo Pylon workflows. Scripts are Python only, enabling complex data processing, calculations, custom integrations, and business logic that goes beyond what the built-in nodes provide.

This node is the escape hatch for scenarios where standard transformation, filter, or API nodes are not sufficient. It gives you full programmatic control over the data at any point in a workflow.

Inputs

Data from the previous workflow node is passed into the script via the input_values parameter of the execute function. It is always a list of dictionaries - one dictionary per row.

Outputs

Script output is passed to the next workflow node.
The return value must be a list of dictionaries - the rows going out. Each dictionary is one row, with its fields at the top level, so pick names that will not clash with the fields already on the row.

Output format example - a list, one dictionary per row:

[
  { "invoice_number": "INV-1", "processed": true, "score": 95, "status": "approved" },
  { "invoice_number": "INV-2", "processed": true, "score": 60, "status": "review" }
]

Parameters

Name	Type	Required	Default	Description
Script Content	Code editor	Yes	Empty	The Python code to be executed when this node runs. Must define an `execute` function as the entry point.
Script Encryption (`encryption`)	Toggle	No	Off	When enabled, the script code is encrypted and cannot be viewed or edited without the correct password. Use to protect proprietary logic.
Encryption Password (`password`)	Password field	Required if encryption is enabled	Empty	The password used to protect the encrypted script. Required when Script Encryption is enabled.

Python Execution Contract

Each Python script must define an execute function as the entry point. The workflow engine calls this function when the node executes.

Function signature - this exact signature is required. Renaming the parameters is valid Python but the node rejects it, and the script must not include an if __name__ == "__main__": block:

def execute(input_values, meta):
    # input_values: list of dicts - the rows coming in
    # meta:         dict of runtime helpers
    # return:       list of dicts - the rows going out
    return [
        {**row, "processed": True}
        for row in input_values
    ]

Parameter descriptions

input_values - The rows passed from the previous node, always a list of dictionaries. The exact keys are not fixed: they come from whatever the upstream node produced, so read defensively rather than assuming a shape.
meta - Contains workflow context and metadata provided by the workflow engine. Use this for accessing runtime context when needed.

Rules

The execute function is the required entry point. The workflow engine will not run the script if this function is missing.
The return value must be a Python list of dictionaries. Returning a bare dictionary fails validation.
Do not wrap the result in a top-level key. Each dictionary in the list is one output row; carry through the fields you need and add your own alongside them, choosing names that do not collide with fields the upstream node already produced.
The return value must be JSON-serializable. Avoid returning Python objects, classes, or types that cannot be converted to JSON (such as datetime objects without serialization, custom class instances, or numpy arrays without conversion).

Example Usage

Loyalty Points Calculator

A workflow processes customer purchase records and calculates loyalty points using custom business rules that combine purchase amount, customer tier multipliers, and seasonal bonus factors.

def execute(input_values, meta):
    multipliers = {"standard": 1.0, "silver": 1.5, "gold": 2.0}
    out = []

    for row in input_values:
        tier = row.get("tier", "standard")
        multiplier = multipliers.get(tier, 1.0)
        out.append({
            **row,
            "points_earned": int(row.get("purchase_amount", 0) * multiplier),
            "multiplier_applied": multiplier,
        })

    return out

Invoice Validation

A workflow validates incoming invoice records before payment processing, flagging records that are incomplete or contain suspicious values.

def execute(input_values, meta):
    out = []

    for row in input_values:
        issues = []
        if not row.get("vendor_name"):
            issues.append("Missing vendor name")
        if not row.get("invoice_number"):
            issues.append("Missing invoice number")
        if row.get("amount", 0) <= 0:
            issues.append("Invalid amount")

        out.append({
            **row,
            "valid": len(issues) == 0,
            "issues": issues,
        })

    return out

Data Enrichment with Computed Fields

A workflow receives sales records and adds computed fields such as margin percentage and performance category.

def execute(input_values, meta):
    out = []

    for row in input_values:
        try:
            revenue = float(row.get("revenue", 0))
            cost = float(row.get("cost", 0))
            margin = ((revenue - cost) / revenue * 100) if revenue > 0 else 0

            if margin >= 40:
                category = "high_margin"
            elif margin >= 20:
                category = "standard_margin"
            else:
                category = "low_margin"

            out.append({
                **row,
                "margin_percent": round(margin, 2),
                "performance_category": category,
            })
        except Exception as e:
            # Keep the row and mark it, so one bad record does not
            # discard the rest of the batch.
            out.append({
                **row,
                "margin_percent": None,
                "performance_category": "error",
                "error_message": str(e),
            })

    return out

Best Practices

Start simple and test often: Build your script incrementally. Test with small input datasets before deploying to production workflows. Use the node’s test functionality to verify output format.
Apply error handling with try/except: Unhandled exceptions in the script will cause the workflow to fail at this node. Wrap your logic in try/except blocks and return a structured error response rather than letting exceptions propagate.
Return a list, one dictionary per row: The node rejects a bare dictionary. Carry the incoming fields through and add yours alongside them, so a downstream node still sees what it expects.
Keep scripts focused: Each Custom Script node should do one logical thing. If you need multiple independent transformations, use multiple Custom Script nodes in sequence. This improves readability and makes debugging easier.
Avoid importing large libraries unless necessary: The script runtime environment has access to a set of standard libraries. Test that any imports you use are available in the environment before deploying.
Use encryption for proprietary logic: If the script contains business-sensitive algorithms or proprietary rules, enable Script Encryption and store the password securely. Note that losing the encryption password means the script cannot be recovered or edited.
Name your fields descriptively, and avoid collisions: Your keys sit at the top level of each row beside the incoming ones, and ordinary Python merge order decides a clash. In {**row, "status": "approved"} your value is written last, so it replaces any status the upstream node produced - silently, with no error, for every row. A downstream node still reading that field gets yours. Prefer specific names such as loyalty_points, validation_issues, or margin_percent, rather than result or status.

Troubleshooting

Workflow fails at the Custom Script node

Check the error message in the workflow execution log. Common causes include:

A syntax error in the Python code.
An unhandled exception thrown during execution.
A missing execute function definition.
An import that is not available in the runtime environment.

Test the script with a sample input using the node’s test functionality before running the full workflow.

Output fields from this node are missing in downstream nodes

Confirm the script returns a list of dictionaries. Returning a bare dictionary fails validation, and returning an empty list produces no rows downstream.

Then check the field name a downstream node is referencing. Your keys sit at the top level of each row, so a field named score is referenced as score - there is no wrapper segment in the path.

If a field is not missing but holds the wrong value, suspect a name collision. Writing {**row, "status": ...} puts your key last, so it replaces the upstream status rather than sitting beside it, and nothing reports the overwrite. Rename yours in the script, or put **row last if the upstream value is the one that should win.

Script works locally but fails in the workflow

The workflow runtime environment may not have all the same libraries available as your local Python environment. Test using only the standard library and commonly available packages. Avoid dependencies on packages that are unlikely to be pre-installed.

JSON serialization error

The return value contains a non-JSON-serializable Python type. Common culprits include datetime objects, set types, custom class instances, and numpy arrays. Convert these to JSON-compatible types (strings, lists, dicts, numbers) before returning.

Encrypted script cannot be edited

Once a script is encrypted, it requires the correct password to view or edit. If the password is lost, the script cannot be recovered. As a precaution, maintain a backup of the unencrypted script in a secure location outside the workflow editor.

Test Cases

Given: Input [{"purchase_amount": 100, "tier": "gold"}] with a loyalty calculator script - Expected: A list of one row, carrying purchase_amount and tier through, with points_earned added at the top level and holding the correct calculated value.
Given: Input with missing required fields passed to a validation script - Expected: A list of one row per input record, each with valid = false and an issues list naming the missing fields.
Given: A script returning a bare dictionary rather than a list - Expected: Validation fails at this node; the return value must be a list.
Given: Script with a syntax error - Expected: Workflow fails at this node with a syntax error message in the execution log.
Given: Script that raises an unhandled exception - Expected: Workflow fails at this node; error is reported in execution log.
Given: Script that returns a non-JSON-serializable value - Expected: Serialization error at runtime; update the script to convert the value before returning.

Set Transformation - For common data reshaping operations (join, group, filter, foreach) that do not require custom code.
String Operation - For simple string manipulation that does not require scripting.
ConvertFromJSON - Can produce Python dictionary strings that a downstream Custom Script node may process.
Annotation - Use to document the purpose and logic of a Custom Script node on the workflow canvas.