Workflow agent code step examples

In Optimizely Opal, these examples provide handler code to copy into a code step and adapt. Every example lists the input variables to declare and the sample data for a test run. Each one also gives the handler and the value it returns.

For the configuration fields and the handler contract these examples rely on, go to Workflow agent code steps.

Filter and aggregate data

This example filters a list of orders against a dollar threshold and returns the matching orders with a total.

Declare the following input variables:

Enter the following sample data:

{
  "orders": [
    { "id": "ORD-001", "customer": "Acme Corp", "amount": 250 },
    { "id": "ORD-002", "customer": "Globex Inc", "amount": 75 },
    { "id": "ORD-003", "customer": "Initech", "amount": 500 },
    { "id": "ORD-004", "customer": "Umbrella LLC", "amount": 30 }
  ],
  "threshold": 100
}

Use the following Python handler:

def handler(input, ctx):
    orders = input["orders"]
    threshold = input["threshold"]

    filtered = [o for o in orders if o["amount"] > threshold]
    total = sum(o["amount"] for o in filtered)

    return {
        "filtered_orders": filtered,
        "count": len(filtered),
        "total": total,
        "summary": f"Filtered {len(filtered)} of {len(orders)} orders above ${threshold}"
    }

The handler returns the following value:

{
  "filtered_orders": [
    { "id": "ORD-001", "customer": "Acme Corp", "amount": 250 },
    { "id": "ORD-003", "customer": "Initech", "amount": 500 }
  ],
  "count": 2,
  "total": 750
}

The handler keeps only the orders with an amount greater than threshold, computes the total, and returns a structured dictionary. Opal lifts the summary value into the execution log and removes it from the return value.

Format output for a downstream agent

This example reshapes raw data into a structured brief that a downstream specialized agent consumes. An upstream agent gathers customer feedback entries, and the code step groups them for a summarization agent.

Declare the following input variables:

Enter the following sample data:

{
  "feedback_entries": [
    { "source": "Support ticket", "text": "Dashboard loads slowly when filtering by date range.", "sentiment": "negative" },
    { "source": "NPS survey", "text": "Love the reporting features. Very intuitive.", "sentiment": "positive" },
    { "source": "Support ticket", "text": "Export to CSV is missing column headers.", "sentiment": "negative" },
    { "source": "NPS survey", "text": "Would be great to have dark mode.", "sentiment": "neutral" }
  ],
  "product_name": "Acme Tech"
}

Use the following Python handler:

def handler(input, ctx):
    entries = input["feedback_entries"]
    product = input["product_name"]

    by_sentiment = {}
    for entry in entries:
        sentiment = entry.get("sentiment", "unknown")
        by_sentiment.setdefault(sentiment, []).append(entry)

    sections = []
    for sentiment in ["negative", "neutral", "positive"]:
        items = by_sentiment.get(sentiment, [])
        if not items:
            continue
        lines = [f"- [{item['source']}] {item['text']}" for item in items]
        sections.append(f"### {sentiment.capitalize()} ({len(items)})\n" + "\n".join(lines))

    brief = f"# Customer Feedback Summary for {product}\n\n"
    brief += f"Total entries: {len(entries)}\n\n"
    brief += "\n\n".join(sections)

    return {
        "formatted_brief": brief,
        "counts": {s: len(items) for s, items in by_sentiment.items()},
        "summary": f"Formatted {len(entries)} feedback entries for {product}"
    }

The handler returns the following value:

{
  "formatted_brief": "# Customer Feedback Summary for Acme Tech\n\nTotal entries: 4\n\n### Negative (2)\n- [Support ticket] Dashboard loads slowly when filtering by date range.\n- [Support ticket] Export to CSV is missing column headers.\n\n### Neutral (1)\n- [NPS survey] Would be great to have dark mode.\n\n### Positive (1)\n- [NPS survey] Love the reporting features. Very intuitive.",
  "counts": { "negative": 2, "positive": 1, "neutral": 1 }
}

A downstream specialized agent receives formatted_brief as its input. The agent produces a polished summary without parsing raw data itself.

Validate and clean data

