mcpField
N
n8n-patterns

n8n-patterns

github.com/mindmorass开发281 次安装37 周前更新
代码仓库

关于 n8n-patterns

n8n-patterns 是面向 n8n 工作流构建者的实践指南，涵盖触发器、数据处理、错误处理及部署等模式，帮助组织可复用的自动化流程。

主要功能

  1. 01Webhook、定时与应用触发模式
  2. 02数据转换及条件控制示例
  3. 03错误处理与失败重试模式
  4. 04子工作流与 REST API 调用
  5. 05凭据管理和自托管配置

适用场景

  1. 01设计事件驱动或定时工作流
  2. 02处理 API 分页和数据转换
  3. 03为流程添加错误处理与重试
  4. 04规划可复用的子工作流

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/mindmorass/reflex --skill n8n-patterns

安装后在 Claude Code 中输入/n8n-patterns确认已加载。

详细介绍

n8n 工作流模式

使用 n8n 构建可靠的工作流自动化——这是一款开源工作流自动化工具。

概述

n8n 是一个可自托管的工作流自动化平台，可连接应用和服务。主要功能：

  • 可视化工作流构建器，支持 400 多种集成
  • 支持自托管或云端部署
  • 使用代码节点编写自定义逻辑（JavaScript/Python）
  • 使用 Webhook 触发器实现实时自动化
  • 使用子工作流进行模块化设计

核心概念

工作流结构
┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│   Trigger   │───▶│    Node     │───▶│   Output    │
│  (Start)    │    │  (Process)  │    │  (Action)   │
└─────────────┘    └─────────────┘    └─────────────┘
节点类型
类型用途示例
触发器启动工作流Webhook、计划任务、应用触发器
操作执行操作HTTP 请求、数据库、电子邮件
转换修改数据Set、Code、IF、Switch
流程控制执行Merge、Split、Wait、Loop

触发器模式

Webhook 触发器
{
  "nodes": [
    {
      "name": "Webhook",
      "type": "n8n-nodes-base.webhook",
      "parameters": {
        "httpMethod": "POST",
        "path": "my-webhook",
        "responseMode": "responseNode",
        "options": {
          "rawBody": true
        }
      }
    }
  ]
}

最佳实践：

  • 使用 responseNode 设置自定义响应
  • 启用 rawBody 以验证签名
  • 添加身份验证（Header Auth、Basic Auth）
计划任务触发器
{
  "name": "Schedule Trigger",
  "type": "n8n-nodes-base.scheduleTrigger",
  "parameters": {
    "rule": {
      "interval": [
        {
          "field": "cronExpression",
          "expression": "0 9 * * 1-5"
        }
      ]
    }
  }
}

常见计划：

  • 0 * * * * - 每小时
  • 0 9 * * 1-5 - 工作日早上 9 点
  • 0 0 * * 0 - 每周日午夜
  • */15 * * * * - 每 15 分钟
应用触发器（轮询）
{
  "name": "GitHub Trigger",
  "type": "n8n-nodes-base.githubTrigger",
  "parameters": {
    "owner": "{{$env.GITHUB_OWNER}}",
    "repository": "{{$env.GITHUB_REPO}}",
    "events": ["issues", "pull_request"]
  }
}

数据转换模式

Set 节点（转换数据）
{
  "name": "Transform Data",
  "type": "n8n-nodes-base.set",
  "parameters": {
    "mode": "manual",
    "duplicateItem": false,
    "assignments": {
      "assignments": [
        {
          "name": "fullName",
          "value": "={{ $json.firstName }} {{ $json.lastName }}",
          "type": "string"
        },
        {
          "name": "timestamp",
          "value": "={{ DateTime.now().toISO() }}",
          "type": "string"
        }
      ]
    }
  }
}
Code 节点（JavaScript）
// Process items with custom logic
const results = [];

for (const item of $input.all()) {
  const data = item.json;

  // Transform data
  results.push({
    json: {
      id: data.id,
      processed: true,
      score: calculateScore(data),
      timestamp: new Date().toISOString()
    }
  });
}

function calculateScore(data) {
  return data.value * 0.8 + data.bonus * 0.2;
}

return results;
Code 节点（Python）
# Enable Python in n8n settings
import json
from datetime import datetime

results = []

