n8n-code-python

Name: n8n-code-python
Author: czlonkowski

github.com/czlonkowski开发23.2k85 次安装4 天前更新

代码仓库

关于 n8n-code-python

为需要 Python 语法或标准库功能的工作流构建者提供 n8n Code 节点指南，说明执行模式、数据访问、返回格式及 Python 执行限制。

主要功能

01选择批量或逐项执行模式
02访问输入与 Webhook 数据
03返回格式正确的数据项
04说明 Python Beta 与 Native 模式
05仅使用标准库的编写指导
06常见转换与分析模式

适用场景

01转换或筛选工作流数据
02使用标准库进行统计计算
03验证并清理输入数据

安装

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

$npx skills add https://github.com/czlonkowski/n8n-mcp --skill n8n-code-python

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

包含的文件

除 SKILL.md 外，安装时一起下载的参考资料和脚本。

COMMON_PATTERNS.md

DATA_ACCESS.md

ERROR_PATTERNS.md

README.md

STANDARD_LIBRARY.md

详细介绍

Python 代码节点（Beta）

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

⚠️ 重要：优先使用 JavaScript

建议：95% 的使用场景都应使用 JavaScript。仅在以下情况使用 Python：

需要特定的 Python 标准库函数
使用 Python 语法能让你轻松许多
进行更适合用 Python 实现的数据转换

优先使用 JavaScript 的原因：

可使用完整的 n8n 辅助函数（this.helpers.httpRequest 等）
可使用 Luxon DateTime 库进行高级日期/时间操作
不受外部库限制
n8n 文档和社区支持更完善

快速入门

# Basic template for Python Code nodes
items = _input.all()

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

return processed

基本规则

先考虑使用 JavaScript——仅在必要时使用 Python
访问数据：_input.all()、_input.first() 或 _input.item
重要：必须返回 [{"json": {...}}] 格式
重要：Webhook 数据位于 _json["body"] 下（而非直接位于 _json 中）
重要限制：不能使用外部库（不支持 requests、pandas、numpy）
仅限标准库：json、datetime、re、base64、hashlib、urllib.parse、math、random、statistics

模式选择指南

与 JavaScript 相同——根据使用场景选择：

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

**适用场景：**95% 的使用场景

工作方式：无论输入数量多少，代码都只会执行一次
数据访问：_input.all() 或 _items 数组（Native 模式）
适用场景：聚合、筛选、批处理、转换
性能：多个项目时速度更快（只执行一次）

# Example: Calculate total from 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（Native 模式）
适用场景：针对单个项目的逻辑、独立操作、逐项验证
性能：数据集较大时速度较慢（需要多次执行）

# Example: Add processing timestamp to each item
item = _input.item

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

Python 模式：Beta 与 Native

n8n 提供两种 Python 执行模式：

Python（Beta）——推荐

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

# Python (Beta) example
items = _input.all()
now = _now  # Built-in datetime object

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

Python（Native）（Beta）

使用：仅限 _items、_item 变量
没有辅助函数：不支持 _input、_now 等
功能更受限：仅支持标准 Python
适用场景：需要使用不含 n8n 辅助函数的纯 Python

# Python (Native) example
processed = []

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

return processed

建议：使用 Python（Beta），以便更好地集成 n8n。

数据访问模式

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

# Pattern 1: _input.all() - Most common. Arrays, batch ops, aggregations
all_items = _input.all()            # list of {"json": {...}} dicts

# Pattern 2: _input.first() - Very common. Single objects, API responses
data = _input.first()["json"]       # built-in safety vs all_items[0]

# Pattern 3: _input.item - "Run Once for Each Item" mode ONLY
current = _input.item["json"]       # None/error in All Items mode

# Pattern 4: _node - Reference a specific named 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"] 下

# ❌ WRONG - Will raise KeyError
name = _json["name"]
email = _json["email"]

# ✅ CORRECT - Webhook data is under ["body"]
name = _json["body"]["name"]
email = _json["body"]["email"]

