n8n-code-python

Python Code 节点（Beta）

在 n8n Code 节点中编写 Python 代码的专家指南。

---

⚠️ 重要提示：优先使用 JavaScript

建议：95% 的用例使用 JavaScript。仅在以下情况使用 Python：

• 需要特定的 Python 标准库函数
• 你更熟悉 Python 语法
• 进行更适合 Python 的数据转换

为什么优先使用 JavaScript：

• 完整的 n8n 辅助函数（this.helpers.httpRequest 等）
• Luxon DateTime 库用于高级日期/时间操作
• 无外部库限制
• 更好的 n8n 文档和社区支持

---

快速开始

# Python Code 节点的基本模板
items = _input.all()

# 处理数据
processed = []
for item in items:
    processed.append({
        "json": {
            **item["json"],
            "processed": True,
            "timestamp": datetime.now().isoformat()
        }
    })

return processed

基本规则

1. 优先考虑 JavaScript - 仅在必要时使用 Python
2. 访问数据：_input.all()、_input.first() 或 _input.item
3. 关键：必须返回 [{"json": {...}}] 格式
4. 关键：Webhook 数据位于 _json["body"] 下（而不是直接 _json）
5. 关键限制：无外部库（没有 requests、pandas、numpy）
6. 仅标准库：json、datetime、re、base64、hashlib、urllib.parse、math、random、statistics

---

模式选择指南

与 JavaScript 相同——根据用例选择：

对所有项目运行一次（推荐 - 默认）

此模式适用于： 95% 的用例

• 工作原理：无论输入数量多少，代码执行 一次
• 数据访问：_input.all() 或 _items 数组（原生模式）
• 最适合：聚合、过滤、批量处理、转换
• 性能：对于多个项目更快（单次执行）

# 示例：计算所有项目的总和
all_items = _input.all()
total = sum(item["json"].get("amount", 0) for item in all_items)

return [{
    "json": {
        "total": total,
        "count": len(all_items),
        "average": total / len(all_items) if all_items else 0
    }
}]

对每个项目运行一次

此模式适用于： 仅特殊用例

• 工作原理：代码对每个输入项目 分别 执行
• 数据访问：_input.item 或 _item（原生模式）
• 最适合：项目特定逻辑、独立操作、逐项验证
• 性能：对于大数据集较慢（多次执行）

# 示例：为每个项目添加处理时间戳
item = _input.item

return [{
    "json": {
        **item["json"],
        "processed": True,
        "processed_at": datetime.now().isoformat()
    }
}]

---

Python 模式：Beta 与原生

n8n 提供两种 Python 执行模式：

Python（Beta）- 推荐

• 使用：_input、_json、_node 辅助语法
• 最适合：大多数 Python 用例
• 可用辅助函数：_now、_today、_jmespath()
• 导入：from datetime import datetime

# Python（Beta）示例
items = _input.all()
now = _now  # 内置 datetime 对象

return [{
    "json": {
        "count": len(items),
        "timestamp": now.isoformat()
    }
}]

Python（原生）（Beta）

• 使用：仅 _items、_item 变量
• 无辅助函数：没有 _input、_now 等
• 更受限：仅标准 Python
• 何时使用：需要纯 Python 而不需要 n8n 辅助函数时

# Python（原生）示例
processed = []

for item in _items:
    processed.append({
        "json": {
            "id": item["json"].get("id"),
            "processed": True
        }
    })

return processed

建议：使用 Python（Beta） 以获得更好的 n8n 集成。

---

数据访问模式

通过以下划线为前缀的变量访问输入数据。每个项目是一个形状为 {"json": {...}} 的字典，因此实际字段位于 ["json"] 下。

# 模式 1：_input.all() - 最常用。数组、批量操作、聚合
all_items = _input.all()            # 列表，包含 {"json": {...}} 字典

# 模式 2：_input.first() - 非常常用。单个对象、API 响应
data = _input.first()["json"]       # 内置安全性优于 all_items[0]

# 模式 3：_input.item - 仅用于“对每个项目运行一次”模式
current = _input.item["json"]       # 在“所有项目”模式下为 None/错误

# 模式 4：_node - 引用特定的命名节点
webhook_data = _node["Webhook"]["json"]
http_data = _node["HTTP Request"]["json"]

参见：DATA_ACCESS.md 获取完整指南——六个 _input.all() 配方（过滤、转换、聚合、排序、分组、去重）、_input.first() 和 _input.item 示例、多节点组合、JS 与 Python 变量对照表以及决策树。

---

关键：Webhook 数据结构

最常见的错误：Webhook 数据嵌套在 ["body"] 下

# ❌ 错误 - 会引发 KeyError
name = _json["name"]
email = _json["email"]

# ✅ 正确 - Webhook 数据位于 ["body"] 下
name = _json["body"]["name"]
email = _json["body"]["email"]

# ✅ 更安全 - 使用 .get() 安全访问
webhook_data = _json.get("body", {})
name = webhook_data.get("name")