for item in _input.all():
    data = item.json

    # Transform data
    results.append({
        "json": {
            "id": data.get("id"),
            "processed": True,
            "timestamp": datetime.now().isoformat()
        }
    })

return results

控制流模式

IF 节点（条件判断）
{
  "name": "Check Status",
  "type": "n8n-nodes-base.if",
  "parameters": {
    "conditions": {
      "options": {
        "caseSensitive": true,
        "leftValue": "",
        "typeValidation": "strict"
      },
      "conditions": [
        {
          "leftValue": "={{ $json.status }}",
          "rightValue": "active",
          "operator": {
            "type": "string",
            "operation": "equals"
          }
        }
      ],
      "combinator": "and"
    }
  }
}
Switch 节点（多分支）
{
  "name": "Route by Type",
  "type": "n8n-nodes-base.switch",
  "parameters": {
    "mode": "rules",
    "rules": {
      "values": [
        {
          "outputKey": "order",
          "conditions": {
            "conditions": [
              {
                "leftValue": "={{ $json.type }}",
                "rightValue": "order",
                "operator": { "type": "string", "operation": "equals" }
              }
            ]
          }
        },
        {
          "outputKey": "refund",
          "conditions": {
            "conditions": [
              {
                "leftValue": "={{ $json.type }}",
                "rightValue": "refund",
                "operator": { "type": "string", "operation": "equals" }
              }
            ]
          }
        }
      ]
    },
    "fallbackOutput": "extra"
  }
}
遍历项目
{
  "name": "Loop Over Items",
  "type": "n8n-nodes-base.splitInBatches",
  "parameters": {
    "batchSize": 10,
    "options": {
      "reset": false
    }
  }
}
Merge 节点（合并数据）
{
  "name": "Merge Results",
  "type": "n8n-nodes-base.merge",
  "parameters": {
    "mode": "combine",
    "mergeByFields": {
      "values": [
        {
          "field1": "id",
          "field2": "userId"
        }
      ]
    },
    "options": {}
  }
}

错误处理模式

使用 Error Trigger 进行 Try/Catch
{
  "nodes": [
    {
      "name": "Error Trigger",
      "type": "n8n-nodes-base.errorTrigger",
      "parameters": {}
    },
    {
      "name": "Send Alert",
      "type": "n8n-nodes-base.slack",
      "parameters": {
        "channel": "#alerts",
        "text": "Workflow failed: {{ $json.workflow.name }}\nError: {{ $json.execution.error.message }}"
      }
    }
  ]
}
失败时重试
{
  "name": "HTTP Request",
  "type": "n8n-nodes-base.httpRequest",
  "parameters": {
    "url": "https://api.example.com/data",
    "options": {}
  },
  "retryOnFail": true,
  "maxTries": 3,
  "waitBetweenTries": 1000
}
Stop and Error 节点
{
  "name": "Validation Failed",
  "type": "n8n-nodes-base.stopAndError",
  "parameters": {
    "errorMessage": "Invalid input: {{ $json.error }}"
  }
}

子工作流模式

Execute Workflow 节点
{
  "name": "Process Order",
  "type": "n8n-nodes-base.executeWorkflow",
  "parameters": {
    "source": "database",
    "workflowId": "order-processing-workflow-id",
    "mode": "each",
    "options": {
      "waitForSubWorkflow": true
    }
  }
}

最佳实践：

  • 使用子工作流封装可复用逻辑
  • 在工作流之间仅传递必要数据
  • 根据需要设置 waitForSubWorkflow
  • 使用工作流标签进行整理

HTTP 请求模式

REST API 调用
{
  "name": "API Request",
  "type": "n8n-nodes-base.httpRequest",
  "parameters": {
    "method": "POST",
    "url": "https://api.example.com/v1/resource",
    "authentication": "predefinedCredentialType",
    "nodeCredentialType": "httpHeaderAuth",
    "sendHeaders": true,
    "headerParameters": {
      "parameters": [
        {
          "name": "Content-Type",
          "value": "application/json"
        }
      ]
    },
    "sendBody": true,
    "bodyParameters": {
      "parameters": [
        {
          "name": "data",
          "value": "={{ JSON.stringify($json) }}"
        }
      ]
    },
    "options": {
      "timeout": 30000,
      "response": {
        "response": {
          "fullResponse": false,
          "responseFormat": "json"
        }
      }
    }
  }
}
分页模式
// Code node for API pagination
const allResults = [];
let page = 1;
let hasMore = true;

