n8n 与 Claude API 工作流

n8n 是开源的工作流自动化平台——把 Claude 接入 n8n，就能用可视化方式把 AI 嵌入你的每一个业务流程，无需写代码。

n8n + Claude 自动化架构

n8n 工作流配置与 Claude 集成

"""
n8n + Claude API 工作流设计指南
展示常见自动化场景的配置方法
"""
import json
from dataclasses import dataclass, field
from enum import Enum
class TriggerType(Enum):
SCHEDULE = "Schedule Trigger"      # 定时
WEBHOOK = "Webhook"                # HTTP 触发
EMAIL = "Email Trigger (IMAP)"    # 邮件触发
FORM = "n8n Form Trigger"         # 表单提交
GITHUB = "GitHub Trigger"          # 代码事件
@dataclass
class N8nNode:
"""n8n 节点定义"""
node_type: str
name: str
parameters: dict
position: tuple[int, int] = (0, 0)
@dataclass
class ClaudeAPINode(N8nNode):
"""Claude API HTTP Request 节点配置"""
def __init__(self, name: str, prompt_expression: str, max_tokens: int = 1000):
super().__init__(
node_type="HTTP Request",
name=name,
parameters={
"method": "POST",
"url": "https://api.anthropic.com/v1/messages",
"authentication": "genericCredentialType",
"genericAuthType": "httpHeaderAuth",
"headers": {
"x-api-key": "={{ $credentials.httpHeaderAuth.value }}",
"anthropic-version": "2023-06-01",
"content-type": "application/json",
},
"body": json.dumps({
"model": "claude-haiku-4-5-20251001",
"max_tokens": max_tokens,
"messages": [{"role": "user", "content": f"={{{{ {prompt_expression} }}}}"}],
}),
"responseFormat": "json",
}
)
@dataclass
class WorkflowTemplate:
"""n8n 工作流模板"""
name: str
description: str
trigger: TriggerType
nodes: list[str]      # 节点描述
claude_prompt: str
output_destination: str
business_value: str
setup_minutes: int
class N8nWorkflowLibrary:
"""常用 n8n + Claude 工作流库"""
WORKFLOWS = [
WorkflowTemplate(
name="邮件智能分类归档",
description="新邮件到达 → Claude 分类 → 自动打标签+归档",
trigger=TriggerType.EMAIL,
nodes=["Email Trigger", "Claude API", "Gmail", "Slack Notify"],
claude_prompt="""分析这封邮件，返回 JSON：
{{
"category": "客户/内部/营销/垃圾",
"priority": "urgent/high/normal/low",
"action": "reply/archive/forward/delete",
"summary": "一句话摘要"
}}
邮件: {email_content}""",
output_destination="Gmail 标签 + Slack 通知",
business_value="每天节省 1–2 小时邮件处理时间",
setup_minutes=30,
),
WorkflowTemplate(
name="GitHub PR 自动 Code Review",
description="PR 提交 → Claude 审查代码 → 自动评论",
trigger=TriggerType.GITHUB,
nodes=["GitHub Trigger", "HTTP Request (Get PR)", "Claude API", "GitHub Comment"],
claude_prompt="""作为高级代码审查员，审查以下代码变更：
{diff_content}
提供：
1. 总体评价（1–2句）
2. 潜在 Bug（如有）
3. 安全隐患（如有）
4. 改进建议（最多3条）
5. 评分：1–10
用 Markdown 格式输出。""",
output_destination="GitHub PR 评论",
business_value="PR 审查效率提升 60%",
setup_minutes=45,
),
WorkflowTemplate(
name="每日报告自动生成",
description="每天 8 点 → 聚合数据 → Claude 生成报告 → 发 Slack",
trigger=TriggerType.SCHEDULE,
nodes=["Schedule (8am)", "Google Analytics", "Sheets", "Claude API", "Slack"],
claude_prompt="""基于以下数据，生成简洁的日报（适合 Slack 发送）：
昨日数据：
- 网站访问量: {pageviews}
- 转化率: {conversion_rate}
- 营收: {revenue}
- 新用户: {new_users}
请生成：
1. 关键指标摘要（表情符号风格）
2. 与上周对比（涨跌分析）
3. 今日需要关注的 1 件事""",
output_destination="Slack #daily-metrics 频道",
business_value="数据报告自动化，无需人工整理",
setup_minutes=60,
),
WorkflowTemplate(
name="客户表单智能回复",
description="表单提交 → Claude 生成个性化回复 → 自动发送",
trigger=TriggerType.FORM,
nodes=["Form Trigger", "Claude API", "Send Email", "CRM Update"],
claude_prompt="""客户提交了以下咨询表单，生成专业且温暖的回复邮件：
姓名: {name}
公司: {company}
问题: {inquiry}
产品兴趣: {product_interest}
要求：
- 称呼用客户名字
- 直接回答核心问题
- 提供下一步行动建议
- 100–150 字，不要超过""",
output_destination="客户邮件 + CRM 系统",
business_value="回复时间从 24h 缩短到 5 分钟",
setup_minutes=40,
),
WorkflowTemplate(
name="竞品监控简报",
description="每周一 → 抓取竞品更新 → Claude 分析 → 发给团队",
trigger=TriggerType.SCHEDULE,
nodes=["Schedule (Mon 9am)", "HTTP Request (RSS)", "Claude API", "Notion", "Slack"],
claude_prompt="""分析这周竞品的动态，提炼关键信息：
竞品更新内容：
{competitor_updates}
请分析：
1. 最重要的功能/产品变化（优先级排序）
2. 对我们的威胁程度（高/中/低）
3. 建议我们的应对策略（1–2条）""",
output_destination="Notion 竞品数据库 + Slack",
business_value="竞品监控自动化，不错过任何重要动态",
setup_minutes=90,
),
]
@classmethod
def print_library(cls):
print("=== n8n + Claude 工作流模板库 ===\n")
for wf in cls.WORKFLOWS:
print(f"【{wf.name}】⏱ 配置约 {wf.setup_minutes} 分钟")
print(f"  触发器: {wf.trigger.value}")
print(f"  节点链: {' → '.join(wf.nodes)}")
print(f"  输出到: {wf.output_destination}")
print(f"  价值: {wf.business_value}")
print()
class N8nDeployGuide:
"""n8n 部署方案"""
DEPLOYMENT_OPTIONS = {
"n8n Cloud (SaaS)": {
"月费": "$20+",
"上手时间": "5 分钟",
"适合": "快速验证，无需运维",
"限制": "执行次数有限，数据在 n8n 云端",
},
"Railway 自托管": {
"月费": "$5–10",
"上手时间": "30 分钟",
"适合": "个人/小团队，成本敏感",
"限制": "需要配置持久化存储",
},
"VPS 自托管（Docker）": {
"月费": "$5–20",
"上手时间": "1–2 小时",
"适合": "团队使用，数据本地控制",
"限制": "需要运维知识",
},
}
DOCKER_COMPOSE = """
# n8n Docker Compose 一键部署
version: '3.8'
services:
n8n:
image: n8nio/n8n
ports:
- "5678:5678"
environment:
- N8N_BASIC_AUTH_ACTIVE=true
- N8N_BASIC_AUTH_USER=admin
- N8N_BASIC_AUTH_PASSWORD=your_secure_password
- WEBHOOK_URL=https://your-domain.com
volumes:
- n8n_data:/home/node/.n8n
restart: unless-stopped
volumes:
n8n_data:
"""
@classmethod
def print_deploy_options(cls):
print("=== n8n 部署方案对比 ===\n")
for method, details in cls.DEPLOYMENT_OPTIONS.items():
print(f"【{method}】")
for k, v in details.items():
print(f"  {k}: {v}")
print()
# 演示
N8nWorkflowLibrary.print_library()
N8nDeployGuide.print_deploy_options()
print("=== Claude API 节点配置示例 ===")
claude_node = ClaudeAPINode(
name="Claude 邮件分类",
prompt_expression="'分类这封邮件: ' + $json.body.text",
max_tokens=500,
)
print(f"节点名称: {claude_node.name}")
print(f"请求方式: {claude_node.parameters['method']} {claude_node.parameters['url']}")

n8n + Claude 配置速查

行动清单

• [ ] 注册 n8n Cloud 免费试用，或用 Docker Compose 本地部署（见上方配置）
• [ ] 在 n8n 的 Credentials 中添加 Claude API Key（Header Auth，Key = x-api-key）
• [ ] 从"邮件智能分类"工作流开始——这是投入产出比最高的自动化
• [ ] 连接 Slack/Notion：在 n8n 中添加对应 Credential，测试通知推送
• [ ] 用"每日报告"模板替代手动整理数据——设置定时触发，接入你的 Analytics
• [ ] 建立错误告警工作流：n8n 执行失败 → Claude 分析错误 → Slack 通知修复建议

下一节：02-文档会议与邮件全自动化 — 把文档记录、会议摘要、邮件撰写全部接入自动化流水线。