Code step - Anyreach

The Code step runs Python in a sandboxed environment. Use it when no other step type fits — complex data transformations, custom logic, calling Python client libraries, or looping over arrays.

When to use it

Use Code when	Use something else when
Transforming data shapes (reshaping JSON, computing derived fields)	A single HTTP call is enough — use HTTP API step
Looping over an array	A simple branch is enough — use Condition step
Calling a Python client library (`boto3`, `stripe`, etc.)	The integration exists in Pipedream — use Action step
Logic too complex for a JSONata expression	The transform is one line — keep it inline

How it works

Your code runs inside a function called run(ctx). The current workflow context is injected as ctx — a Python dict containing every upstream step’s output. Whatever run returns becomes this step’s output in ctx.

def run(ctx):
    contact = ctx["Lookup contact"]["body"]
    return {
        "display_name": f"{contact['first_name']} {contact['last_name']}",
        "is_premium": contact.get("tier") in ("gold", "platinum"),
    }

Inspector

The Code step inspector has two tabs:

Configuration — Draft your Python code directly in the provided editor. For an enhanced experience, click the purple Open IDE button to access a full-screen coding environment.
- Here, you can write and edit your Python logic with greater visibility.
- Specify any external Python packages your code requires—these will be installed into the sandboxed environment automatically.
- Here you can write your pythion code
- Enter packages that needs to be installed in the sandbox
Output — After saving your code in the editor, this tab displays the structure of the data returned by your run function. You can reference these output fields as variables in downstream steps using the workflow context.

Accessing context

The ctx dict contains all upstream step outputs keyed by step name:

def run(ctx):
    # Access trigger/input data
    email = ctx["input"]["customer_email"]
    
    # Access upstream step output
    contact = ctx["Lookup contact"]["body"]
    status = ctx["Lookup contact"]["status_code"]
    
    return {"email": email, "name": contact["first_name"]}

Examples

Reshape an HTTP response

def run(ctx):
    contact = ctx["Lookup contact"]["body"]
    return {
        "display_name": f"{contact['first_name']} {contact['last_name']}",
        "tags": [t["name"] for t in contact.get("tags", [])],
        "is_premium": contact.get("tier") in ("gold", "platinum"),
    }

Call a third-party SDK

import stripe

def run(ctx):
    stripe.api_key = ctx["input"]["stripe_key"]
    customer = stripe.Customer.retrieve(ctx["input"]["customer_id"])
    return {
        "email": customer.email,
        "balance": customer.balance,
    }

(Add stripe to the packages list in the inspector.)

Limits

Limit	Value
Execution timeout	300 seconds
Max return value size	10 MB
Network	Outbound HTTPS allowed
Filesystem	Writable scratch space — not persisted between runs

Considerations

Don’t cache state between runs. Each run gets a fresh sandbox.
Let exceptions propagate. Caught-and-swallowed exceptions hide bugs. If you handle an error, return a structured {"failed": True, "reason": "..."} and branch on it with a Condition step.
Avoid heavy packages on the critical path. pandas and numpy add cold-start latency. For simple data shaping, write it without dependencies.

Workflows

Documentation Index

​When to use it

​How it works

​Inspector

​Accessing context

​Examples

​Reshape an HTTP response

​Call a third-party SDK

​Limits

​Considerations