原因：Webhook 节点将所有请求数据包装在 body 属性下。这包括 POST 数据、查询参数和 JSON 负载。

参见：DATA_ACCESS.md 获取完整的 Webhook 结构详情

---

返回格式要求

关键规则：始终返回包含 "json" 键的字典列表

正确的返回格式

# ✅ 单个结果
return [{
    "json": {
        "field1": value1,
        "field2": value2
    }
}]

# ✅ 多个结果
return [
    {"json": {"id": 1, "data": "first"}},
    {"json": {"id": 2, "data": "second"}}
]

# ✅ 列表推导式
transformed = [
    {"json": {"id": item["json"]["id"], "processed": True}}
    for item in _input.all()
    if item["json"].get("valid")
]
return transformed

# ✅ 空结果（当没有数据返回时）
return []

# ✅ 条件返回
if should_process:
    return [{"json": processed_data}]
else:
    return []

错误的返回格式

# ❌ 错误：没有列表包装的字典
return {
    "json": {"field": value}
}

# ❌ 错误：没有 json 包装的列表
return [{"field": value}]

# ❌ 错误：纯字符串
return "processed"

# ❌ 错误：结构不完整
return [{"data": value}]  # 应为 {"json": value}

为什么重要：后续节点期望列表格式。格式不正确会导致工作流执行失败。

参见：ERROR_PATTERNS.md #2 获取详细的错误解决方案

---

关键限制：无外部库

最重要的 Python 限制：默认安装无法导入外部包。

自托管例外：外部包的可用性完全取决于实例的 Python 运行环境配置。如果用户说明其自托管实例的 Python 运行环境中具有特定包，则可以使用它们——不要拒绝。如果不确定，请询问或仅使用标准库编写代码。

❌ 不可用（会引发 ModuleNotFoundError）：requests、pandas、numpy、scipy、bs4/BeautifulSoup、lxml。

✅ 可用（仅标准库）：json、datetime、re、base64、hashlib、urllib.parse、math、random、statistics。

变通方法

需要 HTTP 请求？

• ✅ 在 Code 节点之前使用 HTTP Request 节点
• ✅ 或者切换到 JavaScript 并使用 this.helpers.httpRequest()（裸 $helpers 全局变量在任务运行器沙箱中未定义）

需要数据分析（pandas/numpy）？

• ✅ 使用 Python statistics 模块进行基本统计
• ✅ 或者切换到 JavaScript 进行大多数操作
• ✅ 使用列表和字典手动计算

需要网页抓取（BeautifulSoup）？

• ✅ 使用 HTTP Request 节点 + HTML Extract 节点
• ✅ 或者切换到 JavaScript 使用正则表达式/字符串方法

参见：STANDARD_LIBRARY.md 获取完整参考

---

常见模式概述

基于生产工作流，最有用的 Python 模式是：

1. 数据转换 - 使用列表推导式转换所有项目
2. 过滤与聚合 - 使用内置函数求和、过滤、计数
3. 字符串处理与正则表达式 - 使用 re 从文本中提取模式
4. 数据验证 - 验证和清理数据，附加错误列表
5. 统计分析 - 使用 statistics 模块计算均值/中位数/标准差

所有五个模式的即用片段位于 COMMON_PATTERNS.md，以及 10 个完整详细的生成模式（多源聚合、Markdown 解析、JSON 比较、CRM 标准化、字典查找、Top-N 过滤等）。

---

错误预防 - 前 5 个错误

1. 导入外部库（Python 特有）→ import requests 会引发 ModuleNotFoundError。改用 HTTP Request 节点或 JavaScript。
2. 代码为空或缺少 return → 每个路径必须以 return [{"json": ...}] 结束。
3. 返回格式不正确 → 包装在列表中：{"json": {...}} 变为 [{"json": {...}}]。
4. 字典访问的 KeyError → 使用 .get()：_json.get("user", {}).get("name", "Unknown")。
5. Webhook body 嵌套 → 通过 ["body"] 读取：_json.get("body", {}).get("email", "no-email")。

参见：ERROR_PATTERNS.md 获取完整指南——每个错误都有错误与正确代码、错误消息、嵌套访问修复、一个 AttributeError 额外案例、预防检查清单和快速修复表。

---

标准库参考

最有用的模块：json（解析/生成）、datetime（日期 + timedelta）、re（正则表达式）、base64（编码/解码）、hashlib（哈希）、urllib.parse（URL 操作）和 statistics（均值/中位数/标准差）。也可用：math、random、collections、itertools、functools。

有关简洁速查表以及每个模块的完整示例，请参见 STANDARD_LIBRARY.md。

---

最佳实践

1. 始终使用 .get() 进行字典访问

# ✅ 安全：如果字段缺失不会崩溃
value = item["json"].get("field", "default")

# ❌ 危险：如果字段不存在会崩溃
value = item["json"]["field"]

2. 显式处理 None/Null 值

