n8n-code-python by czlonkowski/n8n-skills
npx skills add https://github.com/czlonkowski/n8n-skills --skill n8n-code-python在 n8n 代码节点中编写 Python 代码的专家指南。
建议:95% 的使用场景请使用 JavaScript。仅在以下情况下使用 Python:
为什么首选 JavaScript:
# Python 代码节点的基本模板
items = _input.all()
# 处理数据
processed = []
for item in items:
processed.append({
"json": {
**item["json"],
"processed": True,
"timestamp": datetime.now().isoformat()
}
})
return processed
_input.all()、_input.first() 或 广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
_input.item[{"json": {...}}] 格式_json["body"] 下(而非直接访问 _json)与 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() } }]
n8n 提供两种 Python 执行模式:
使用:_input、_json、_node 辅助语法
最适合:大多数 Python 使用场景
可用辅助函数:_now、_today、_jmespath()
导入:from datetime import datetime
items = _input.all() now = _now # 内置的 datetime 对象
return [{ "json": { "count": len(items), "timestamp": now.isoformat() } }]
使用:仅限 _items、_item 变量
无辅助函数:没有 _input、_now 等
更受限:仅限标准 Python
使用时机:需要纯 Python 而不使用 n8n 辅助函数时
processed = []
for item in _items: processed.append({ "json": { "id": item["json"].get("id"), "processed": True } })
return processed
建议:为了更好的 n8n 集成,请使用 Python(测试版)。
使用时机:处理数组、批处理操作、聚合时
# 从前一个节点获取所有项
all_items = _input.all()
# 根据需要过滤、转换
valid = [item for item in all_items if item["json"].get("status") == "active"]
processed = []
for item in valid:
processed.append({
"json": {
"id": item["json"]["id"],
"name": item["json"]["name"]
}
})
return processed
使用时机:处理单个对象、API 响应时
# 仅获取第一项
first_item = _input.first()
data = first_item["json"]
return [{
"json": {
"result": process_data(data),
"processed_at": datetime.now().isoformat()
}
}]
使用时机:在“对每个项运行一次”模式下
# 循环中的当前项(仅限“对每个项运行一次”模式)
current_item = _input.item
return [{
"json": {
**current_item["json"],
"item_processed": True
}
}]
使用时机:需要工作流中特定节点的数据时
# 从特定节点获取输出
webhook_data = _node["Webhook"]["json"]
http_data = _node["HTTP Request"]["json"]
return [{
"json": {
"combined": {
"webhook": webhook_data,
"api": http_data
}
}
}]
参见:DATA_ACCESS.md 获取完整指南
最常见的错误: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 限制:无法导入外部包
# ❌ 不可用 - 将引发 ModuleNotFoundError
import requests # ❌ 否
import pandas # ❌ 否
import numpy # ❌ 否
import scipy # ❌ 否
from bs4 import BeautifulSoup # ❌ 否
import lxml # ❌ 否
# ✅ 可用 - 仅限标准库
import json # ✅ JSON 解析
import datetime # ✅ 日期/时间操作
import re # ✅ 正则表达式
import base64 # ✅ Base64 编码/解码
import hashlib # ✅ 哈希函数
import urllib.parse # ✅ URL 解析
import math # ✅ 数学函数
import random # ✅ 随机数
import statistics # ✅ 统计函数
需要 HTTP 请求?
$helpers.httpRequest()需要数据分析(pandas/numpy)?
需要网页抓取(BeautifulSoup)?
参见:STANDARD_LIBRARY.md 获取完整参考
基于生产工作流,以下是最有用的 Python 模式:
使用列表推导式转换所有项
items = _input.all()
return [
{
"json": {
"id": item["json"].get("id"),
"name": item["json"].get("name", "Unknown").upper(),
"processed": True
}
}
for item in items
]
使用内置函数进行求和、过滤、计数
items = _input.all()
total = sum(item["json"].get("amount", 0) for item in items)
valid_items = [item for item in items if item["json"].get("amount", 0) > 0]
return [{
"json": {
"total": total,
"count": len(valid_items)
}
}]
从文本中提取模式
import re
items = _input.all()
email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
all_emails = []
for item in items:
text = item["json"].get("text", "")
emails = re.findall(email_pattern, text)
all_emails.extend(emails)
# 去重
unique_emails = list(set(all_emails))
return [{
"json": {
"emails": unique_emails,
"count": len(unique_emails)
}
}]
验证并清理数据
items = _input.all()
validated = []
for item in items:
data = item["json"]
errors = []
# 验证字段
if not data.get("email"):
errors.append("Email required")
if not data.get("name"):
errors.append("Name required")
validated.append({
"json": {
**data,
"valid": len(errors) == 0,
"errors": errors if errors else None
}
})
return validated
使用 statistics 模块计算统计信息
from statistics import mean, median, stdev
items = _input.all()
values = [item["json"].get("value", 0) for item in items if "value" in item["json"]]
if values:
return [{
"json": {
"mean": mean(values),
"median": median(values),
"stdev": stdev(values) if len(values) > 1 else 0,
"min": min(values),
"max": max(values),
"count": len(values)
}
}]
else:
return [{"json": {"error": "No values found"}}]
参见:COMMON_PATTERNS.md 获取 10 个详细的 Python 模式
# ❌ 错误:尝试导入外部库
import requests # ModuleNotFoundError!
# ✅ 正确:使用 HTTP 请求节点或 JavaScript
# 在代码节点之前添加 HTTP 请求节点
# 或者切换到 JavaScript 并使用 $helpers.httpRequest()
# ❌ 错误:没有 return 语句
items = _input.all()
# 处理中...
# 忘记返回了!
# ✅ 正确:始终返回数据
items = _input.all()
# 处理中...
return [{"json": item["json"]} for item in items]
# ❌ 错误:返回字典而非列表包装
return {"json": {"result": "success"}}
# ✅ 正确:需要列表包装
return [{"json": {"result": "success"}}]
# ❌ 错误:直接访问,若字段缺失会导致崩溃
name = _json["user"]["name"] # KeyError!
# ✅ 正确:使用 .get() 进行安全访问
name = _json.get("user", {}).get("name", "Unknown")
# ❌ 错误:直接访问 webhook 数据
email = _json["email"] # KeyError!
# ✅ 正确:Webhook 数据位于 ["body"] 下
email = _json["body"]["email"]
# ✅ 更好:使用 .get() 进行安全访问
email = _json.get("body", {}).get("email", "no-email")
参见:ERROR_PATTERNS.md 获取全面的错误指南
# JSON 操作
import json
data = json.loads(json_string)
json_output = json.dumps({"key": "value"})
# 日期/时间
from datetime import datetime, timedelta
now = datetime.now()
tomorrow = now + timedelta(days=1)
formatted = now.strftime("%Y-%m-%d")
# 正则表达式
import re
matches = re.findall(r'\d+', text)
cleaned = re.sub(r'[^\w\s]', '', text)
# Base64 编码
import base64
encoded = base64.b64encode(data).decode()
decoded = base64.b64decode(encoded)
# 哈希
import hashlib
hash_value = hashlib.sha256(text.encode()).hexdigest()
# URL 解析
import urllib.parse
params = urllib.parse.urlencode({"key": "value"})
parsed = urllib.parse.urlparse(url)
# 统计
from statistics import mean, median, stdev
average = mean([1, 2, 3, 4, 5])
参见:STANDARD_LIBRARY.md 获取完整参考
# ✅ 安全:如果字段缺失不会崩溃
value = item["json"].get("field", "default")
# ❌ 有风险:如果字段不存在会导致崩溃
value = item["json"]["field"]
# ✅ 良好:如果为 None 则默认为 0
amount = item["json"].get("amount") or 0
# ✅ 良好:显式检查 None
text = item["json"].get("text")
if text is None:
text = ""
# ✅ 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)
# ✅ 一致:始终返回包含 "json" 键的列表
return [{"json": result}] # 单个结果
return results # 多个结果(已格式化)
return [] # 无结果
# 调试语句会出现在浏览器控制台(F12)
items = _input.all()
print(f"Processing {len(items)} items")
print(f"First item: {items[0] if items else 'None'}")
statistics 模块进行统计操作n8n 表达式语法:
{{ }} 语法{{ }})n8n MCP 工具专家:
search_nodes({query: "code"})get_node_essentials("nodes-base.code")validate_node_operation()n8n 节点配置:
n8n 工作流模式:
n8n 验证专家:
n8n 代码 JavaScript:
部署 Python 代码节点前,请验证:
{"json": {...}}_input.all()、_input.first() 或 _input.item.get() 避免 KeyError["body"] 访问准备好使用 n8n 代码节点编写 Python 代码 - 但请首先考虑 JavaScript! 针对特定需求使用 Python,参考错误模式指南以避免常见错误,并有效利用标准库。
每周安装数
628
仓库
GitHub 星标数
3.4K
首次出现
2026年1月20日
安全审计
已安装于
opencode518
gemini-cli499
codex482
github-copilot437
claude-code430
cursor406
Expert guidance for writing Python code in n8n Code nodes.
Recommendation : Use JavaScript for 95% of use cases. Only use Python when:
Why JavaScript is preferred:
# 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
_input.all(), _input.first(), or _input.item[{"json": {...}}] format_json["body"] (not _json directly)Same as JavaScript - choose based on your use case:
Use this mode for: 95% of use cases
How it works : Code executes once regardless of input count
Data access : _input.all() or _items array (Native mode)
Best for : Aggregation, filtering, batch processing, transformations
Performance : Faster for multiple items (single execution)
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 } }]
Use this mode for: Specialized cases only
How it works : Code executes separately for each input item
Data access : _input.item or _item (Native mode)
Best for : Item-specific logic, independent operations, per-item validation
Performance : Slower for large datasets (multiple executions)
item = _input.item
return [{ "json": { **item["json"], "processed": True, "processed_at": datetime.now().isoformat() } }]
n8n offers two Python execution modes:
Use : _input, _json, _node helper syntax
Best for : Most Python use cases
Helpers available : _now, _today, _jmespath()
Import : from datetime import datetime
items = _input.all() now = _now # Built-in datetime object
return [{ "json": { "count": len(items), "timestamp": now.isoformat() } }]
Use : _items, _item variables only
No helpers : No _input, _now, etc.
More limited : Standard Python only
Use when : Need pure Python without n8n helpers
processed = []
for item in _items: processed.append({ "json": { "id": item["json"].get("id"), "processed": True } })
return processed
Recommendation : Use Python (Beta) for better n8n integration.
Use when : Processing arrays, batch operations, aggregations
# Get all items from previous node
all_items = _input.all()
# Filter, transform as needed
valid = [item for item in all_items if item["json"].get("status") == "active"]
processed = []
for item in valid:
processed.append({
"json": {
"id": item["json"]["id"],
"name": item["json"]["name"]
}
})
return processed
Use when : Working with single objects, API responses
# Get first item only
first_item = _input.first()
data = first_item["json"]
return [{
"json": {
"result": process_data(data),
"processed_at": datetime.now().isoformat()
}
}]
Use when : In "Run Once for Each Item" mode
# Current item in loop (Each Item mode only)
current_item = _input.item
return [{
"json": {
**current_item["json"],
"item_processed": True
}
}]
Use when : Need data from specific nodes in workflow
# Get output from specific node
webhook_data = _node["Webhook"]["json"]
http_data = _node["HTTP Request"]["json"]
return [{
"json": {
"combined": {
"webhook": webhook_data,
"api": http_data
}
}
}]
See : DATA_ACCESS.md for comprehensive guide
MOST COMMON MISTAKE : Webhook data is nested under ["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")
Why : Webhook node wraps all request data under body property. This includes POST data, query parameters, and JSON payloads.
See : DATA_ACCESS.md for full webhook structure details
CRITICAL RULE : Always return list of dictionaries with "json" key
# ✅ 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}
Why it matters : Next nodes expect list format. Incorrect format causes workflow execution to fail.
See : ERROR_PATTERNS.md #2 for detailed error solutions
MOST IMPORTANT PYTHON LIMITATION : Cannot import external packages
# ❌ NOT AVAILABLE - Will raise ModuleNotFoundError
import requests # ❌ No
import pandas # ❌ No
import numpy # ❌ No
import scipy # ❌ No
from bs4 import BeautifulSoup # ❌ No
import lxml # ❌ No
# ✅ AVAILABLE - Standard library only
import json # ✅ JSON parsing
import datetime # ✅ Date/time operations
import re # ✅ Regular expressions
import base64 # ✅ Base64 encoding/decoding
import hashlib # ✅ Hashing functions
import urllib.parse # ✅ URL parsing
import math # ✅ Math functions
import random # ✅ Random numbers
import statistics # ✅ Statistical functions
Need HTTP requests?
$helpers.httpRequest()Need data analysis (pandas/numpy)?
Need web scraping (BeautifulSoup)?
See : STANDARD_LIBRARY.md for complete reference
Based on production workflows, here are the most useful Python patterns:
Transform all items with list comprehensions
items = _input.all()
return [
{
"json": {
"id": item["json"].get("id"),
"name": item["json"].get("name", "Unknown").upper(),
"processed": True
}
}
for item in items
]
Sum, filter, count with built-in functions
items = _input.all()
total = sum(item["json"].get("amount", 0) for item in items)
valid_items = [item for item in items if item["json"].get("amount", 0) > 0]
return [{
"json": {
"total": total,
"count": len(valid_items)
}
}]
Extract patterns from text
import re
items = _input.all()
email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
all_emails = []
for item in items:
text = item["json"].get("text", "")
emails = re.findall(email_pattern, text)
all_emails.extend(emails)
# Remove duplicates
unique_emails = list(set(all_emails))
return [{
"json": {
"emails": unique_emails,
"count": len(unique_emails)
}
}]
Validate and clean data
items = _input.all()
validated = []
for item in items:
data = item["json"]
errors = []
# Validate fields
if not data.get("email"):
errors.append("Email required")
if not data.get("name"):
errors.append("Name required")
validated.append({
"json": {
**data,
"valid": len(errors) == 0,
"errors": errors if errors else None
}
})
return validated
Calculate statistics with statistics module
from statistics import mean, median, stdev
items = _input.all()
values = [item["json"].get("value", 0) for item in items if "value" in item["json"]]
if values:
return [{
"json": {
"mean": mean(values),
"median": median(values),
"stdev": stdev(values) if len(values) > 1 else 0,
"min": min(values),
"max": max(values),
"count": len(values)
}
}]
else:
return [{"json": {"error": "No values found"}}]
See : COMMON_PATTERNS.md for 10 detailed Python patterns
# ❌ WRONG: Trying to import external library
import requests # ModuleNotFoundError!
# ✅ CORRECT: Use HTTP Request node or JavaScript
# Add HTTP Request node before Code node
# OR switch to JavaScript and use $helpers.httpRequest()
# ❌ WRONG: No return statement
items = _input.all()
# Processing...
# Forgot to return!
# ✅ CORRECT: Always return data
items = _input.all()
# Processing...
return [{"json": item["json"]} for item in items]
# ❌ WRONG: Returning dict instead of list
return {"json": {"result": "success"}}
# ✅ CORRECT: List wrapper required
return [{"json": {"result": "success"}}]
# ❌ WRONG: Direct access crashes if missing
name = _json["user"]["name"] # KeyError!
# ✅ CORRECT: Use .get() for safe access
name = _json.get("user", {}).get("name", "Unknown")
# ❌ WRONG: Direct access to webhook data
email = _json["email"] # KeyError!
# ✅ CORRECT: Webhook data under ["body"]
email = _json["body"]["email"]
# ✅ BETTER: Safe access with .get()
email = _json.get("body", {}).get("email", "no-email")
See : ERROR_PATTERNS.md for comprehensive error guide
# JSON operations
import json
data = json.loads(json_string)
json_output = json.dumps({"key": "value"})
# Date/time
from datetime import datetime, timedelta
now = datetime.now()
tomorrow = now + timedelta(days=1)
formatted = now.strftime("%Y-%m-%d")
# Regular expressions
import re
matches = re.findall(r'\d+', text)
cleaned = re.sub(r'[^\w\s]', '', text)
# Base64 encoding
import base64
encoded = base64.b64encode(data).decode()
decoded = base64.b64decode(encoded)
# Hashing
import hashlib
hash_value = hashlib.sha256(text.encode()).hexdigest()
# URL parsing
import urllib.parse
params = urllib.parse.urlencode({"key": "value"})
parsed = urllib.parse.urlparse(url)
# Statistics
from statistics import mean, median, stdev
average = mean([1, 2, 3, 4, 5])
See : STANDARD_LIBRARY.md for complete reference
# ✅ SAFE: Won't crash if field missing
value = item["json"].get("field", "default")
# ❌ RISKY: Crashes if field doesn't exist
value = item["json"]["field"]
# ✅ 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 = ""
# ✅ 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)
# ✅ CONSISTENT: Always list with "json" key
return [{"json": result}] # Single result
return results # Multiple results (already formatted)
return [] # No results
# 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'}")
statistics module for statistical operationsn8n Expression Syntax :
{{ }} syntax in other nodes{{ }})n8n MCP Tools Expert :
search_nodes({query: "code"})get_node_essentials("nodes-base.code")validate_node_operation()n8n Node Configuration :
n8n Workflow Patterns :
n8n Validation Expert :
n8n Code JavaScript :
Before deploying Python Code nodes, verify:
{"json": {...}}_input.all(), _input.first(), or _input.item.get() to avoid KeyError["body"] if from webhookReady to write Python in n8n Code nodes - but consider JavaScript first! Use Python for specific needs, reference the error patterns guide to avoid common mistakes, and leverage the standard library effectively.
Weekly Installs
628
Repository
GitHub Stars
3.4K
First Seen
Jan 20, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykWarn
Installed on
opencode518
gemini-cli499
codex482
github-copilot437
claude-code430
cursor406
React 组合模式指南:Vercel 组件架构最佳实践,提升代码可维护性
102,200 周安装