SERP API + n8n 工作流自动化监控教程

场景

之前给客户做 SEO 监控,每天 9 点 cron 跑 50 个关键词查 SERP 排名,变化超 3 位发 Slack 告警。这个流程用 n8n 搭只需 30 分钟,代码 0 行。

5 分钟搭起来

1. 启动 n8n

docker run -it –rm \\
–name n8n \\
-p 5678:5678 \\
-v ~/.n8n:/home/node/.n8n \\
n8nio/n8n

浏览器打开 http://localhost:5678。

2. 创建 workflow

点 “New workflow” → 加 3 个节点:

节点 1:Cron(每天 9 点)

Type: Schedule
Time: 0 9 * * *

节点 2:HTTP Request(调 SERP API)

Type: HTTP Request
Method: POST
URL: https://api.serpbase.dev/google/search
Authentication: Generic Credential Type
– Header Auth
– Name: X-API-Key
– Value: your-serpbase-key
Body:
{
"q": "your brand keyword",
"gl": "us",
"hl": "en",
"page": 1,
"num": 10
}

节点 3:Function(处理响应)

// n8n Function node
const results = items[0].json;
const newData = [];

for (const item of results.organic || []) {
if (item.link && item.link.includes("yourdomain.com")) {
newData.push({
title: item.title,
link: item.link,
rank: item.rank,
});
}
}

return [{ json: { results: newData } }];

节点 4:Save to Database

Type: Postgres / MySQL / Airtable
Operation: Insert

存排名数据,做历史对比。

节点 5:Compare with Yesterday

SELECT rank FROM serp_history
WHERE keyword = '{{$json.keyword}}'
ORDER BY ts DESC LIMIT 1;

节点 6:If (变化超 3 位)

Condition: {{$json.today_rank}} – {{$json.yesterday_rank}} >= 3

节点 7:Slack Alert

Type: Slack
Channel: #seo-alerts
Text: 🚨 SERP Alert
{{$json.keyword}} 排名变化:
– 昨天: #{{$json.yesterday_rank}}
– 今天: #{{$json.today_rank}}
– 变化: {{$json.today_rank – $json.yesterday_rank}}
{{$json.url}}

完整 workflow JSON

{
"name": "SERP Monitor",
"nodes": [
{
"parameters": {
"rule": {
"interval": [{}]
}
},
"name": "Schedule Trigger",
"type": "n8n-nodes-base.scheduleTrigger",
"position": [0, 0]
},
{
"parameters": {
"url": "https://api.serpbase.dev/google/search",
"authentication": "predefinedCredentialType",
"nodeCredentialType": "httpHeaderAuth",
"requestMethod": "POST",
"jsonBody": "{\\"q\\":\\"SERP API\\",\\"gl\\":\\"us\\",\\"hl\\":\\"en\\",\\"page\\":1,\\"num\\":10}",
"options": {}
},
"name": "SERP API",
"type": "n8n-nodes-base.httpRequest",
"position": [200, 0]
},
{
"parameters": {
"functionCode": "const r = items[0].json.organic || []; return r.map(i => ({title: i.title, link: i.link, rank: i.rank}));"
},
"name": "Process",
"type": "n8n-nodes-base.function",
"position": [400, 0]
}
],
"connections": {
"Schedule Trigger": {
"main": [[{"node": "SERP API", "type": "main", "index": 0}]]
},
"SERP API": {
"main": [[{"node": "Process", "type": "main", "index": 0}]]
}
}
}

5 分钟跑通

实战:3 个有用 workflow

Workflow 1:每日排名监控 + Slack 告警

[9:00 每天]
→ [调 SERP API 查 50 关键词]
→ [对比昨天数据(从 DB)]
→ [If 变化 > 3 位]
→ [Slack 告警]
→ [保存今天数据到 DB]

Workflow 2:竞品 SERP 监控

[周一 10:00 每周]
→ [查 5 个竞品域名 × 10 关键词 SERP]
→ [分析竞品位置变化]
→ [生成周报]
→ [Email 给 PM]

Workflow 3:AI Overview 引用监控

[每天 18:00]
→ [查 20 个品牌关键词 SERP]
→ [提取 ai_overview 字段]
→ [If 品牌被引用]
→ [Slack 告警 + 写记录到 Notion]
→ [生成每日 AI 引用报告]

