n8n

# n8n Workflow Management

针对 n8n 平台的全面工作流自动化管理，具备创建、测试、执行监控和性能优化功能。

## ⚠️ 关键：工作流创建规则

**在创建 n8n 工作流时，请务必：**

1. ✅ **生成完整的工作流**，包含所有功能性节点 2. ✅ **包含实际的 HTTP Request 节点**用于 API 调用（ImageFX, Gemini, Veo, Suno 等） 3. ✅ **添加 Code 节点**用于数据转换和逻辑处理 4. ✅ **在所有节点之间建立正确的连接** 5. ✅ **使用真实的节点类型**（n8n-nodes-base.httpRequest, n8n-nodes-base.code, n8n-nodes-base.set）

**切勿：** - ❌ 创建“设置说明”占位符节点 - ❌ 生成仅包含 TODO 注释的工作流 - ❌ 创建需要手动添加节点的不完整工作流 - ❌ 使用纯文本节点代替真实功能

**优秀工作流示例：** ``` Manual Trigger → Set Config → HTTP Request (API call) → Code (parse) → Response ```

**糟糕工作流示例：** ``` Manual Trigger → Code ("Add HTTP nodes here, configure APIs...") ```

请始终构建完整、功能正常的工作流，并配置和连接所有必要的节点。

## 设置

**必需的环境变量：** - `N8N_API_KEY` — 您的 n8n API 密钥（在 n8n 界面中的 Settings → API 获取） - `N8N_BASE_URL` — 您的 n8n 实例 URL

**通过 OpenClaw 设置配置凭据：**

添加到 `~/.config/openclaw/settings.json`： ```json { "skills": { "n8n": { "env": { "N8N_API_KEY": "your-api-key-here", "N8N_BASE_URL": "your-n8n-url-here" } } } } ```

或者设置单会话环境变量（**请勿**在 shell rc 文件中持久化存储密钥）： ```bash export N8N_API_KEY="your-api-key-here" export N8N_BASE_URL="your-n8n-url-here" ```

**验证连接：** ```bash python3 scripts/n8n_api.py list-workflows --pretty ```

> **安全提示：** 切勿以明文形式将 API 密钥存储在 shell 配置文件（`~/.bashrc`, `~/.zshrc`）中。请使用 OpenClaw 设置文件或安全的密钥管理器。

## 快速参考

### 工作流管理

#### 列出工作流 ```bash python3 scripts/n8n_api.py list-workflows --pretty python3 scripts/n8n_api.py list-workflows --active true --pretty ```

#### 获取工作流详情 ```bash python3 scripts/n8n_api.py get-workflow --id <workflow-id> --pretty ```

#### 创建工作流 ```bash # From JSON file python3 scripts/n8n_api.py create --from-file workflow.json ```

#### 激活/停用 ```bash python3 scripts/n8n_api.py activate --id <workflow-id> python3 scripts/n8n_api.py deactivate --id <workflow-id> ```

### 测试与验证

#### 验证工作流结构 ```bash # Validate existing workflow python3 scripts/n8n_tester.py validate --id <workflow-id>

# Validate from file python3 scripts/n8n_tester.py validate --file workflow.json --pretty

# Generate validation report python3 scripts/n8n_tester.py report --id <workflow-id> ```

#### 试运行测试 ```bash # Test with data python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data '{"email": "[email protected]"}'

# Test with data file python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data-file test-data.json

# Full test report (validation + dry run) python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data-file test.json --report ```

#### 测试套件 ```bash # Run multiple test cases python3 scripts/n8n_tester.py test-suite --id <workflow-id> --test-suite test-cases.json ```

### 执行监控

#### 列出执行记录 ```bash # Recent executions (all workflows) python3 scripts/n8n_api.py list-executions --limit 10 --pretty

# Specific workflow executions python3 scripts/n8n_api.py list-executions --id <workflow-id> --limit 20 --pretty ```

#### 获取执行详情 ```bash python3 scripts/n8n_api.py get-execution --id <execution-id> --pretty ```

#### 手动执行 ```bash # Trigger workflow python3 scripts/n8n_api.py execute --id <workflow-id>

# Execute with data python3 scripts/n8n_api.py execute --id <workflow-id> --data '{"key": "value"}' ```

### 性能优化

#### 分析性能 ```bash # Full performance analysis python3 scripts/n8n_optimizer.py analyze --id <workflow-id> --pretty

# Analyze specific period python3 scripts/n8n_optimizer.py analyze --id <workflow-id> --days 30 --pretty ```

#### 获取优化建议 ```bash # Priority-ranked suggestions python3 scripts/n8n_optimizer.py suggest --id <workflow-id> --pretty ```

#### 生成优化报告 ```bash # Human-readable report with metrics, bottlenecks, and suggestions python3 scripts/n8n_optimizer.py report --id <workflow-id> ```

