Custom Script Node - Synthreo Builder

---

Purpose

The Custom Script node allows you to execute custom code inside Synthreo Builder workflows. It supports Python, 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_json parameter of the execute function.

---

Outputs

• Script output is passed to the next workflow node.
• The return value must be a dictionary wrapped inside a unique top-level key (e.g., {"custom_result": {...}}). This prevents naming collisions with fields from other nodes in the same workflow.

Output format example:

{
  "custom_result": {
    "processed": true,
    "score": 95,
    "status": "approved"
  }
}

---

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.
Programming Language (language)	Dropdown	No	Python	The scripting language for the code. Defaults to Python.
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:

def execute(input_json, meta):
    # input_json: data received from the previous node (dict or list)
    # meta: metadata provided by the workflow engine (dict)
    result = {
        "custom_result": {
            "original": input_json,
            "processed": True
        }
    }
    return result

Parameter descriptions

• input_json - Contains the data passed from the previous node. This is typically a dictionary or list of dictionaries, depending on the upstream node’s output format.
• 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 dictionary.
• The return value must be wrapped in a unique top-level key (e.g., "custom_result") to avoid field name collisions with other nodes in the workflow. Do not return data at the top level without a wrapper key.
• 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_json, meta):
    purchase_amount = input_json.get("purchase_amount", 0)
    customer_tier = input_json.get("tier", "standard")

    multipliers = {"standard": 1.0, "silver": 1.5, "gold": 2.0}
    multiplier = multipliers.get(customer_tier, 1.0)

    points = int(purchase_amount * multiplier)

    return {
        "loyalty_result": {
            "points_earned": points,
            "tier": customer_tier,
            "multiplier_applied": multiplier
        }
    }

Invoice Validation

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

def execute(input_json, meta):
    issues = []

    if not input_json.get("vendor_name"):
        issues.append("Missing vendor name")
    if not input_json.get("invoice_number"):
        issues.append("Missing invoice number")
    if input_json.get("amount", 0) <= 0:
        issues.append("Invalid amount")

    return {
        "validation_result": {
            "valid": len(issues) == 0,
            "issues": issues,
            "invoice_number": input_json.get("invoice_number")
        }
    }

Data Enrichment with Computed Fields

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

def execute(input_json, meta):
    try:
        revenue = float(input_json.get("revenue", 0))
        cost = float(input_json.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"

        return {
            "enriched_record": {
                "margin_percent": round(margin, 2),
                "performance_category": category
            }
        }
    except Exception as e:
        return {
            "enriched_record": {
                "margin_percent": None,
                "performance_category": "error",
                "error_message": str(e)
            }
        }

---

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.
• Always wrap output in a unique key: Return values must be inside a named wrapper (e.g., {"my_result": {...}}). Returning bare data at the top level risks overwriting fields from other nodes with the same names.
• 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 the output key descriptively: Use a key that reflects the script’s purpose, such as loyalty_result, validation_result, or enriched_record, rather than a generic name like result or output.

---

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 that the return value is structured as a dictionary with a unique top-level wrapper key. If data is returned without a wrapper, it may conflict with existing field names or not be passed correctly. Also confirm that the downstream node is referencing the correct field path, including the wrapper key (e.g., custom_result.score not just score).

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: Output contains loyalty_result.points_earned with the correct calculated value.
• Given: Input with missing required fields passed to a validation script - Expected: Output contains validation_result.valid = false and validation_result.issues listing the missing fields.
• 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.

---

Related Nodes

• 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.