通过更新 HTTP 请求节点 URL、切换 API 密钥标头以及调整响应字段路径，将 n8n 工作流程从 SerpAPI 迁移到 Scavio。迁移时间不到 15 分钟，因为两个 API 都返回类似的结构化搜索数据。 Scavio 使用带有 JSON 正文的 POST 端点，而不是带有查询参数的 SerpAPI 的 GET，并且响应字段名称略有不同。本教程映射每个字段更改并提供每个步骤的之前/之后代码。
前置条件
- 正在运行的 n8n 实例（自托管或 n8n 云）
- 来自 scavio.dev 的 Scavio API 密钥
- 使用 SerpAPI 的现有 n8n 工作流程
- SerpAPI API 密钥（用于比较测试）
操作指南
步骤 1: 映射请求格式
从 SerpAPI GET 请求转换为 Scavio POST 请求格式。
Python
import os, requests
# BEFORE (SerpAPI):
# resp = requests.get('https://serpapi.com/search',
# params={'api_key': os.environ['SERPAPI_KEY'], 'q': 'test', 'engine': 'google'})
# AFTER (Scavio):
API_KEY = os.environ['SCAVIO_API_KEY']
def search(query: str, platform: str = 'google') -> dict:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json={'platform': platform, 'query': query}, timeout=15)
return resp.json()
# Key differences:
# SerpAPI: GET, api_key in params, engine=google
# Scavio: POST, x-api-key in header, platform=google in body
result = search('test query')
print(f"Results: {len(result.get('organic_results', []))}")步骤 2: 映射响应字段
将 SerpAPI 响应字段名称映射到每种数据类型的 Scavio 字段名称。
Python
# Field mapping: SerpAPI -> Scavio
FIELD_MAP = {
# Organic results
'organic_results': 'organic_results', # Same!
'title': 'title', # Same!
'link': 'link', # Same!
'snippet': 'snippet', # Same!
'position': 'position', # Same!
# Other fields
'search_information': 'search_information',
'related_questions': 'people_also_ask', # Different!
'answer_box': 'featured_snippet', # Different!
}
def serpapi_to_scavio(serpapi_result: dict) -> dict:
"""Convert SerpAPI response format to match expected fields."""
# Most organic result fields are identical
# Main differences are in the wrapper fields
return {
'organic_results': serpapi_result.get('organic_results', []),
'people_also_ask': serpapi_result.get('related_questions', []),
'featured_snippet': serpapi_result.get('answer_box', {}),
}
# In practice, Scavio returns these fields directly:
data = search('best CRM 2026')
print(f"Organic: {len(data.get('organic_results', []))}")
print(f"PAA: {len(data.get('people_also_ask', []))}")步骤 3: 更新 n8n HTTP 请求节点
为 Scavio API 配置 n8n HTTP 请求节点。
Python
# n8n HTTP Request Node Configuration:
#
# BEFORE (SerpAPI):
# Method: GET
# URL: https://serpapi.com/search
# Query Parameters:
# api_key: {{$env.SERPAPI_KEY}}
# q: {{$json.query}}
# engine: google
#
# AFTER (Scavio):
# Method: POST
# URL: https://api.scavio.dev/api/v1/search
# Headers:
# x-api-key: {{$env.SCAVIO_API_KEY}}
# Body Content Type: JSON
# Body: {"platform": "google", "query": "{{$json.query}}"}
# Test both side by side:
def compare_responses(query: str) -> dict:
scavio = search(query)
scavio_results = scavio.get('organic_results', [])
return {
'query': query,
'scavio_count': len(scavio_results),
'scavio_top': scavio_results[0].get('title', '') if scavio_results else 'none',
}
comp = compare_responses('best CRM 2026')
print(f"Scavio: {comp['scavio_count']} results, top: {comp['scavio_top'][:50]}")步骤 4: 更新下游节点
调整引用 SerpAPI 特定字段路径的任何 n8n 表达式。
Python
# n8n Expression Updates:
#
# BEFORE (SerpAPI response paths):
# {{$json.organic_results[0].title}} -> Same
# {{$json.organic_results[0].link}} -> Same
# {{$json.organic_results[0].snippet}} -> Same
# {{$json.related_questions}} -> {{$json.people_also_ask}}
# {{$json.answer_box.snippet}} -> {{$json.featured_snippet.snippet}}
# {{$json.search_information.total_results}} -> {{$json.search_information.total_results}}
# Python equivalent for testing:
def extract_n8n_fields(data: dict) -> dict:
results = data.get('organic_results', [])
return {
'top_title': results[0].get('title', '') if results else '',
'top_link': results[0].get('link', '') if results else '',
'top_snippet': results[0].get('snippet', '') if results else '',
'paa': data.get('people_also_ask', []),
'featured': data.get('featured_snippet', {}),
'result_count': len(results),
}
data = search('best CRM software')
fields = extract_n8n_fields(data)
for key, value in fields.items():
display = str(value)[:60] if value else 'empty'
print(f' {key}: {display}')步骤 5: 测试迁移
运行验证测试以确认迁移产生相同的结果。
Python
def validate_migration(queries: list) -> dict:
passed = 0
failed = 0
for query in queries:
data = search(query)
results = data.get('organic_results', [])
# Check essential fields exist
checks = {
'has_results': len(results) > 0,
'has_titles': all(r.get('title') for r in results[:3]) if results else False,
'has_links': all(r.get('link') for r in results[:3]) if results else False,
'valid_json': isinstance(data, dict),
}
all_pass = all(checks.values())
if all_pass:
passed += 1
else:
failed += 1
print(f'FAIL: {query}: {checks}')
print(f'\nMigration validation: {passed} passed, {failed} failed')
return {'passed': passed, 'failed': failed}
validate_migration([
'best CRM 2026',
'python web framework comparison',
'how to deploy docker',
])
# Cost comparison:
# SerpAPI: $25/mo (5K searches), $75/mo (15K)
# Scavio: Free 250/mo, $30/mo (7K credits)
print('\nMigration complete. Cost savings start immediately.')Python 示例
Python
import requests, os
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
# SerpAPI drop-in replacement:
def search(query, platform='google'):
data = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': platform, 'query': query}).json()
return data.get('organic_results', [])
results = search('best CRM 2026')
for r in results[:3]:
print(f"{r.get('title', '')}: {r.get('link', '')}")JavaScript 示例
JavaScript
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
// SerpAPI drop-in replacement:
async function search(query, platform = 'google') {
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H,
body: JSON.stringify({platform, query})
});
return (await r.json()).organic_results || [];
}
search('best CRM 2026').then(results => results.slice(0, 3).forEach(r => console.log(r.title)));预期输出
JSON
A fully migrated n8n workflow using Scavio instead of SerpAPI, with updated HTTP nodes, response field mappings, and validation tests confirming equivalent output.