Marico's space

如果你每月花20刀用n8n云版，或者花50刀用Zapier，我有个问题想问你：为什么？

你明明会Python或者JavaScript，五分钟就能写一个webhook处理器。为什么还要付费订阅，在GUI里拖来拖去，而不是用更快、更智能、几乎免费的方式自己搞定？

这篇文章里，我会展示如何用Claude API构建一个轻量级自动化引擎，替代大多数开发者使用n8n或Zapier时90%的场景——成本却只有零头。

开发者眼中的n8n和Zapier问题

n8n确实是款不错的软件。对非技术用户很友好。但如果你在看这篇文章，你大概率不是非技术用户。

你实际付费买的是这些东西：

• 一个你根本不需要的拖拽界面
• 预置连接器，其实就是10行Python能搞定的事
• 托管服务器跑你的工作流（这你自己免费就能跑）
• 厂商锁定，节点定义都是私有的

真正的问题不在于成本，而在于这些工具不是为想给自动化注入推理能力的开发者设计的。当你需要决定如何处理接收到的数据（不只是转换它），你需要的是LLM（大语言模型），而不是流程图构建器。

这就是Claude的用武之地。

架构：三层技术栈

Claude驱动的自动化引擎有三个组件：

[触发器] → [Claude推理层] → [动作执行器]

第一层：触发检测
外部事件发生：邮件到达、定时任务触发、webhook命中、文件出现在目录里。引擎被唤醒。

第二层：Claude推理
把原始事件传给Claude，同时附带一个系统提示描述决策规则。Claude返回结构化JSON：需要执行什么动作、用什么参数。

第三层：动作执行器
代码读取JSON并执行动作：发钉钉消息、创建飞书任务、写入数据库、调用API。

核心思路：Claude处理的是决策逻辑，这东西在n8n里要一堆条件分支，在Zapier里要写自定义代码。你用自然语言描述意图，Claude搞定路由。

动手构建：邮件转任务流水线

来个具体例子：读取收到的邮件，在SQLite数据库（换成飞书任务、Todoist也行）里创建任务。

前置条件：

• Python 3.9+
• anthropic库（pip install anthropic）
• 支持IMAP的邮箱账号（QQ邮箱、网易邮箱都支持）

核心引擎（约40行代码）

import imaplib
import email
import json
import sqlite3
import anthropic
from datetime import datetime client = anthropic.Anthropic() def fetch_unread_emails(host, user, password): mail = imaplib.IMAP4_SSL(host) mail.login(user, password) mail.select("inbox") _, data = mail.search(None, "UNSEEN") emails = [] for num in data[0].split(): _, msg_data = mail.fetch(num, "(RFC822)") msg = email.message_from_bytes(msg_data[0][1]) body = "" if msg.is_multipart(): for part in msg.walk(): if part.get_content_type() == "text/plain": body = part.get_payload(decode=True).decode() break else: body = msg.get_payload(decode=True).decode() emails.append({"subject": msg["Subject"], "from": msg["From"], "body": body[:500]}) mail.store(num, "+FLAGS", "\\Seen") mail.logout() return emails def classify_email(email_data): response = client.messages.create( model="claude-sonnet-4-6", max_tokens=256, system="""
You are an email classifier. Given an email, return JSON with:
- action: "create_task" | "reply_needed" | "ignore"
- priority: "high" | "medium" | "low"
- title: short task title if action is create_task
- due_days: integer days until due (0=today, 1=tomorrow, 7=next week)
Return only valid JSON, no commentary.""", messages=[{"role": "user", "content": f"Subject: {email_data['subject']}\nFrom: {email_data['from']}\n\n{email_data['body']}"}] ) return json.loads(response.content[0].text) def execute_action(decision, email_data): if decision["action"] == "create_task": conn = sqlite3.connect("tasks.db") conn.execute("""CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY, title TEXT, priority TEXT, source TEXT, created_at TEXT)""") conn.execute("INSERT INTO tasks (title, priority, source, created_at) VALUES (?, ?, ?, ?)", (decision["title"], decision["priority"], email_data["from"], datetime.now().isoformat())) conn.commit() conn.close() print(f"[TASK CREATED] {decision['priority'].upper()}: {decision['title']}") def run_pipeline(host, user, password): emails = fetch_unread_emails(host, user, password) for e in emails: decision = classify_email(e) execute_action(decision, e) if __name__ == "__main__": run_pipeline("imap.gmail.com", "you@gmail.com", "your-app-password")

