AI Automation
Pro

n8n is the most powerful open-source automation tool available. Unlike Make.com, when you run n8n Self-Hosted — there are no quotas, no per-operation cost, and your data stays with you. The fastest way to install n8n in a production environment is Docker Compose.

Prerequisites

Configuring docker-compose.yml

# create a folder
mkdir n8n-data && cd n8n-data

# docker-compose.yml
cat > docker-compose.yml << 'EOF'
version: '3.1'
services:
  n8n:
    image: docker.n8n.io/n8nio/n8n
    restart: always
    ports:
      - "5678:5678"
    environment:
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=admin
      - N8N_BASIC_AUTH_PASSWORD=your-secure-password
      - WEBHOOK_URL=https://your-domain.com/
      - N8N_HOST=your-domain.com
      - N8N_PORT=5678
      - N8N_PROTOCOL=https
    volumes:
      - ~/.n8n:/home/node/.n8n
EOF

docker compose up -d

Explanation of the variables

n8n offers a Code Node that lets you write JavaScript (or Python) directly inside the Workflow. This makes anything that can be coded — possible. Data cleaning, complex calculations, building dynamic Payloads, and processing business logic that does not exist as a built-in Node.

// n8n Code Node — lead processing from CRM
const items = $input.all();
const processed = items.map(item => {
  const data = item.json;

  // clean an Israeli phone number
  const phone = data.phone?.replace(/\D/g, '');
  const formattedPhone = phone?.startsWith('972')
    ? `+${phone}`
    : `+972${phone?.replace(/^0/, '')}`;

  // compute Lead Score
  let score = 0;
  if (data.email?.endsWith('.co.il')) score += 20;
  if (data.company) score += 30;
  if (data.budget > 10000) score += 50;

  return {
    json: {
      ...data,
      phone: formattedPhone,
      leadScore: score,
      processedAt: new Date().toISOString()
    }
  };
});

return processed;

The code receives all the items ($input.all()), iterates over them with map, normalizes the phone, computes a Lead Score according to business criteria, and returns the enriched data to the next Nodes. Note that you must always return an array of objects with a field json.

n8n 1.x includes full support for LangChain directly in the interface. You can build AI Agents, Prompt chains, and conversations with Memory — all without a line of code.

When you run n8n in production with access to sensitive business data, security is not optional. Here is the list of essential checks:

So far you have used AI through prepared nodes in Make and n8n. The node does three things for you: it holds the key, builds the request, and parses the response. The moment you call the API directly, all three move to you — and that is exactly what gives you control over cost, format and error handling.

The key, and where it must never be

A leaked key is a bill you pay until you revoke it. The three places keys actually leak: code pushed to GitHub, client-side code in a browser, and a screenshot in a group chat. One rule covers all of them — the key lives in an environment variable, and the file holding it is in .gitignore.

pip install openai python-dotenv

# .env — never enters git
echo 'OPENAI_API_KEY=sk-proj-...' > .env
echo '.env' >> .gitignore

import os
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

resp = client.chat.completions.create(
    model="gpt-5-mini",
    messages=[
        {"role": "system", "content": "You are terse. Two lines maximum."},
        {"role": "user",   "content": "What does a webhook do?"},
    ],
)

print(resp.choices[0].message.content)
print(resp.usage)   # prompt_tokens, completion_tokens, total_tokens

Note the last line. resp.usage is your bill, and it comes back on every call. People who never look at it discover their consumption at the end of the month.

Which model

The common mistake is reaching for the strongest model for every task. In most automations the majority of calls are simple — classify, extract a field, rewrite a sentence — and a small model does them at the same quality for a fraction of the cost and half the latency.

Estimate cost before running it a thousand times

A token is roughly four characters of English — and fewer in most other languages, because the tokenizer is less efficient there. The same text in Hebrew can cost close to twice what it costs in English. Do not guess: measure one real example and multiply.

