Skip to main content
The Python action executes custom Python code within your workflow. Use it when built-in actions can’t handle your specific transformation, calculation, or logic requirements—or when you prefer Python over JavaScript.

How it works

  1. Write Python code in the built-in editor
  2. Access workflow data directly via the nodes dictionary
  3. Assign the output to a variable named result
  4. Use the output in downstream actions via expressions
# Example: Calculate lead score
company_size = nodes["enrich"]["company_size"]
funding = nodes["enrich"]["funding"]
industry = nodes["enrich"]["industry"]

score = 0
if company_size > 100:
    score += 30
if funding > 10000000:
    score += 40
if industry == "Technology":
    score += 20

tier = "hot" if score >= 70 else "warm" if score >= 40 else "cold"

result = {"score": score, "tier": tier}

Accessing workflow data

Inside Python actions, you have direct access to all workflow data through global dictionaries—no need for {{}} expression syntax.

The nodes dictionary

Access output from any previous node in your workflow:
# Access the start node (tool/play inputs)
company_domain = nodes["start"]["company_domain"]
contact_email = nodes["start"]["contact_email"]

# Access output from an enrichment node
company_name = nodes["enrich"]["company_name"]
employee_count = nodes["enrich"]["employee_count"]

# Access results from a search node
contacts = nodes["search"]["results"]

# Access the full output dictionary from any node
all_enrich_data = nodes["enrich"]
ExpressionDescription
nodes["start"]The input data passed to the tool or play
nodes["node_name"]Output from a specific node by its name
nodes["node_name"]["foo"]Access a specific property from a node’s output

The parentNodes dictionary

When your Python action is inside a Group node, use parentNodes to access data from nodes outside the group:
# Inside a Group: access the current item being iterated
current_contact = nodes["group"]["item"]
current_index = nodes["group"]["index"]

# Access data from nodes OUTSIDE the group
original_company = parentNodes["enrich"]["company_name"]
all_settings = parentNodes["config"]["settings"]
ExpressionDescription
nodes["group"]["item"]The current item in the loop
nodes["group"]["index"]The zero-based index of the current item
parentNodes["node_name"]Access nodes outside the current group
parentNodes["start"]Access the original tool/play input
Use parentNodes when you need to reference data that was computed before the Group started—like company-level data when iterating over contacts.

Returning data

Assign your output to a variable named result. This becomes the node’s output:
# Return a simple value
result = 42

# Return a dictionary
result = {
    "processed": True,
    "message": "Success",
    "timestamp": "2024-01-15T10:30:00Z"
}

# Return a list
contacts = nodes["search"]["results"]
result = [
    {**item, "processed": True}
    for item in contacts
]
Access the returned data in downstream nodes:
ExpressionReturns
{{nodes.python.result}}The full returned value
{{nodes.python.processed}}true (if dictionary returned)
{{nodes.python.message}}"Success"
Replace python with the actual name of your Python node.

Common use cases

Data transformation

Reshape data between nodes:
contacts = nodes["search"]["results"]

# Transform list of contacts into a formatted list
result = [
    {
        "full_name": f"{c['first_name']} {c['last_name']}",
        "email": c["email"].lower(),
        "domain": c["email"].split("@")[1]
    }
    for c in contacts
]

Complex calculations

Implement business logic that doesn’t fit built-in actions:
deal_size = nodes["start"]["deal_size"]
contract_length = nodes["start"]["contract_length"]
is_enterprise = nodes["start"]["is_enterprise"]

# Calculate annual contract value with enterprise multiplier
acv = deal_size * 12
if contract_length >= 24:
    acv *= 0.9  # Annual discount
if is_enterprise:
    acv *= 1.2  # Enterprise premium

tier = "enterprise" if acv >= 100000 else "mid-market" if acv >= 25000 else "smb"

result = {
    "acv": round(acv),
    "tier": tier
}

String manipulation

Format and parse text data:
raw_address = nodes["enrich"]["address"]

# Parse a raw address string into components
parts = [p.strip() for p in raw_address.split(",")]

state_zip = parts[2].split(" ") if len(parts) > 2 else ["", ""]

result = {
    "street": parts[0] if len(parts) > 0 else "",
    "city": parts[1] if len(parts) > 1 else "",
    "state": state_zip[0] if state_zip else "",
    "zip": state_zip[1] if len(state_zip) > 1 else "",
    "country": parts[3] if len(parts) > 3 else "USA"
}