This example validates a list of email addresses and separates the valid entries from the invalid ones.

Declare the following input variable:

Enter the following sample data:

{
 "emails": [
   "alice@example.com",
   "bob@",
   "charlie@company.org",
   "not-an-email",
   "dana@sub.domain.co.uk",
   "@missing-local.com",
   "eve@example"
 ]
}

Use the following Python handler:

import re

def handler(input, ctx):
    emails = input["emails"]
    pattern = re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")

    valid = []
    invalid = []

    for email in emails:
        email_clean = email.strip().lower()
        if pattern.match(email_clean):
            valid.append(email_clean)
        else:
            invalid.append(email_clean)

    return {
        "valid_emails": valid,
        "invalid_emails": invalid,
        "valid_count": len(valid),
        "invalid_count": len(invalid),
        "summary": f"Validated {len(emails)} emails: {len(valid)} valid, {len(invalid)} invalid"
   }

The handler returns the following value:

{
  "valid_emails": ["alice@example.com", "charlie@company.org", "dana@sub.domain.co.uk"],
  "invalid_emails": ["bob@", "not-an-email", "@missing-local.com", "eve@example"],
  "valid_count": 3,
  "invalid_count": 4
}

Downstream steps read valid_emails and proceed with verified addresses only.

Parse CSV text into structured data

This example parses a comma-separated values (CSV) string from an upstream step into an array of objects. It also converts numeric columns.

Declare the following input variable:

Enter the following sample data:

{
  "csv_text": "name,department,salary\nAlice,Engineering,95000\nBob,Marketing,72000\nCharlie,Engineering,105000\nDana,Sales,68000"
}

Use the following Python handler:

import csv
import io

def handler(input, ctx):
    csv_text = input["csv_text"]
    reader = csv.DictReader(io.StringIO(csv_text))

    records = []
    for row in reader:
        cleaned = {}
        for key, value in row.items():
            try:
                cleaned[key] = int(value)
            except (ValueError, TypeError):
                try:
                    cleaned[key] = float(value)
                except (ValueError, TypeError):
                    cleaned[key] = value
        records.append(cleaned)

    columns = reader.fieldnames or []

    return {
        "records": records,
        "columns": columns,
        "row_count": len(records),
        "summary": f"Parsed {len(records)} rows with {len(columns)} columns"
    }

The handler returns the following value:

{
  "records": [
    { "name": "Alice", "department": "Engineering", "salary": 95000 },
    { "name": "Bob", "department": "Marketing", "salary": 72000 },
    { "name": "Charlie", "department": "Engineering", "salary": 105000 },
    { "name": "Dana", "department": "Sales", "salary": 68000 }
  ],
  "columns": ["name", "department", "salary"],
  "row_count": 4
}

The handler uses the csv module from the Python standard library. It detects the header row and converts numeric values from strings. Downstream steps filter, sort, or aggregate the returned array of dictionaries directly.

Score and rank items with JavaScript

This example scores a list of sales leads against weighted criteria and returns them ranked by score.

Declare the following input variables:

<>

Enter the following sample data:

{
  "leads": [
    { "name": "Acme Corp", "company_size": 500, "engagement_score": 85, "has_budget": true },
    { "name": "Startup Inc", "company_size": 15, "engagement_score": 92, "has_budget": false },
    { "name": "BigCo Ltd", "company_size": 5000, "engagement_score": 40, "has_budget": true },
    { "name": "MidRange LLC", "company_size": 200, "engagement_score": 70, "has_budget": true }
  ],
  "weights": {
    "company_size": 0.3,
    "engagement": 0.5,
    "budget": 0.2
  }
}

Use the following JavaScript handler:

async function handler(input, ctx) {
    const leads = input.leads;
    const weights = input.weights;

    function normalizeSize(size) {
        return Math.min(size / 1000, 1) * 100;
    }

    const scored = leads.map(lead => {
        const sizeScore = normalizeSize(lead.company_size) * weights.company_size;
        const engagementScore = lead.engagement_score * weights.engagement;
        const budgetScore = (lead.has_budget ? 100 : 0) * weights.budget;

        return {
            ...lead,
            score: Math.round(sizeScore + engagementScore + budgetScore),
            breakdown: {
                size: Math.round(sizeScore),
                engagement: Math.round(engagementScore),
                budget: Math.round(budgetScore)
            }
        };
    });

    scored.sort((a, b) => b.score - a.score);

    return {
        ranked_leads: scored,
        top_lead: scored[0].name,
        summary: `Scored ${scored.length} leads. Top lead: ${scored[0].name}`
    };
}