PRICE_IN  = 0.25 / 1_000_000    # $ per input token — check the price list
PRICE_OUT = 2.00 / 1_000_000    # $ per output token

u = resp.usage
per_call = u.prompt_tokens * PRICE_IN + u.completion_tokens * PRICE_OUT

for runs in (100, 1_000, 10_000):
    print(f"{runs:>6} runs: ${per_call * runs:.2f}")

This is the lesson that separates a demo from an automation that runs overnight. A model returning free text is a liability: today it answers {"status":"hot"}, tomorrow it prefixes "Sure! Here's the JSON:", and the day after it writes "Hot" with a capital H. The next node breaks, and nobody finds out until someone checks.

The fix is not asking more politely. It is enforcing a schema, at the API level.

from pydantic import BaseModel, Field
from typing import Literal

class Lead(BaseModel):
    company:  str
    contact:  str
    intent:   Literal["hot", "warm", "cold"]
    budget:   int | None = Field(None, description="Budget if stated, else null")
    next_step: str

resp = client.chat.completions.parse(
    model="gpt-5-mini",
    messages=[
        {"role": "system", "content":
         "Extract lead details from the enquiry. Never invent a field "
         "that is not present."},
        {"role": "user", "content": email_body},
    ],
    response_format=Lead,          # schema enforced server-side
)

lead = resp.choices[0].message.parsed   # a Lead object, not a string
print(lead.intent, lead.budget)

Two details decide whether this survives production. First, Literal instead of str on a classification field — the model now cannot return a fourth category. Second, | None on anything optional. Without it the model will invent a budget to satisfy the schema, which is worse than an empty field because it looks like data.

Function calling — when the model decides what to do

Structured outputs hand you data. Function calling lets the model choose which action to take and with what arguments. The distinction matters: in the first you know in advance what comes back, in the second you are delegating a decision.

tools = [{
    "type": "function",
    "function": {
        "name": "create_deal",
        "description": ("Open a new CRM deal. Use ONLY when both a company "
                        "name and a contact are present. Do not use for "
                        "general questions."),
        "parameters": {
            "type": "object",
            "properties": {
                "company": {"type": "string"},
                "amount":  {"type": "number"},
                "stage":   {"type": "string",
                            "enum": ["lead", "proposal", "negotiation"]},
            },
            "required": ["company", "stage"],
            "additionalProperties": False,
        },
    },
}]

resp = client.chat.completions.create(
    model="gpt-5-mini", messages=messages, tools=tools,
)

calls = resp.choices[0].message.tool_calls
if calls:
    args = json.loads(calls[0].function.arguments)
    create_deal(**args)          # your code executes. the model only chose.

Retry — and what not to retry

API errors split into two kinds and the handling is opposite. 429 and 5xx are transient — back off and try again. 400 and 401 are your bug — retrying only burns calls and bills you for them.

import time, random
from openai import RateLimitError, APIStatusError

def call_with_retry(fn, attempts=4):
    for i in range(attempts):
        try:
            return fn()
        except RateLimitError:
            pass                                  # transient — worth retrying
        except APIStatusError as e:
            if e.status_code < 500:
                raise                             # ours — do not retry
        wait = 2 ** i + random.random()           # jitter avoids a thundering herd
        print(f"attempt {i+1} failed, waiting {wait:.1f}s")
        time.sleep(wait)
    raise RuntimeError("failed after all attempts")

The random() in the backoff is not decoration. Without it, a hundred calls that failed together retry together at exactly the same instant, and fail together again.

This is the fastest payback in the module. Invoices, receipts, order screenshots, delivery notes — all of it arrives at a business as images, and somebody retypes it by hand. A vision model reads them and returns JSON.

Two ways to send an image

A URL if the image is already on the web and reachable without auth. Base64 if it is yours — a local file, a file from Drive, or an image that arrived on a webhook. The second is the common case in automation.

import base64
from pydantic import BaseModel