42行代码实现了n8n里15个节点的工作流——而且它真的能理解你的邮件，而不是只会模式匹配。

成本对比：真实数字

假设每天收到500封邮件，每封都走Claude处理。

n8n云版（入门版）：每月20刀

Claude API费用：

• 模型：claude-haiku-4-5（最便宜的，处理分类足够）
• 输入：每封邮件约200 tokens × 500封 = 每天10万tokens = 每月300万tokens
• 费用按0.25刀/百万输入tokens = 0.75刀/月
• 输出：每封邮件约50 tokens × 500 = 每月150万tokens
• 费用按1.25刀/百万输出tokens = 1.88刀/月

Claude API总费用：约2.63刀/月 vs n8n的20刀/月

节省了87%。而且你的自动化现在有了推理能力——需要更智能时把Haiku换成Sonnet，多付点就行。开关在你手里。

如果开启提示缓存（cache_control参数），重复的系统提示会以0.1倍价格缓存。规模化后成本还能压得更低。

添加定时触发和Webhook触发

邮件轮询是拉取式的。想加推送触发，加个简单的Flask接口就行：

from flask import Flask, request
import threading app = Flask(__name__) @app.route("/webhook", methods=["POST"])
def handle_webhook(): data = request.json decision = classify_email(data) execute_action(decision, data) return {"status": "ok"} def cron_loop(): import time while True: run_pipeline("imap.gmail.com", "you@gmail.com", "app-password") time.sleep(300) # every 5 minutes if __name__ == "__main__": threading.Thread(target=cron_loop, daemon=True).start() app.run(port=8080)

现在你同时有了拉取（每5分钟轮询邮件）和推送（webhook）两种触发方式，都走同一个Claude推理层。

错误处理和重试模式

生产环境的自动化需要韧性。三种模式覆盖大多数失败场景：

1. Claude API调用的指数退避

import time def classify_with_retry(email_data, max_retries=3): for attempt in range(max_retries): try: return classify_email(email_data) except anthropic.RateLimitError: time.sleep(2 ** attempt) return {"action": "ignore", "priority": "low"}

2. JSON解析降级方案

Claude在明确指示后几乎总是返回有效JSON。但加个安全网：

import re def safe_parse(response_text): try: return json.loads(response_text) except json.JSONDecodeError: match = re.search(r'\{.*\}', response_text, re.DOTALL) if match: return json.loads(match.group()) return {"action": "ignore"}

3. 死信队列

把失败记录下来供人工审查，而不是悄悄丢弃：

def execute_action_safe(decision, email_data): try: execute_action(decision, email_data) except Exception as e: conn = sqlite3.connect("tasks.db") conn.execute("""CREATE TABLE IF NOT EXISTS failed_items (subject TEXT, error TEXT, raw_decision TEXT, created_at TEXT)""") conn.execute("INSERT INTO failed_items VALUES (?, ?, ?, ?)", (email_data["subject"], str(e), json.dumps(decision), datetime.now().isoformat())) conn.commit() conn.close()

什么时候继续用n8n

公平地说，以下场景n8n依然更好：

• 团队里非工程师需要修改工作流
• 复杂多步骤流水线，有20+个集成需要对接，不想手写代码
• 需求还不明确时快速原型验证

但如果你是懂Python的开发者，需求清晰，想要自动化有推理能力——你不需要为那个GUI界面付税。

下一步可以做什么

这个引擎是基础。在此之上可以：

• 接入向量数据库，让Claude记住过去的决策
• 构建多步骤工作流，Claude输出动作链
• 部署到5刀/月的VPS，24/7运行，不需要云订阅
• 开启工具调用，让Claude直接调用API而不是返回JSON

"无代码自动化工具"和"定制化自动化引擎"之间的差距，以前需要一个完整的工程冲刺才能跨越。有了Claude API，一个下午就够。

想要完整实现？我打包了一个生产级的Claude浏览器自动化启动包，包含可用代码、提示模板和部署指南。

获取Claude浏览器智能体启动包 —— Payhip