The handler returns the following value:

{
  "ranked_leads": [
    {
      "name": "Acme Corp",
      "company_size": 500,
      "engagement_score": 85,
      "has_budget": true,
      "score": 78,
      "breakdown": { "size": 15, "engagement": 43, "budget": 20 }
    },
    {
      "name": "BigCo Ltd",
      "company_size": 5000,
      "engagement_score": 40,
      "has_budget": true,
      "score": 70,
      "breakdown": { "size": 30, "engagement": 20, "budget": 20 }
    },
    {
      "name": "MidRange LLC",
      "company_size": 200,
      "engagement_score": 70,
      "has_budget": true,
      "score": 61,
      "breakdown": { "size": 6, "engagement": 35, "budget": 20 }
    },
    {
      "name": "Startup Inc",
      "company_size": 15,
      "engagement_score": 92,
      "has_budget": false,
      "score": 47,
      "breakdown": { "size": 0, "engagement": 46, "budget": 0 }
    }
  ],
  "top_lead": "Acme Corp"
}

The weights variable uses Fixed mode, so you adjust the scoring without editing code. The breakdown object records how each criterion contributed to the final score.

Generate a summary report with JavaScript

This example computes campaign metrics and produces both a readable report and structured data.

Declare the following input variable:

Enter the following sample data:

{
  "campaign_results": [
    { "name": "Summer Sale", "impressions": 50000, "clicks": 2500, "conversions": 150 },
    { "name": "Product Launch", "impressions": 120000, "clicks": 8400, "conversions": 620 },
    { "name": "Retargeting Q3", "impressions": 30000, "clicks": 1800, "conversions": 210 },
    { "name": "Brand Awareness", "impressions": 200000, "clicks": 6000, "conversions": 90 }
  ]
}

Use the following JavaScript handler:

async function handler(input, ctx) {
    const campaigns = input.campaign_results;

    const totalImpressions = campaigns.reduce((sum, c) => sum + c.impressions, 0);
    const totalClicks = campaigns.reduce((sum, c) => sum + c.clicks, 0);
    const totalConversions = campaigns.reduce((sum, c) => sum + c.conversions, 0);

    const avgCtr = totalImpressions > 0
        ? ((totalClicks / totalImpressions) * 100).toFixed(2) : "0.00";
    const avgConvRate = totalClicks > 0
        ? ((totalConversions / totalClicks) * 100).toFixed(2) : "0.00";

    const withRates = campaigns.map(c => ({
        ...c,
        ctr: c.impressions > 0 ? ((c.clicks / c.impressions) * 100).toFixed(2) : "0.00",
        conversion_rate: c.clicks > 0 ? ((c.conversions / c.clicks) * 100).toFixed(2) : "0.00"
    }));

    withRates.sort((a, b) => parseFloat(b.conversion_rate) - parseFloat(a.conversion_rate));
    const best = withRates[0];
    const worst = withRates[withRates.length - 1];

    return {
        report_text: `Campaign Performance Report\nTotal: ${campaigns.length} campaigns, ${avgCtr}% CTR, ${avgConvRate}% conv rate\nTop: ${best.name} (${best.conversion_rate}%)\nNeeds attention: ${worst.name} (${worst.conversion_rate}%)`,
        metrics: { totalImpressions, totalClicks, totalConversions, avgCtr, avgConvRate },
        campaign_details: withRates,
        best_campaign: best.name,
        worst_campaign: worst.name,
        summary: `Report for ${campaigns.length} campaigns: ${avgCtr}% CTR`
    };
}

The handler returns the following value:

{
  "report_text": "Campaign Performance Report\nTotal: 4 campaigns, 4.68% CTR, 5.72% conv rate\nTop: Retargeting Q3 (11.67%)\nNeeds attention: Brand Awareness (1.50%)",
  "metrics": {
    "totalImpressions": 400000,
    "totalClicks": 18700,
    "totalConversions": 1070,
    "avgCtr": "4.68",
    "avgConvRate": "5.72"
  },
  "campaign_details": [
    { "name": "Retargeting Q3", "impressions": 30000, "clicks": 1800, "conversions": 210, "ctr": "6.00", "conversion_rate": "11.67" },
    { "name": "Product Launch", "impressions": 120000, "clicks": 8400, "conversions": 620, "ctr": "7.00", "conversion_rate": "7.38" },
    { "name": "Summer Sale", "impressions": 50000, "clicks": 2500, "conversions": 150, "ctr": "5.00", "conversion_rate": "6.00" },
    { "name": "Brand Awareness", "impressions": 200000, "clicks": 6000, "conversions": 90, "ctr": "3.00", "conversion_rate": "1.50" }
  ],
  "best_campaign": "Retargeting Q3",
  "worst_campaign": "Brand Awareness"
}

A downstream agent uses report_text as context for generating insights. Other steps branch on the values in metrics.

Read an input file

This example reads a file uploaded to the conversation, parses it as comma-separated values, and returns summary data. It falls back to listing the available files when the one you named is absent.

Declare the following input variable:

Enter the following sample data:

"filename": "sales_q3.csv"

Use the following Python handler:

import csv
import os