def to_data_url(path: str) -> str:
    mime = "image/png" if path.endswith(".png") else "image/jpeg"
    with open(path, "rb") as f:
        return f"data:{mime};base64," + base64.b64encode(f.read()).decode()

class Invoice(BaseModel):
    supplier:   str
    invoice_no: str
    date:       str          # YYYY-MM-DD
    total:      float
    tax:        float | None
    line_items: list[str]

resp = client.chat.completions.parse(
    model="gpt-5-mini",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text":
             "Extract the invoice fields. If a field is not clearly present, "
             "return null. Never estimate a number."},
            {"type": "image_url",
             "image_url": {"url": to_data_url("invoice.jpg"), "detail": "high"}},
        ],
    }],
    response_format=Invoice,
)

inv = resp.choices[0].message.parsed

detail — the parameter that changes cost several-fold

A flow that saves money: run low first just to classify what the image is, and only run high if it turns out to be an invoice. On a mailbox that also receives spam, that skips most of the expensive calls.

Keyword search finds words. Semantic search finds meaning. The query "how do I cancel an order" will match a document titled "Returns policy" without sharing a single word — and that is precisely what you need to build a chatbot over company documents.

The mechanism: an embedding converts text into a list of numbers that represents its meaning. Two texts with similar meaning get lists that sit close together. Search is measuring that closeness.

def embed(texts: list[str]) -> list[list[float]]:
    # batch it — one call for 100 chunks, not 100 calls
    r = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
    )
    return [d.embedding for d in r.data]

vecs = embed(["Returns policy", "Opening hours", "Product warranty"])
print(len(vecs[0]))     # vector dimensions

Pinecone — storing and querying

from pinecone import Pinecone

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index = pc.Index("company-docs")

# store the text in metadata, otherwise you have numbers with no source
index.upsert([
    {"id": f"doc-{i}",
     "values": v,
     "metadata": {"text": t, "source": "policy.pdf", "page": i}}
    for i, (t, v) in enumerate(zip(chunks, embed(chunks)))
])

q = embed(["how do I cancel an order?"])[0]
hits = index.query(vector=q, top_k=5, include_metadata=True)

for h in hits.matches:
    print(round(h.score, 3), h.metadata["source"], h.metadata["text"][:60])

Non-English content

Modern embedding models are multilingual and handle Hebrew reasonably, but two things are worth knowing. First, do not mix languages in one index without a reason — a Hebrew and an English document on the same subject will not necessarily land near each other. Second, test it for real: take five actual customer questions and run the search. If the top hit is not the right document, the problem is almost always your chunking rather than the model — which is exactly the next lesson.

RAG is four steps: split documents into chunks, turn them into vectors, retrieve the ones relevant to a question, and hand them to the model as context. You already have the first three from the previous lesson. This lesson connects them — and dwells on the step that decides whether any of it works.

Chunking — where most systems fail

Chunking is the highest-impact decision in the pipeline and gets the least attention. A chunk that is too large carries noise that dilutes relevance; too small and it loses context and returns a stranded sentence. The working rule: a chunk should answer one question completely.

def chunk(text: str, size=900, overlap=150) -> list[str]:
    """Split on paragraph boundaries, never mid-sentence."""
    paras = [p.strip() for p in text.split("\n\n") if p.strip()]
    out, cur = [], ""
    for p in paras:
        if len(cur) + len(p) < size:
            cur += ("\n\n" if cur else "") + p
        else:
            if cur:
                out.append(cur)
            # overlap: the tail of the previous chunk opens the next
            cur = (cur[-overlap:] + "\n\n" + p) if cur else p
    if cur:
        out.append(cur)
    return out

The overlap exists so that an answer sitting on the seam between two chunks is not cut in half. Without it, a question whose answer starts at the end of one chunk and finishes in the next is fully present in neither.

The whole pipeline

SYSTEM = """Answer only from the sources below.
If the answer is not in them, say "I don't have that in the documents".
Do not fill gaps from your general knowledge.
Cite the source in brackets after each claim."""