List operations

Filter, sort, and aggregate lists:
leads = nodes["search"]["results"]

# Filter high-value leads and sort by score
qualified = sorted(
    [lead for lead in leads if lead.get("score", 0) >= 50 and lead.get("email")],
    key=lambda x: x.get("score", 0),
    reverse=True
)

result = {
    "qualified_leads": qualified,
    "total": len(qualified),
    "top_lead": qualified[0] if qualified else None
}

JSON parsing

Parse JSON strings from API responses:
import json

api_response = nodes["http_request"]["body"]

try:
    data = json.loads(api_response)
    result = {
        "success": True,
        "data": data.get("results", []),
        "count": data.get("total", 0)
    }
except json.JSONDecodeError as e:
    result = {
        "success": False,
        "error": "Invalid JSON response",
        "raw": api_response
    }

Date operations

Work with dates and timestamps:
from datetime import datetime, timedelta

created_at = nodes["start"]["created_at"]
trial_days = nodes["start"]["trial_days"]

created_date = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
trial_end = created_date + timedelta(days=trial_days)

now = datetime.now(created_date.tzinfo)
days_remaining = (trial_end - now).days

result = {
    "trial_end_date": trial_end.strftime("%Y-%m-%d"),
    "days_remaining": max(0, days_remaining),
    "is_expired": days_remaining <= 0,
    "urgency": "high" if days_remaining <= 3 else "medium" if days_remaining <= 7 else "low"
}

Using parentNodes in a Group

When processing items inside a Group, combine item data with parent context:
# Inside a Group iterating over contacts
contact = nodes["group"]["item"]
index = nodes["group"]["index"]

# Access company data from outside the group
company_name = parentNodes["enrich"]["company_name"]
company_industry = parentNodes["enrich"]["industry"]

result = {
    "contact_name": f"{contact['first_name']} {contact['last_name']}",
    "company": company_name,
    "industry": company_industry,
    "position": index + 1,
    "email_subject": f"{contact['first_name']}, a quick note about {company_name}"
}

Available Python features

The Python action runs via Pyodide, a Python runtime compiled to WebAssembly. This provides access to a full Python environment with many packages.

Supported features

FeatureSupportNotes
Python 3.11+ syntaxf-strings, walrus operator, match/case
List comprehensionsFiltering, mapping, nested comprehensions
Dictionary methods.get(), .items(), .keys(), .values()
String methods.split(), .join(), .replace(), etc.
json moduleloads, dumps
math moduleAll standard functions
datetime moduleDate and time manipulation
re moduleRegular expressions
async/awaitAsynchronous code supported

Available packages

Packages are automatically loaded when you import them. Simply use import statements and Pyodide will fetch the required packages:
import numpy as np
import pandas as pd

# Your code using numpy and pandas
data = nodes["search"]["results"]
df = pd.DataFrame(data)
result = df.describe().to_dict()
Popular packages available include:
PackageDescriptionExample use
numpyNumerical computingArray operations, math
pandasData analysis and manipulationDataFrames, CSV processing
regexAdvanced regular expressionsComplex pattern matching
beautifulsoup4HTML/XML parsingWeb scraping, HTML extraction
pyyamlYAML parsingConfig file processing
python-dateutilDate parsing utilitiesFlexible date parsing
requestsHTTP library (limited)Simple HTTP requests
Pyodide supports hundreds of packages. If an import fails, the package may not be available in the Pyodide environment. Check the Pyodide packages list for availability.

Not supported

FeatureReason
os.system()Blocked for security
Direct JS accessBlocked for security
File system operationsSandboxed environment
subprocessNo shell access
Network socketsUse HTTP-based alternatives
For HTTP requests, consider using dedicated HTTP Request or integration actions for better error handling and monitoring.

Error handling

Try-except blocks

Handle errors gracefully within your script:
data = nodes["api"]["response"]

try:
    import json
    parsed = json.loads(data)
    processed = complex_operation(parsed)
    result = {"success": True, "data": processed}
except Exception as e:
    result = {
        "success": False,
        "error": str(e),
        "fallback": "Default value"
    }

Validation

Validate data before processing:
email = nodes["start"].get("email")
company_size = nodes["start"].get("company_size")