# ✅ 好：如果为 None 则默认为 0
amount = item["json"].get("amount") or 0

# ✅ 好：显式检查 None
text = item["json"].get("text")
if text is None:
    text = ""

3. 使用列表推导式进行过滤

# ✅ Pythonic：列表推导式
valid = [item for item in items if item["json"].get("active")]

# ❌ 冗长：手动循环
valid = []
for item in items:
    if item["json"].get("active"):
        valid.append(item)

4. 返回一致的结构

# ✅ 一致：始终使用带 "json" 键的列表
return [{"json": result}]  # 单个结果
return results  # 多个结果（已格式化）
return []  # 无结果

5. 使用 print() 语句调试

# 调试语句显示在浏览器控制台中（F12）
items = _input.all()
print(f"Processing {len(items)} items")
print(f"First item: {items[0] if items else 'None'}")

---

生产环境注意事项

SplitInBatches 循环语义

SplitInBatches 节点有两个输出：

• main[0] = 完成 — 在所有批次完成后触发一次
• main[1] = 每个批次 — 为每个批次触发（循环体）

始终在完成输出后添加一个 Limit 1 节点。

正确的节点引用语法

# ❌ 错误
data = _node['HTTP Request']['json']

# ✅ 正确 - 调用 .first() 然后访问 json
data = _node['HTTP Request'].first()['json']

跨迭代数据在 Python 中不可用

$getWorkflowStaticData('global') 在 Python Beta 模式下可能不可用。如果需要在 SplitInBatches 迭代之间累积数据，请改用 JavaScript Code 节点进行累积逻辑。

---

何时使用 Python 与 JavaScript

使用 Python 当：

• ✅ 需要 statistics 模块进行统计操作
• ✅ 你更熟悉 Python 语法
• ✅ 你的逻辑适合列表推导式
• ✅ 需要特定的标准库函数

使用 JavaScript 当：

• ✅ 需要 HTTP 请求（this.helpers.httpRequest()）
• ✅ 需要高级日期/时间（DateTime/Luxon）
• ✅ 想要更好的 n8n 集成
• ✅ 对于 95% 的用例（推荐）

考虑其他节点当：

• ❌ 简单的字段映射 → 使用 Set 节点
• ❌ 基本过滤 → 使用 Filter 节点
• ❌ 简单条件 → 使用 IF 节点 或 Switch 节点
• ❌ 仅 HTTP 请求 → 使用 HTTP Request 节点

---

与其他技能的集成

配合使用：

n8n 表达式语法：

• 表达式在其他节点中使用 {{ }} 语法
• Code 节点直接使用 Python（无 {{ }}）
• 何时使用表达式与代码

n8n MCP 工具专家：

• 如何找到 Code 节点：search_nodes({query: "code"})
• 获取配置帮助：get_node({nodeType: "nodes-base.code"})
• 验证代码：validate_node({nodeType: "nodes-base.code", config: {...}})

n8n 节点配置：

• 模式选择（所有项目与每个项目）
• 语言选择（Python 与 JavaScript）
• 理解属性依赖

n8n 工作流模式：

• 转换步骤中的 Code 节点
• 在模式中何时使用 Python 与 JavaScript

n8n 验证专家：

• 验证 Code 节点配置
• 处理验证错误
• 自动修复常见问题

n8n Code JavaScript：

• 何时使用 JavaScript 替代
• JavaScript 与 Python 功能对比
• 从 Python 迁移到 JavaScript

---

快速参考检查清单

在部署 Python Code 节点之前，请验证：

• [ ] 已优先考虑 JavaScript - 仅在必要时使用 Python
• [ ] 代码不为空 - 必须有有意义的逻辑
• [ ] 存在 return 语句 - 必须返回字典列表
• [ ] 正确的返回格式 - 每个项目：{"json": {...}}
• [ ] 数据访问正确 - 使用 _input.all()、_input.first() 或 _input.item
• [ ] 无外部导入 - 仅标准库（json、datetime、re 等）
• [ ] 安全的字典访问 - 使用 .get() 避免 KeyError
• [ ] Webhook 数据 - 如果来自 webhook，通过 ["body"] 访问
• [ ] 模式选择 - 大多数情况下使用“所有项目”
• [ ] 输出一致 - 所有代码路径返回相同结构

---

其他资源

相关文件

• DATA_ACCESS.md - 全面的 Python 数据访问模式
• COMMON_PATTERNS.md - 10 个 n8n Python 模式
• ERROR_PATTERNS.md - 前 5 个错误及解决方案
• STANDARD_LIBRARY.md - 完整的标准库参考

n8n 文档

• Code 节点指南：https://docs.n8n.io/code/code-node/
• n8n 中的 Python：https://docs.n8n.io/code/builtin/python-modules/

---

准备好编写 n8n Code 节点的 Python 代码——但请优先考虑 JavaScript！ 为特定需求使用 Python，参考错误模式指南以避免常见错误，并有效利用标准库。