4 个关键设计

1. 错误处理

// 在 HTTP Request 节点加 onError workflow
{
"onError": "continueErrorOutput" // 失败不中断整个 workflow
}

或用 IF 节点判断 status code:

If {{$json.statusCode}} >= 400:
→ Send to DLQ (webhook)
→ Log to Sentry

2. 缓存(避免重复调)

// n8n 用 Redis 节点做 cache
if cache.get(`serp:${query}`):
return cached

result = callSerp(query)
cache.set(`serp:${query}`, result, ttl=300)

或用 n8n built-in 节点 “Wait + Repeat”:

If last call < 1 min ago:
Wait
Use cached result

3. 批量(避免 QPS 超)

n8n 用 SplitInBatches 节点:

[SERP API call]
→ [SplitInBatches: batch_size=5]
→ [Wait 1 sec]
→ [SERP API call batch 5]
→ [Loop 5 times]

或 Code 节点:

const keywords = ["python", "javascript", "rust", "go", "ruby"];
const results = [];
for (const k of keywords) {
// 调 SERP API
const r = await this.helpers.httpRequest({
method: "POST",
url: "https://api.serpbase.dev/google/search",
headers: { "X-API-Key": API_KEY },
body: { q: k, num: 5 },
});
results.push({ keyword: k, organic: r.organic });
// 限速
await this.helpers.sleep(200);
}
return results;

4. 监控 + 告警

n8n 自带 Slack / Email / Telegram / Webhook 节点,直接接告警。

告警模板:

:rotating_light: SERP 监控告警

关键词: {{$json.keyword}}
昨天: #{{$json.yesterday_rank}}
今天: #{{$json.today_rank}}
变化: {{$json.delta}}
URL: {{$json.url}}

5 个 n8n 最佳实践

部署到生产

Docker Compose

version: '3'
services:
n8n:
image: n8nio/n8n
ports:
– "5678:5678"
volumes:
– ~/.n8n:/home/node/.n8n
environment:
– N8N_HOST=0.0.0.0
– N8N_PORT=5678
– WEBHOOK_URL=https://yourdomain.com/
– GENERIC_TIMEZONE=America/New_York

K8s 部署

apiVersion: apps/v1
kind: Deployment
metadata:
name: n8n
spec:
replicas: 1
template:
spec:
containers:
– name: n8n
image: n8nio/n8n
ports:
– containerPort: 5678
env:
– name: DB_TYPE
value: postgres
– name: DB_POSTGRESDB_HOST
value: postgres
– name: SERPBASE_KEY
valueFrom:
secretKeyRef:
name: serp–secrets
key: api–key

集成 Slack / Email / Webhook

n8n 自带 200+ 集成节点:

• Slack(告警)
• Email(报告)
• Webhook(自建 hook)
• Airtable(报告存储)
• Notion(文档)
• Telegram(告警)
• Discord(告警)
• Google Sheets(报告)

实战数据(客户项目 30 天)

n8n vs 自建脚本对比

中小项目用 n8n,大项目用 Python + n8n 混合。

4 个常见坑

坑 1:workflow 没 fail-safe

n8n 默认 HTTP 节点失败时整个 workflow 停。加 onError → “continueErrorOutput” 避免。

坑 2:数据太多爆内存

n8n 处理 100k 数据项会 OOM。批量处理:每次 100 项,Loop 到完成。

坑 3:SERP API rate limit 没考虑

n8n 跑快时 1 秒发 50 个请求,触发 429。加 Wait 节点(200ms) 或 SplitInBatches。

坑 4:key 写在 workflow 里

n8n 有 Credentials 管理,key 写在 Credentials 里,workflow 只引用,不要硬编码。

小结

SERP API + n8n 适合:

• 5-50 关键词的 SEO 监控
• 每天 1-100 次 SERP 调用
• 团队协作(UI 共享 workflow)
• 快速试错(可视化调试)

不适合:

• 1000+ QPS 大流量(自建 Python 更好)
• 复杂 ETL(用 Airflow)
• 长流程(> 10 节点,UI 难读)

我自己的客户项目跑 30 天,工程师 0 介入,自动化 100%。SerpBase + n8n + Slack + Postgres 四个组件,搭建只需 30 分钟。

SerpBase 的 auto-refund 让 n8n workflow 失败不扣 credit,可以放心做实验。