#### 获取工作流统计 ```bash # Execution statistics python3 scripts/n8n_api.py stats --id <workflow-id> --days 7 --pretty ```

## Python API

### 基本用法

```python from scripts.n8n_api import N8nClient

client = N8nClient()

# List workflows workflows = client.list_workflows(active=True)

# Get workflow workflow = client.get_workflow('workflow-id')

# Create workflow new_workflow = client.create_workflow({ 'name': 'My Workflow', 'nodes': [...], 'connections': {...} })

# Activate/deactivate client.activate_workflow('workflow-id') client.deactivate_workflow('workflow-id')

# Executions executions = client.list_executions(workflow_id='workflow-id', limit=10) execution = client.get_execution('execution-id')

# Execute workflow result = client.execute_workflow('workflow-id', data={'key': 'value'}) ```

### 验证与测试

```python from scripts.n8n_api import N8nClient from scripts.n8n_tester import WorkflowTester

client = N8nClient() tester = WorkflowTester(client)

# Validate workflow validation = tester.validate_workflow(workflow_id='123') print(f"Valid: {validation['valid']}") print(f"Errors: {validation['errors']}") print(f"Warnings: {validation['warnings']}")

# Dry run result = tester.dry_run( workflow_id='123', test_data={'email': '[email protected]'} ) print(f"Status: {result['status']}")

# Test suite test_cases = [ {'name': 'Test 1', 'input': {...}, 'expected': {...}}, {'name': 'Test 2', 'input': {...}, 'expected': {...}} ] results = tester.test_suite('123', test_cases) print(f"Passed: {results['passed']}/{results['total_tests']}")

# Generate report report = tester.generate_test_report(validation, result) print(report) ```

### 性能优化

```python from scripts.n8n_optimizer import WorkflowOptimizer

optimizer = WorkflowOptimizer()

# Analyze performance analysis = optimizer.analyze_performance('workflow-id', days=7) print(f"Performance Score: {analysis['performance_score']}/100") print(f"Health: {analysis['execution_metrics']['health']}")

# Get suggestions suggestions = optimizer.suggest_optimizations('workflow-id') print(f"Priority Actions: {len(suggestions['priority_actions'])}") print(f"Quick Wins: {len(suggestions['quick_wins'])}")

# Generate report report = optimizer.generate_optimization_report(analysis) print(report) ```

## 常见工作流

### 1. 验证并测试工作流

```bash # Validate workflow structure python3 scripts/n8n_tester.py validate --id <workflow-id> --pretty

# Test with sample data python3 scripts/n8n_tester.py dry-run --id <workflow-id> \ --data '{"email": "[email protected]", "name": "Test User"}'

# If tests pass, activate python3 scripts/n8n_api.py activate --id <workflow-id> ```

### 2. 调试失败的工作流

```bash # Check recent executions python3 scripts/n8n_api.py list-executions --id <workflow-id> --limit 10 --pretty

# Get specific execution details python3 scripts/n8n_api.py get-execution --id <execution-id> --pretty

# Validate workflow structure python3 scripts/n8n_tester.py validate --id <workflow-id>

# Generate test report python3 scripts/n8n_tester.py report --id <workflow-id>

# Check for optimization issues python3 scripts/n8n_optimizer.py report --id <workflow-id> ```

### 3. 优化工作流性能

```bash # Analyze current performance python3 scripts/n8n_optimizer.py analyze --id <workflow-id> --days 30 --pretty

# Get actionable suggestions python3 scripts/n8n_optimizer.py suggest --id <workflow-id> --pretty

# Generate comprehensive report python3 scripts/n8n_optimizer.py report --id <workflow-id>

# Review execution statistics python3 scripts/n8n_api.py stats --id <workflow-id> --days 30 --pretty

# Test optimizations with dry run python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data-file test-data.json ```

### 4. 监控工作流健康状况

```bash # Check active workflows python3 scripts/n8n_api.py list-workflows --active true --pretty

# Review recent execution status python3 scripts/n8n_api.py list-executions --limit 20 --pretty

# Get statistics for each critical workflow python3 scripts/n8n_api.py stats --id <workflow-id> --pretty

# Generate health reports python3 scripts/n8n_optimizer.py report --id <workflow-id> ```

## 验证检查

测试模块执行全面的验证：

### 结构验证 - ✓ 必填字段存在（节点、连接） - ✓ 所有节点均具有名称和类型 - ✓ 连接目标存在 - ✓ 无断开连接的节点（警告）

### 配置验证 - ✓ 需要凭据的节点已配置 - ✓ 必需参数已设置 - ✓ HTTP 节点具有 URL - ✓ Webhook 节点具有路径 - ✓ Email 节点具有内容

### 流程验证 - ✓ 工作流具有触发器节点 - ✓ 执行流程正确 - ✓ 无循环依赖 - ✓ 已识别结束节点

## 优化分析

优化器分析多个维度：

