Dap Scripts

Blueprint Script Authoring — Dell Automation Platform

Runtime Mode

You are running in IDE mode with shell access. Use the bpa CLI for BPA operations. Do not use the function interface unless explicitly instructed by the host.

Python scripts are used in blueprint operations when the logic is custom and doesn't fit an existing plugin. The script is referenced by a relative path in the operation's implementation: field and is executed by the DAP agent.

When to Use Scripts vs Other Execution Methods

Default preference: Use an existing plugin when one fits. Use a Python script when you need custom orchestrator-side logic (accessing ctx, reading/writing runtime_properties, calling the REST client, or implementing complex conditional flows that plugins don't support).

Required imports (MANDATORY)

Every DAP Python script must use the dell module — never cloudify or nativeedge.

from dell import ctx                                    # ctx object: node, instance, logger, operation
from dell.state import ctx_parameters as inputs         # operation inputs passed from the blueprint
from dell.exceptions import NonRecoverableError, RecoverableError  # error handling

The json.load(sys.stdin) pattern used in older Cloudify scripts is forbidden — always use inputs.get() instead.

Full canonical script pattern

from dell import ctx
from dell.state import ctx_parameters as inputs
from dell.exceptions import NonRecoverableError, RecoverableError
def main():
    # 1. Read operation inputs (passed via the blueprint's operation inputs: section)
    cluster_name = inputs.get('cluster_name')
    config_data  = inputs.get('config_data', {})
    credential   = inputs.get('credential')   # resolved from get_secret in blueprint

    if not cluster_name:
        raise NonRecoverableError("'cluster_name' is required")

    # 2. Log via ctx.logger — integrates with DAPO's event stream
    ctx.logger.info(f"Starting operation for cluster: {cluster_name}")
    ctx.logger.info(f"Node: {ctx.node.id}  Instance: {ctx.instance.id}  Op: {ctx.operation.name}")

    # 3. Read static node properties (defined in blueprint node_templates.properties)
    some_property = ctx.node.properties.get('some_property', 'default')

    try:
        result = do_work(cluster_name, config_data, credential)

        # 4. Write outputs to runtime_properties — downstream nodes read these
        #    via  get_attribute: [this_node, result]
        ctx.instance.runtime_properties['result']       = result
        ctx.instance.runtime_properties['status']       = 'completed'
        ctx.instance.runtime_properties['cluster_name'] = cluster_name

        ctx.logger.info("Operation completed successfully")

    except TemporaryFailure as e:
        # 5. RecoverableError → DAPO retries the operation (respects max_retries)
        raise RecoverableError(f"Temporary failure, will retry: {e}")

    except Exception as e:
        # 6. NonRecoverableError → operation fails immediately, no retry
        ctx.logger.error(f"Fatal error: {e}")
        raise NonRecoverableError(f"Operation failed: {e}")
class TemporaryFailure(Exception):
    pass
if __name__ == "__main__":
    main()

ctx object reference

Error types

Use RecoverableError for transient conditions (API timeouts, resource not ready yet). Use NonRecoverableError for configuration errors, missing inputs, or any condition that retrying won't fix.

How operation inputs flow from blueprint to script

Blueprint:

node_templates:
  my_node:
    type: dell.nodes.Root
    interfaces:
      dell.interfaces.lifecycle:
        create:
          implementation: scripts/create.py
          max_retries: 3
          inputs:
            cluster_name: { get_input: cluster_name }
            credential:   { get_secret: { get_input: my_secret_name } }
            config_data:  { get_attribute: [other_node, result] }

Script:

cluster_name = inputs.get('cluster_name')   # from get_input
credential   = inputs.get('credential')     # already-resolved secret value
config_data  = inputs.get('config_data')    # runtime_properties of other_node

The inputs: section of the operation is the bridge. Everything passed there is available via inputs.get() in the script. The orchestrator resolves all intrinsic functions (get_input, get_secret, get_attribute) before invoking the script.

For the full lifecycle sequence (precreate through postdelete) and cross-plugin patterns like ND-003, see knowledge/docs/general/lifecycle-operations.md.

Reporting task results (ctx.returns)

For most lifecycle operations, writing to ctx.instance.runtime_properties is the output — no explicit return is needed.

When an operation must surface a task result value to the orchestrator (most commonly check_drift, but also any custom workflow task whose return value the workflow logic reads), use ctx.returns(<payload>):

ctx.returns({'drift': True, 'diff_count': 3})   # orchestrator reads this
ctx.returns(None)                                 # explicit "no result"

Scripts run in a subprocess — a plain Python return statement is ignored by the orchestrator. ctx.returns() is the only channel whose value is captured as the task result. The payload must be JSON-serializable.

Never use return <value> to communicate with the orchestrator, and never mutate runtime_properties as a substitute return channel in operations that use ctx.returns (e.g. check_drift must be read-only).

For the check_drift-specific contract and the configuration_drift payload shape consumed by the update workflow UI, see dap-deployment-update/SKILL.md → check_drift return payload.

Relationship operations

For operations defined in source_interfaces or target_interfaces on a relationship, use ctx.source and ctx.target instead of ctx.node / ctx.instance:

source_props  = ctx.source.node.properties
target_rp     = ctx.target.instance.runtime_properties
ctx.source.instance.runtime_properties['connected_to'] = ctx.target.node.id

Accessing the manager REST client (advanced)

If the script needs to call the DAPO REST API directly (e.g. to read or write secrets programmatically):

from dell import ctx, manager

rest_client = manager.get_rest_client()
secret = rest_client.secrets.get('my_secret_name')
value  = secret['value']

Script location

Scripts must be inside the blueprint archive. Reference them by path relative to the blueprint root:

implementation: scripts/configure.py        # scripts/ at blueprint root
implementation: application/scripts/run.py  # nested path — fine

Keep all scripts in a scripts/ directory (or subdirectory). Do not inline Python code in YAML.

Forbidden patterns

Retrieval

Search for script-related topics:

bpa knowledge docs search "blueprint script"
bpa knowledge docs search "runtime_properties"