while (hasMore) {
  const response = await this.helpers.httpRequest({
    method: 'GET',
    url: `https://api.example.com/items?page=${page}&limit=100`,
    headers: {
      'Authorization': `Bearer ${$env.API_TOKEN}`
    }
  });

  allResults.push(...response.data);
  hasMore = response.hasNextPage;
  page++;

  // Rate limiting
  await new Promise(r => setTimeout(r, 100));
}

return allResults.map(item => ({ json: item }));

凭据管理

环境变量
// Access in expressions
{{ $env.API_KEY }}
{{ $env.DATABASE_URL }}

// Access in Code node
const apiKey = $env.API_KEY;
凭据类型
类型用途
httpBasicAuth基本身份验证
httpHeaderAuth在标头中传递 API 密钥
oAuth2ApiOAuth 2.0 流程
httpQueryAuth在查询字符串中传递 API 密钥

自托管模式

Docker Compose
version: '3.8'

services:
  n8n:
    image: n8nio/n8n:latest
    restart: unless-stopped
    ports:
      - "5678:5678"
    environment:
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=${N8N_USER}
      - N8N_BASIC_AUTH_PASSWORD=${N8N_PASSWORD}
      - N8N_HOST=${N8N_HOST}
      - N8N_PORT=5678
      - N8N_PROTOCOL=https
      - NODE_ENV=production
      - WEBHOOK_URL=https://${N8N_HOST}/
      - GENERIC_TIMEZONE=UTC
      - N8N_ENCRYPTION_KEY=${N8N_ENCRYPTION_KEY}
      - DB_TYPE=postgresdb
      - DB_POSTGRESDB_HOST=postgres
      - DB_POSTGRESDB_PORT=5432
      - DB_POSTGRESDB_DATABASE=n8n
      - DB_POSTGRESDB_USER=${POSTGRES_USER}
      - DB_POSTGRESDB_PASSWORD=${POSTGRES_PASSWORD}
      - EXECUTIONS_DATA_PRUNE=true
      - EXECUTIONS_DATA_MAX_AGE=168
    volumes:
      - n8n_data:/home/node/.n8n
    depends_on:
      - postgres

  postgres:
    image: postgres:15
    restart: unless-stopped
    environment:
      - POSTGRES_USER=${POSTGRES_USER}
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
      - POSTGRES_DB=n8n
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER}"]
      interval: 10s
      timeout: 5s
      retries: 5

volumes:
  n8n_data:
  postgres_data:
环境变量（.env）
# n8n Configuration
N8N_HOST=n8n.example.com
N8N_USER=admin
N8N_PASSWORD=secure-password-here
N8N_ENCRYPTION_KEY=$(openssl rand -hex 32)

# Database
POSTGRES_USER=n8n
POSTGRES_PASSWORD=secure-db-password

# Optional: Queue mode for scaling
EXECUTIONS_MODE=queue
QUEUE_BULL_REDIS_HOST=redis
队列模式（扩展）
# docker-compose.queue.yml
services:
  n8n:
    environment:
      - EXECUTIONS_MODE=queue
      - QUEUE_BULL_REDIS_HOST=redis
      - QUEUE_HEALTH_CHECK_ACTIVE=true

  n8n-worker:
    image: n8nio/n8n:latest
    command: worker
    environment:
      - EXECUTIONS_MODE=queue
      - QUEUE_BULL_REDIS_HOST=redis
    deploy:
      replicas: 3

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

常见工作流模板