### 执行指标 - 总执行次数 - 成功/失败率 - 健康状况（优秀/良好/一般/较差） - 错误模式

### 性能指标 - 节点数量和复杂度 - 连接模式 - 昂贵的操作（API 调用、数据库查询） - 并行执行机会

### 瓶颈检测 - 顺序执行的昂贵操作 - 高失败率 - 缺少错误处理 - 速率限制问题

### 优化机会 - **并行执行：** 识别可以并发运行的节点 - **缓存：** 建议对重复的 API 调用进行缓存 - **批处理：** 推荐对大数据集进行批处理 - **错误处理：** 添加错误恢复机制 - **降低复杂度：** 拆分复杂的工作流 - **超时设置：** 配置执行限制

## 性能评分

工作流根据以下标准获得性能评分（0-100）：

- **成功率：** 越高越好（50% 权重） - **复杂度：** 越低越好（30% 权重） - **瓶颈：** 越少越好（严重：-20，高：-10，中等：-5） - **优化措施：** 实施的最佳实践（每项 +5）

评分解读： - **90-100：** 优秀 - 已充分优化 - **70-89：** 良好 - 可进行细微改进 - **50-69：** 一般 - 建议优化 - **0-49：** 较差 - 存在重大问题

## 最佳实践

### 开发 1. **规划结构：** 在构建前设计工作流节点和连接 2. **优先验证：** 部署前始终进行验证 3. **充分测试：** 使用多个测试用例进行试运行 4. **错误处理：** 添加错误节点以提高可靠性 5. **文档记录：** 在 Code 节点中注释复杂逻辑

### 测试 1. **示例数据：** 创建真实的测试数据文件 2. **边界情况：** 测试边界条件和错误 3. **增量测试：** 测试每个新增的节点 4. **回归测试：** 变更后重新测试 5. **模拟生产：** 使用镜像生产环境的环境

### 部署 1. **先停用：** 以停用状态部署工作流 2. **渐进式发布：** 最初限制流量进行测试 3. **密切监控：** 仔细观察前几次执行 4. **快速回滚：** 如果出现问题准备好停用 5. **记录变更：** 保留修改日志

### 优化 1. **基准指标：** 在更改前捕获性能数据 2. **逐一更改：** 隔离优化影响 3. **衡量结果：** 对比更改前后的指标 4. **定期审查：** 安排每月优化审查 5. **成本意识：** 监控 API 使用情况和执行成本

### 维护 1. **健康检查：** 每周审查执行统计 2. **错误分析：** 调查失败模式 3. **性能监控：** 跟踪执行时间 4. **凭据轮换：** 定期更新凭据 5. **清理：** 归档或删除未使用的工作流

## 故障排除

### 身份验证错误 ``` Error: N8N_API_KEY not found in environment ``` **解决方案：** 设置环境变量： ```bash export N8N_API_KEY="your-api-key" ```

### 连接错误 ``` Error: HTTP 401: Unauthorized ``` **解决方案：** 1. 验证 API 密钥是否正确 2. 检查 N8N_BASE_URL 设置是否正确 3. 确认 n8n 中已启用 API 访问

### 验证错误 ``` Validation failed: Node missing 'name' field ``` **解决方案：** 检查工作流 JSON 结构，确保所有必填字段存在

### 执行超时 ``` Status: timeout - Execution did not complete ``` **解决方案：** 1. 检查工作流是否存在无限循环 2. 减少测试数据集大小 3. 优化昂贵的操作 4. 在工作流设置中配置执行超时

### 速率限制 ``` Error: HTTP 429: Too Many Requests ``` **解决方案：** 1. 在 API 调用之间添加 Wait 节点 2. 实现指数退避 3. 使用批处理 4. 检查 API 速率限制

### 凭据缺失 ``` Warning: Node 'HTTP_Request' may require credentials ``` **解决方案：** 1. 在 n8n 界面中配置凭据 2. 为节点分配凭据 3. 激活前测试连接

## 文件结构

``` ~/clawd/skills/n8n/ ├── SKILL.md # This file ├── scripts/ │ ├── n8n_api.py # Core API client (extended) │ ├── n8n_tester.py # Testing & validation │ └── n8n_optimizer.py # Performance optimization └── references/ └── api.md # n8n API reference ```

## API 参考

详细的 n8n REST API 文档，请参阅 [references/api.md](references/api.md) 或访问： https://docs.n8n.io/api/

## 支持

**文档：** - n8n 官方文档：https://docs.n8n.io - n8n 社区论坛：https://community.n8n.io - n8n API 参考：https://docs.n8n.io/api/

**调试：** 1. 使用验证：`python3 scripts/n8n_tester.py validate --id <workflow-id>` 2. 检查执行日志：`python3 scripts/n8n_api.py get-execution --id <execution-id>` 3. 查看优化报告：`python3 scripts/n8n_optimizer.py report --id <workflow-id>` 4. 使用试运行测试：`python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data-file test.json`