# Validate required fields
if not email or not isinstance(email, str):
    result = {"error": "Invalid email", "valid": False}
elif "@" not in email:
    result = {"error": "Email must contain @", "valid": False}
elif company_size is not None and not isinstance(company_size, (int, float)):
    result = {"error": "Company size must be a number", "valid": False}
else:
    result = {"valid": True}
Always validate data when it comes from external sources or user input. Return structured error dictionaries so downstream nodes can handle failures.

Debugging tips

Use descriptive returns

Instead of returning just a result, include debug information:
items = nodes["search"]["results"]

processed = [i for i in items if i.get("active")]

result = {
    "data": processed,
    "debug": {
        "input_count": len(items),
        "output_count": len(processed),
        "filtered_out": len(items) - len(processed)
    }
}

Test with sample data

Before using in production:
  1. Run the tool with test inputs
  2. Check the Python node’s output in the execution log
  3. Verify the data structure matches what downstream nodes expect

Check for missing keys

Python’s KeyError can cause failures when accessing missing dictionary keys:
user = nodes["enrich"]["user"]

# Dangerous: might raise KeyError
email = user["contact"]["email"]

# Safer: use .get() with defaults
email = user.get("contact", {}).get("email", "[email protected]")

Performance considerations

PracticeImpact
Keep scripts simpleFaster execution, easier debugging
Avoid deep nestingBetter readability and performance
Limit list operationsLarge lists can slow execution
Use early returnsSkip unnecessary processing
Minimize package importsPackages load on first import
Scripts have a default timeout. Complex operations on large datasets may need optimization or should be split across multiple nodes.

Best practices

Each script should do one thing well. If your script is growing complex, consider splitting it into multiple Python nodes or defining helper functions within the script.
Always return dictionaries with clear key names. This makes it easier to access data in downstream nodes and debug issues.
External data often contains None or missing keys. Use .get() with default values to handle these gracefully.
Check that required fields exist and have the expected types before performing operations. Return early with error information when validation fails.
Scripts should be pure functions—given the same inputs, they should always return the same outputs. Don’t rely on external state.

Example: Lead qualification script

A complete example that qualifies leads based on multiple criteria:
enrich_data = nodes["enrich"]

company_size = enrich_data.get("company_size", 0)
industry = enrich_data.get("industry", "")
funding_amount = enrich_data.get("funding_amount", 0)
has_website = enrich_data.get("has_website", False)
technologies = enrich_data.get("technologies", "")

# Define scoring rules
INDUSTRY_SCORES = {
    "Technology": 25,
    "Finance": 20,
    "Healthcare": 20,
    "E-commerce": 15
}

TECH_KEYWORDS = ["react", "node", "python", "aws", "kubernetes"]

# Calculate component scores
score = 0
factors = []

# Company size scoring
if company_size >= 500:
    score += 30
    factors.append("Enterprise size (+30)")
elif company_size >= 100:
    score += 20
    factors.append("Mid-market size (+20)")
elif company_size >= 20:
    score += 10
    factors.append("SMB size (+10)")

# Industry scoring
industry_score = INDUSTRY_SCORES.get(industry, 5)
score += industry_score
factors.append(f"Industry: {industry} (+{industry_score})")

# Funding scoring
if funding_amount >= 50000000:
    score += 25
    factors.append("Series C+ funding (+25)")
elif funding_amount >= 10000000:
    score += 15
    factors.append("Series A/B funding (+15)")

# Technology match scoring
tech_list = technologies.lower()
matched_tech = [t for t in TECH_KEYWORDS if t in tech_list]
if matched_tech:
    tech_score = len(matched_tech) * 5
    score += tech_score
    factors.append(f"Tech stack match: {', '.join(matched_tech)} (+{tech_score})")

# Website presence
if has_website:
    score += 5
    factors.append("Has website (+5)")

# Determine qualification tier
if score >= 70:
    tier, priority = "A", "high"
elif score >= 50:
    tier, priority = "B", "medium"
elif score >= 30:
    tier, priority = "C", "low"
else:
    tier, priority = "D", "nurture"

result = {
    "score": score,
    "tier": tier,
    "priority": priority,
    "qualified": score >= 50,
    "factors": factors,
    "summary": f"{tier}-tier lead (score: {score}) - {priority} priority"
}