Webhook 到数据库
{
  "name": "Webhook to Database",
  "nodes": [
    {
      "name": "Webhook",
      "type": "n8n-nodes-base.webhook",
      "parameters": {
        "httpMethod": "POST",
        "path": "ingest",
        "responseMode": "responseNode"
      }
    },
    {
      "name": "Validate",
      "type": "n8n-nodes-base.if",
      "parameters": {
        "conditions": {
          "conditions": [
            {
              "leftValue": "={{ $json.id }}",
              "rightValue": "",
              "operator": { "type": "string", "operation": "notEmpty" }
            }
          ]
        }
      }
    },
    {
      "name": "Insert",
      "type": "n8n-nodes-base.postgres",
      "parameters": {
        "operation": "insert",
        "table": "events",
        "columns": "id,type,data,created_at"
      }
    },
    {
      "name": "Success Response",
      "type": "n8n-nodes-base.respondToWebhook",
      "parameters": {
        "respondWith": "json",
        "responseBody": "={{ { \"success\": true, \"id\": $json.id } }}"
      }
    }
  ]
}
定时同步
{
  "name": "Daily Data Sync",
  "nodes": [
    {
      "name": "Schedule",
      "type": "n8n-nodes-base.scheduleTrigger",
      "parameters": {
        "rule": {
          "interval": [{ "field": "cronExpression", "expression": "0 2 * * *" }]
        }
      }
    },
    {
      "name": "Fetch Source",
      "type": "n8n-nodes-base.httpRequest",
      "parameters": {
        "url": "https://api.source.com/data",
        "authentication": "predefinedCredentialType"
      }
    },
    {
      "name": "Transform",
      "type": "n8n-nodes-base.code",
      "parameters": {
        "jsCode": "return $input.all().map(item => ({ json: { ...item.json, synced_at: new Date().toISOString() } }));"
      }
    },
    {
      "name": "Upsert Destination",
      "type": "n8n-nodes-base.postgres",
      "parameters": {
        "operation": "upsert",
        "table": "synced_data"
      }
    }
  ]
}
事件驱动通知
{
  "name": "Alert Pipeline",
  "nodes": [
    {
      "name": "Webhook",
      "type": "n8n-nodes-base.webhook",
      "parameters": { "path": "alert" }
    },
    {
      "name": "Route by Severity",
      "type": "n8n-nodes-base.switch",
      "parameters": {
        "rules": {
          "values": [
            { "outputKey": "critical", "conditions": { "conditions": [{ "leftValue": "={{ $json.severity }}", "rightValue": "critical" }] } },
            { "outputKey": "warning", "conditions": { "conditions": [{ "leftValue": "={{ $json.severity }}", "rightValue": "warning" }] } }
          ]
        }
      }
    },
    {
      "name": "Page On-Call",
      "type": "n8n-nodes-base.pagerDuty"
    },
    {
      "name": "Slack Alert",
      "type": "n8n-nodes-base.slack"
    }
  ]
}

表达式速查表

表达式描述
{{ $json.field }}访问当前项目中的字段
{{ $json["field-name"] }}访问包含特殊字符的字段
{{ $('NodeName').item.json.field }}访问特定节点的数据
{{ $input.first().json }}第一个输入项目
{{ $input.all() }}所有输入项目
{{ $env.VAR_NAME }}环境变量
{{ $now }}当前日期和时间
{{ $today }}当前日期
{{ $runIndex }}当前执行运行索引
{{ $itemIndex }}当前项目索引
{{ $workflow.id }}工作流 ID
{{ $execution.id }}执行 ID

Luxon DateTime 示例

// n8n uses Luxon for dates
{{ $now.toISO() }}                    // ISO format
{{ $now.toFormat('yyyy-MM-dd') }}     // Custom format
{{ $now.plus({ days: 7 }).toISO() }}  // Add 7 days
{{ $now.startOf('month').toISO() }}   // Start of month
{{ DateTime.fromISO($json.date) }}    // Parse ISO string

最佳实践

  1. 命名：使用描述性节点名称（动词 + 名词）
  2. 错误处理：始终添加错误工作流
  3. 凭据：切勿将密钥硬编码
  4. 批处理：对大型数据集使用 splitInBatches
  5. 超时：为 HTTP 节点设置适当的超时时间
  6. 日志记录：在 Code 节点中使用 console.log 进行调试
  7. 测试：启用工作流前先手动执行
  8. 版本控制：将工作流导出为 JSON 并提交到 git
  9. 文档：为复杂逻辑添加便笺
  10. 模块化设计：使用子工作流实现逻辑复用

调试提示

// In Code node - log to n8n console
console.log('Debug:', JSON.stringify($json, null, 2));

// Return debug info
return [{
  json: {
    debug: true,
    input: $json,
    env: $env.NODE_ENV,
    timestamp: new Date().toISOString()
  }
}];

Resources

评论

支持 Markdown