def handler(input, ctx):
    filename = input["filename"]
    attachments = os.path.join(ctx["input_dir"], "attachments")
    filepath = os.path.join(attachments, filename)

    if not os.path.isfile(filepath):
        available = sorted(os.listdir(attachments)) if os.path.isdir(attachments) else []
        return {
            "error": f"File not found: {filename}",
            "available_files": available,
            "summary": f"{filename} is not in the attachments directory"
        }

    with open(filepath, "r") as f:
        reader = csv.DictReader(f)
        records = list(reader)
        columns = reader.fieldnames or []

    return {
        "filename": filename,
        "row_count": len(records),
        "columns": columns,
        "first_rows": records[:3],
        "summary": f"Read {filename}: {len(records)} rows, {len(columns)} columns"

Given a file with region, sales, and target columns, the handler returns the following value:

{
  "filename": "sales_q3.csv",
  "row_count": 4,
  "columns": ["region", "sales", "target"],
  "first_rows": [
    { "region": "North", "sales": "45000", "target": "50000" },
    { "region": "South", "sales": "62000", "target": "55000" },
    { "region": "East", "sales": "38000", "target": "40000" }
  ]
}

A test run has no input files. This example needs a full workflow execution with a file uploaded to Opal chat. The fallback branch reports what did arrive, which turns a missing file into readable output rather than a failed step.

csv.DictReader returns every value as a string, so sales arrives as "45000" rather than 45000. The Parse CSV text into structured data example shows how to convert the numeric columns.

Write output files

This example writes results to the output directory so downstream steps read them as files.

Declare the following input variables:

Enter the following sample data:

{
  "records": [
    { "id": 1, "name": "Alice", "score": 95 },
    { "id": 2, "name": "Bob", "score": 82 },
    { "id": 3, "name": "Charlie", "score": 91 }
  ],
  "filename": "results.csv"
}

Use the following Python handler:

import csv
import json
import os

def handler(input, ctx):
    records = input["records"]
    filename = input["filename"]

    if not records:
        return {"file_path": None, "summary": "No records to export"}

    output_path = os.path.join(ctx["output_dir"], filename)
    columns = list(records[0].keys())

    with open(output_path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=columns)
        writer.writeheader()
        writer.writerows(records)

    json_path = os.path.join(ctx["output_dir"], filename.replace(".csv", ".json"))
    with open(json_path, "w") as f:
        json.dump(records, f, indent=2)

    return {
        "csv_file": output_path,
        "json_file": json_path,
        "row_count": len(records),
        "columns": columns,
        "summary": f"Exported {len(records)} records to {filename}"
    }

The handler returns the following value:

{
  "csv_file": "/workspace/output/results.csv",
  "json_file": "/workspace/output/results.json",
  "row_count": 3,
  "columns": ["id", "name", "score"]
}

The handler writes both files to the path in ctx["output_dir"]. Those files display in the Output Files section of the execution details panel.

Write output files with Bash

This example uses Bash and jq to process JavaScript Object Notation (JSON) input and write files. File generation is the primary use case for a Bash code step, because a Bash step always returns null.

Declare the following input variable:

<>

Enter the following sample data:

{
  "report_data": {
    "title": "Weekly Sales Summary",
    "items": [
      { "region": "North", "sales": 45000, "target": 50000 },
      { "region": "South", "sales": 62000, "target": 55000 },
      { "region": "East", "sales": 38000, "target": 40000 },
      { "region": "West", "sales": 51000, "target": 48000 }
    ]
  }
}

Use the following Bash handler:

#!/usr/bin/env bash

TITLE=$(jq -r '.report_data.title' <<< "$INPUT_JSON")

echo "region,sales,target,variance" > "$OPAL_OUTPUT_DIR/report.csv"
jq -r '.report_data.items[] | "\(.region),\(.sales),\(.target),\(.sales - .target)"' \
 <<< "$INPUT_JSON" >> "$OPAL_OUTPUT_DIR/report.csv"

{
 echo "$TITLE"
 echo "========================"
 jq -r '.report_data.items[] |
   "\(.region): $\(.sales) / $\(.target) (variance: \(.sales - .target))"' \
   <<< "$INPUT_JSON"
 echo ""
 echo "Total sales: $(jq '[.report_data.items[].sales] | add' <<< "$INPUT_JSON")"
 echo "Total target: $(jq '[.report_data.items[].target] | add' <<< "$INPUT_JSON")"
} > "$OPAL_OUTPUT_DIR/summary.txt"

echo "Wrote report.csv and summary.txt to the output directory"

The script writes the following line to standard output and returns null:

Wrote report.csv and summary.txt to the output directory

The script produces report.csv with the following contents:

region,sales,target,variance
North,45000,50000,-5000
South,62000,55000,7000
East,38000,40000,-2000
West,51000,48000,3000

The script also produces summary.txt with the following contents:

Weekly Sales Summary
========================
North: $45000 / $50000 (variance: -5000)
South: $62000 / $55000 (variance: 7000)
East: $38000 / $40000 (variance: -2000)
West: $51000 / $48000 (variance: 3000)

Total sales: 196000
Total target: 193000

Both files display in the Output Files section of the execution details panel. Downstream steps and agents reference them by path.

Chain two code steps

Chain code steps so the output of one flows into the next. This splits complex logic into discrete stages that you test independently.

This example uses the following layout:

[Agent: Gather survey data] -> [Code step: Clean and normalize]
 -> [Code step: Analyze and score] -> [Agent: Generate report]

Clean and normalize the responses

Declare the following input variable:

<>

Enter the following sample data:

{
  "raw_responses": [
    { "name": "  Alice ", "rating": "9", "comment": "Great product!", "timestamp": "2026-08-01T10:00:00Z" },
    { "name": "Bob", "rating": "3", "comment": " Needs improvement. ", "timestamp": "2026-08-01T11:30:00Z" },
    { "rating": "7", "comment": "Decent experience", "timestamp": "2026-08-02T09:00:00Z" },
    { "name": "Dana", "rating": "15", "comment": "Love it!", "timestamp": "2026-08-02T14:00:00Z" }
  ]
}

Use the following Python handler:

def handler(input, ctx):
    responses = input["raw_responses"]

    cleaned = []
    for r in responses:
        cleaned.append({
            "respondent": r.get("name", "Anonymous").strip(),
            "rating": max(1, min(10, int(r.get("rating", 5)))),
            "comment": r.get("comment", "").strip(),
            "timestamp": r.get("timestamp", "")
        })

    return {
        "cleaned_responses": cleaned,
        "total_responses": len(cleaned),
        "summary": f"Cleaned {len(cleaned)} survey responses"
    }

The handler returns the following value:

{
  "cleaned_responses": [
    { "respondent": "Alice", "rating": 9, "comment": "Great product!", "timestamp": "2026-08-01T10:00:00Z" },
    { "respondent": "Bob", "rating": 3, "comment": "Needs improvement.", "timestamp": "2026-08-01T11:30:00Z" },
    { "respondent": "Anonymous", "rating": 7, "comment": "Decent experience", "timestamp": "2026-08-02T09:00:00Z" },
    { "respondent": "Dana", "rating": 10, "comment": "Love it!", "timestamp": "2026-08-02T14:00:00Z" }
  ],
  "total_responses": 4
}

The handler trims whitespace from names and comments. It defaults a missing name to Anonymous, and clamps the out-of-range rating of 15 down to 10.

Analyze and score the clean data

This step receives the output of the cleaning step. Declare its variable in Automatic mode so the orchestrator maps the upstream output onto it.

Declare the following input variable:

<>

Enter the following sample data:

{
  "cleaned_responses": [
    { "respondent": "Alice", "rating": 9, "comment": "Great product!", "timestamp": "2026-08-01T10:00:00Z" },
    { "respondent": "Bob", "rating": 3, "comment": "Needs improvement.", "timestamp": "2026-08-01T11:30:00Z" },
    { "respondent": "Anonymous", "rating": 7, "comment": "Decent experience", "timestamp": "2026-08-02T09:00:00Z" },
    { "respondent": "Dana", "rating": 10, "comment": "Love it!", "timestamp": "2026-08-02T14:00:00Z" }
  ]
}

Use the following Python handler:

def handler(input, ctx):
    responses = input["cleaned_responses"]

    ratings = [r["rating"] for r in responses]
    avg_rating = sum(ratings) / len(ratings) if ratings else 0
    promoters = sum(1 for r in ratings if r >= 9)
    detractors = sum(1 for r in ratings if r <= 6)
    nps = round(((promoters - detractors) / len(ratings)) * 100) if ratings else 0

    return {
        "average_rating": round(avg_rating, 2),
        "nps_score": nps,
        "promoters": promoters,
        "detractors": detractors,
        "total": len(ratings),
        "summary": f"NPS: {nps}, average rating {round(avg_rating, 2)}"
    }

The handler returns the following value:

{
  "average_rating": 7.25,
  "nps_score": 25,
  "promoters": 2,
  "detractors": 1,
  "total": 4
}

Each code step carries a single responsibility. The first normalizes messy input, and the second computes metrics on clean data. This separation lets you test each step with its own sample data. It also lets you change the scoring without touching the cleaning logic.

Process each item in a loop

Place a code step inside a Loop to process one item at a time. A loop runs the steps inside it once for each item in a collection. A loop cannot start with a code step, so an agent step comes first. See Workflow agent loops for more information. Declare an Automatic variable to receive the current item.

This example uses the following layout:

 [Loop: over products] -> [Agent: Look up competitor price]
        -> [Code step: Enrich product] -> [End loop]

Declare the following input variable:

Enter sample data that represents a single iteration:

{
  "product": {
    "name": "Widget Pro",
    "price": 49.99,
    "cost": 22.50,
    "units_sold": 1200,
    "category": "Hardware"
  }
}

Use the following Python handler:

def handler(input, ctx):
    product = input["product"]

    price = product["price"]
    cost = product["cost"]
    units_sold = product["units_sold"]

    margin = price - cost
    margin_pct = round((margin / price) * 100, 1) if price > 0 else 0

    return {
        "name": product["name"],
        "category": product["category"],
        "price": price,
        "margin_pct": margin_pct,
        "revenue": round(price * units_sold, 2),
        "profit": round(margin * units_sold, 2),
        "summary": f"{product['name']}: {margin_pct}% margin"
    }

The handler returns the following value for one iteration:

{
  "name": "Widget Pro",
  "category": "Hardware",
  "price": 49.99,
  "margin_pct": 55.0,
  "revenue": 59988.0,
  "profit": 32988.0
}

The loop collects the return value of each iteration. When every iteration finishes, downstream steps receive the collected results as an array.