> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cogniagent.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Execute Code Scenarios
> Step-by-step patterns for the Execute Code node — clean data, act on every row, sync what's new, build reports, score and route.
These are the jobs people most often give an Execute Code node, each with the workflow around it, the code, and what to watch for. Copy one, change the keys and names to yours, and test it with a few records first.
New to the node? Start with [Execute Code](/nodes/actions/execute-code) for how inputs, outputs and the editor work, and [Execute Code integrations](/nodes/actions/execute-code-integrations) for calling apps.
## Which node for the job?
Code is one of several ways to do work in a workflow. Pick the simplest one that fits:
| You want to | Use |
| ---------------------------------------------------------------------- | ------------------------------------------------------------------------------ |
| Run one app action with values from earlier steps | [Integration Action](/nodes/actions/integration-action) |
| Let AI read something, decide, and act | [LLM](/nodes/actions/llm) with [integrations](/nodes/actions/llm-integrations) |
| Apply the same fixed rules to many records, or join data from two apps | **Execute Code** |
| Run several nodes for each item, such as an LLM or a Human Step | [Loop](/nodes/logic/loop) |
| Work out one small value from earlier outputs | [Resolve Value](/nodes/actions/resolve-value) |
| Keep a value for later steps or later runs | [Update Variable](/nodes/actions/update-variable), or `set_variable` in code |
A rule of thumb: when every item gets the same treatment, loop in code. It runs the calls back to back in one step, with no model turn per item. When each item needs judgment, or a person, use a Loop node around the nodes that provide it.
## Clean up incoming data
**The job:** a web form posts leads with messy names, emails in capitals and phone numbers full of spaces. Clean them before they reach your CRM.
```text theme={null}
Webhook
└── Execute Code (clean_lead)
└── Condition ({{clean_lead.result.valid}} is true)
└── Integration Action (HubSpot: Create contact)
```
```python theme={null}
body = webhook_1["body"]
email = (body.get("email") or "").strip().lower()
first, _, last = (body.get("name") or "").strip().partition(" ")
phone = "".join(ch for ch in (body.get("phone") or "") if ch.isdigit() or ch == "+")
output_data({
"email": email,
"firstname": first.title(),
"lastname": last.title(),
"phone": phone,
"valid": "@" in email,
})
```
In the Integration Action, map each field from this step, for example **Email** to `{{clean_lead.result.email}}`.
**Watch for:**
* Use `body.get("field")` rather than `body["field"]`. A form that leaves a field out would otherwise stop the step with a `KeyError`.
* Return a `valid` flag and branch on it with a [Condition](/nodes/logic/condition), rather than failing the step on bad input.
## Act on every row of a sheet
**The job:** every morning, post each qualified lead in a Google Sheet to Slack, and report any that failed.
```text theme={null}
Scheduled Trigger (every day at 9:00)
└── Execute Code (notify_sales), with Google Sheets and Slack bound
```
Bind **Google Sheets** with the **Batch get** action and **Slack** with **Send message**, then:
```python theme={null}
result = integrations.googlesheets.batch_get(spreadsheet_id=sheet_id, ranges=["Leads!A2:C"])
rows = result["valueRanges"][0].get("values", [])
sent, failed = 0, []
for number, row in enumerate(rows, start=2):
name, email, status = (row + ["", "", ""])[:3]
if status != "qualified":
continue
try:
integrations.slack.send_message(
channel="#sales",
markdown_text=f"New lead: *{name}* ({email})",
)
sent += 1
except IntegrationError as error:
failed.append({"row": number, "error": error.message})
output_data({"sent": sent, "failed": failed})
```
`sheet_id` is a workflow variable holding the spreadsheet's ID.
**Watch for:**
* Google Sheets leaves out empty cells at the end of a row, so a row can be shorter than you expect. `(row + ["", "", ""])[:3]` pads it.
* Catching `IntegrationError` inside the loop means one bad row doesn't stop the rest. `failed` tells you which rows to look at.
* Every message counts toward the limit of **200 app calls per run**. For bigger sheets, handle a batch per run, as in the next scenario.
* Calls already made are not undone. If the step runs twice, the messages go out twice, so remember what you've done.
## Handle only what's new since the last run
**The job:** a sheet keeps growing. Each hour, handle only the rows added since the last run, at most 150 at a time.
```text theme={null}
Scheduled Trigger (every hour)
└── Execute Code (sync_orders), with Google Sheets and Slack bound
```
Add a workflow variable `rows_done` of type **Number**, starting at `0`. It remembers how many rows are already handled:
```python theme={null}
BATCH = 150 # well under 200 app calls per run
result = integrations.googlesheets.batch_get(spreadsheet_id=sheet_id, ranges=["Orders!A2:C"])
rows = result["valueRanges"][0].get("values", [])
start = int(rows_done)
batch = rows[start:start + BATCH]
for row in batch:
order_id, customer, total = (row + ["", "", ""])[:3]
integrations.slack.send_message(
channel="#orders",
markdown_text=f"Order {order_id} from {customer}: {total}",
)
set_variable("rows_done", start + len(batch))
output_data({"handled": len(batch), "waiting": len(rows) - start - len(batch)})
```
**Watch for:**
* `set_variable` runs once, at the end. If the step fails halfway, `rows_done` keeps its old value and the next run sends that batch again. When a repeat would hurt, call `set_variable` after each row instead. That's up to 100 calls per run.
* This counts rows, so it assumes rows are only ever added at the bottom. If people insert or delete rows, remember an ID or a date instead: store the newest one you've handled, and skip anything older.
## Build a report and send it as a file
**The job:** every Monday, total last week's orders by region into an Excel file and email it.
```text theme={null}
Scheduled Trigger (Mondays at 8:00)
└── Execute Code (weekly_report), with Google Sheets bound
└── Integration Action (Gmail: send email, attachment {{weekly_report.files[0]}})
```
```python theme={null}
import pandas as pd
result = integrations.googlesheets.batch_get(spreadsheet_id=sheet_id, ranges=["Orders!A1:C"])
header, *rows = result["valueRanges"][0].get("values", [[]])
rows = [row + [""] * (len(header) - len(row)) for row in rows]
orders = pd.DataFrame(rows, columns=header)
orders["total"] = pd.to_numeric(orders["total"], errors="coerce").fillna(0)
by_region = orders.groupby("region", as_index=False)["total"].sum()
by_region.sort_values("total", ascending=False).to_excel("weekly-sales.xlsx", index=False)
output_data({"orders": len(orders), "revenue": float(orders["total"].sum())})
```
In the email step, insert `{{weekly_report.files[0]}}` into the attachment field. Use `{{weekly_report.result.revenue}}` in the message text.
**Watch for:**
* Save the file under a plain name, like `weekly-sales.xlsx`. Files in subfolders are not attached.
* `float(...)` turns pandas' number into a plain one, so `output_data` can store it.
* Charts work the same way: `matplotlib` and `plotly` are installed, and `plt.savefig("chart.png")` produces a file to attach.
## Score a lead and route it
**The job:** score each new lead by fixed rules, then send hot leads to sales and the rest to a nurture list.
```text theme={null}
Webhook
└── Execute Code (score_lead)
└── Condition ({{score_lead.result.tier}} equals "hot")
├── yes: Integration Action (Slack: send message to #sales)
└── no: Integration Action (HubSpot: add to nurture list)
```
```python theme={null}
lead = webhook_1["body"]
score = 0
if int(lead.get("company_size") or 0) >= 50:
score += 40
if lead.get("country") in {"US", "CA", "GB"}:
score += 20
if "demo" in (lead.get("message") or "").lower():
score += 40
tier = "hot" if score >= 70 else "warm" if score >= 40 else "cold"
output_data({"score": score, "tier": tier})
```
**Watch for:**
* Form values often arrive as text. `int(...)` makes `"120"` a number before you compare it.
* Code gives the same answer for the same lead every time, and costs no model call. When the decision needs reading a free-text message, use an [LLM](/nodes/actions/llm) instead, or let code do the arithmetic and an LLM the reading.
## Pull facts out of a document
**The job:** invoices arrive as PDFs. Pull out the invoice number, date and total.
```text theme={null}
App Trigger (new email with an attachment)
└── Read File (parse_file_1)
└── Execute Code (invoice_facts)
```
```python theme={null}
import re
text = parse_file_1["content"]
def find(pattern):
match = re.search(pattern, text, re.IGNORECASE)
return match.group(1).strip() if match else None
output_data({
"invoice_number": find(r"invoice\s*(?:no\.?|number|#)\s*[:\-]?\s*([A-Z0-9\-]+)"),
"date": find(r"date\s*[:\-]?\s*([0-9]{1,2}[./-][0-9]{1,2}[./-][0-9]{2,4})"),
"total": find(r"total\s*(?:due)?\s*[:\-]?\s*([$€£]?\s?[0-9.,]+)"),
})
```
**Watch for:**
* Let [Read File](/nodes/actions/read-file) turn the PDF into text. The code works on its `content`.
* Patterns fit documents from the same sender. When layouts vary a lot, an [LLM](/nodes/actions/llm) reads them more reliably.
* Return `None` for a field you couldn't find, and check it in a Condition, rather than guessing.
## Call a web service that isn't in the app list
**The job:** look up public data, such as an exchange rate, from a service without an app in the catalog.
```python theme={null}
import requests
response = requests.get(
"https://api.example.com/rates",
params={"base": "EUR", "symbols": "USD"},
timeout=30,
)
response.raise_for_status()
output_data({"eur_usd": response.json()["rates"]["USD"]})
```
**Watch for:**
* Always pass a `timeout`. Without one, a slow service can hold the step until the 30-minute limit.
* `raise_for_status()` stops the step on an error answer instead of returning bad data.
* Everyone who can edit the workflow can read the code. For a service that needs a key or a login, connect it as an app, or use the [HTTP Request](/nodes/actions/http-request) node.
## Habits that keep code steps reliable
Wrap the work for each record in `try` / `except`, and collect what failed in the output. One bad record then costs one record, not the whole run.
A step can run twice: a retry, a re-run, an overlapping schedule. Before creating something, check whether it exists, or remember what you've done in a workflow variable. Calls already made are never undone.
One run has 30 minutes, 200 app calls and 100 `set_variable` calls. For big jobs, handle a batch per run and remember where you stopped.
Anything written to the error stream fails the step, warnings included. Silence the warnings you expect with `warnings.filterwarnings("ignore")`.
Not sure what an earlier node or an app call returns? `print()` it once and read `stdOut` after a test run. Then read the fields with `.get()` so a missing one doesn't stop the step.
A node that cleans, scores and sends is hard to fix. Split it into steps with clear names; each one's output shows in the run history.
## Related
Inputs, outputs, files, variables, the editor and limits.
Bind apps and call them from Python.
Run several nodes for every item.
Branch on a value your code returned.