def ask(question: str, top_k: int = 5) -> str:
    q_vec = embed([question])[0]
    hits  = index.query(vector=q_vec, top_k=top_k, include_metadata=True)

    # relevance floor — better to not answer than to answer from noise
    good = [h for h in hits.matches if h.score > 0.35]
    if not good:
        return "I don't have that in the documents."

    context = "\n\n---\n\n".join(
        f"[{h.metadata['source']}]\n{h.metadata['text']}" for h in good
    )

    r = client.chat.completions.create(
        model="gpt-5-mini",
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user",
             "content": f"Sources:\n{context}\n\nQuestion: {question}"},
        ],
    )
    return r.choices[0].message.content

Three lines in that code are the entire difference between a system you can trust and one that invents. The grounding instruction, the relevance floor that prefers "I don't know" over answering from irrelevant chunks, and the citation requirement — which makes every answer checkable by the person reading it.

An automation that works on your machine and one that works in production are two different things. The difference is not the code — it is what happens when something changes: a rotated key, a service that goes down, load that grows, or someone editing the flow by accident.

Secrets do not belong in the flow

The worst thing you can do is paste an API key into an HTTP node. It is then stored in the flow backup, present in every export, visible in a screenshot, and travels to anyone who receives a copy. All three platforms have a proper place for it:

Staging is not a luxury

An automation that touches real customers is not tested on real customers. The minimum separation that works: a copy of the flow with a single environment variable that swaps the destinations — a test sheet instead of the real one, a private Slack channel instead of the team's, your inbox instead of the customer's.

import os

ENV = os.getenv("APP_ENV", "staging")   # safe default

TARGETS = {
    "staging":    {"sheet": "leads-test", "slack": "#bot-test",
                   "notify": "me@example.com"},
    "production": {"sheet": "leads",      "slack": "#sales",
                   "notify": "sales@example.com"},
}[ENV]

if ENV == "production" and not os.getenv("I_MEAN_IT"):
    raise SystemExit("production requires an explicit acknowledgement")

Note the default. If the variable is missing, the system runs against staging — not production. The other way round is a bug that surfaces late and expensively.

Pre-deploy checklist

This is the last lesson in the course and the most important one. Not because monitoring is interesting, but because the dangerous automation is the one that fails quietly. A flow that crashes and shouts gets fixed within the hour. A flow that stops running without saying anything keeps "working" in everyone's mind, and surfaces when a customer asks why nobody got back to them.

Three layers, in order of importance

Error Workflow in n8n

n8n lets you define one flow that runs whenever another flow fails. Build it once, then point every flow at it from settings.

const e = $input.first().json;

return [{ json: {
  text:
    `🔴 *${e.workflow.name}* failed\n` +
    `node: ${e.execution.lastNodeExecuted}\n` +
    `error: ${e.execution.error?.message ?? 'unknown'}\n` +
    `<${e.execution.url}|open the execution>`
}}];

Those three fields — which flow, which node, what error — are the difference between an alert you can act on and "something failed". The execution link saves the hunt.

Heartbeat — catching what does not shout

A scheduled flow that got disabled, a server that went down, a trigger that disconnected — none of these produce an error, because nothing ran. The fix inverts it: the flow reports that it is alive, and an external service alerts when the reports stop arriving.

# HTTP Request node, last in the flow, after success
GET https://uptime.example.com/api/push/<token>

# Uptime Kuma configured as a "Push" monitor with a 65-minute
# heartbeat interval for a flow that runs hourly.
# No report arrives → alert.

What to measure — and what not to

Once alerting is in place, four numbers are worth accumulating in a sheet or a dashboard: runs per day, failure rate, median duration, and API cost. Those four surface trends: a failure rate creeping upward, a duration that has doubled, or a cost that jumped after a prompt change.

What is not worth measuring: everything else. A dashboard with twenty charts does not get looked at, and what does not get looked at is not monitoring.