# ✅ SAFER - Use .get() for safe access
webhook_data = _json.get("body", {})
name = webhook_data.get("name")

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

参阅：DATA_ACCESS.md，了解完整的 Webhook 结构详情

返回格式要求

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

正确的返回格式

# ✅ Single result
return [{
    "json": {
        "field1": value1,
        "field2": value2
    }
}]

# ✅ Multiple results
return [
    {"json": {"id": 1, "data": "first"}},
    {"json": {"id": 2, "data": "second"}}
]

# ✅ List comprehension
transformed = [
    {"json": {"id": item["json"]["id"], "processed": True}}
    for item in _input.all()
    if item["json"].get("valid")
]
return transformed

# ✅ Empty result (when no data to return)
return []

# ✅ Conditional return
if should_process:
    return [{"json": processed_data}]
else:
    return []

错误的返回格式

# ❌ WRONG: Dictionary without list wrapper
return {
    "json": {"field": value}
}

# ❌ WRONG: List without json wrapper
return [{"field": value}]

# ❌ WRONG: Plain string
return "processed"

# ❌ WRONG: Incomplete structure
return [{"data": value}]  # Should be {"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 模式包括：

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

这五种模式的即用代码片段都在 COMMON_PATTERNS.md 中，此外还包含 10 种完整详述的生产模式（多源聚合、Markdown 解析、JSON 比较、CRM 规范化、字典查找、Top-N 筛选等）。

错误预防 - 最常见的 5 个错误

导入外部库（Python 特有）→ import requests 会引发 ModuleNotFoundError。请改用 HTTP Request 节点或 JavaScript。
代码为空或缺少 return → 每条执行路径都必须以 return [{"json": ...}] 结束。
返回格式不正确 → 用列表包装：{"json": {...}} 应转换为 [{"json": {...}}]。
访问字典时出现 KeyError → 使用 .get()：_json.get("user", {}).get("name", "Unknown")。
Webhook 请求体嵌套 → 通过 ["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()

# ✅ SAFE: Won't crash if field missing
value = item["json"].get("field", "default")

# ❌ RISKY: Crashes if field doesn't exist
value = item["json"]["field"]

2. 明确处理 None/Null 值

# ✅ GOOD: Default to 0 if None
amount = item["json"].get("amount") or 0

# ✅ GOOD: Check for None explicitly
text = item["json"].get("text")
if text is None:
    text = ""

3. 使用列表推导式进行筛选

# ✅ PYTHONIC: List comprehension
valid = [item for item in items if item["json"].get("active")]

# ❌ VERBOSE: Manual loop
valid = []
for item in items:
    if item["json"].get("active"):
        valid.append(item)

4. 返回一致的结构

# ✅ CONSISTENT: Always list with "json" key
return [{"json": result}]  # Single result
return results  # Multiple results (already formatted)
return []  # No results

5. 使用 print() 语句进行调试

# Debug statements appear in browser console (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 节点。

正确的 Node 引用语法

# ❌ WRONG
data = _node['HTTP Request']['json']

# ✅ CORRECT - call .first() then access 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 节点配置：

模式选择（All Items 与 Each Item）
语言选择（Python 与 JavaScript）
了解属性依赖关系

n8n 工作流模式：

在转换步骤中使用 Code 节点
了解模式中何时使用 Python、何时使用 JavaScript

n8n 验证专家：

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

n8n Code JavaScript：

了解何时改用 JavaScript
对比 JavaScript 与 Python 的功能
从 Python 迁移到 JavaScript

快速参考清单

部署 Python Code 节点前，请确认：

其他资源

n8n 文档

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

准备好在 n8n Code 节点中编写 Python 了吗？但请优先考虑 JavaScript！ 有特定需求时再使用 Python，参考错误模式指南以避免常见错误，并有效利用标准库。

支持 Markdown

登录后评论

基本信息

作者: czlonkowski
仓库: czlonkowski/n8n-mcp
分类: 开发
安装次数: 85
GitHub Stars: 23,182

报告问题