做跨境这几年, 我最大的感受是: 数据从来不是没有, 是"拉数据"本身把人累垮。打开十几个网页, 复制粘贴, 录 Excel, 再用聊天窗口问 AI, 一上午就过去了。后来我把整套流程压成了一个 Python 脚本加一个 n8n 调度, 每天 9 点自动跑, 跑完发到我微信。本文不堆术语, 只讲怎么落地, 代码我尽量写真实的、跑得动的版本。

先说背景。我用的是一个叫 MCP (Model Context Protocol) 的服务, 提供了 79 个工具, 覆盖选品、类目、关键词、产品详情、监控、趋势、评论这些跨境全链路。我自己的脚本不是直接对接它, 而是调一个本地的 CLI 客户端 (我装的是 npm 包 sf-cli, 团队开源的), 客户端内部再去打 MCP。为什么要绕一层 CLI? 因为 CLI 有 checkpoint / resume, 批量跑一半崩了能从断点续, 不浪费配额的活儿; 还有 field alias 自动把接口返回的怪字段名映射成我能读懂的英文, 不用我每次都翻文档。

先放一个最简版的拉取脚本, 我自己用来先跑通链路, 后面再往里加维度。

第一段, 我用 Python 3 的标准库 (没有 requests 也没有 pandas), 调本地 CLI 拿数据, 然后落成 JSON, 失败重试三次, 全跑完打一行日志。这套写法的好处是部署到任何一台 Windows 或者 Linux 都行, 不依赖虚拟环境。

脚本核心是 subprocess.run 调 sf-api.py (CLI 自带的 Python 包装), 参数就是 endpoint + JSON 字符串 + domain。我把 endpoint 抽成配置, 这样后面要加工具不动主循环。

第二段, 我用了一个叫 n8n 的开源工作流平台 (类似 Zapier 但可以自托管), 它的好处是 Cron 节点 + HTTP 节点 + 飞书/企微/钉钉节点 都是现成的。我不需要写 Web 服务, 直接在画布上拖几下, 每天 9 点触发, 调我的 Python 脚本 (我把它包成 HTTP 接口用 FastAPI 暴露), 跑完发个飞书消息到我自己的机器人。

下面把链路拆开讲。

一、脚本结构: 三层, 配置 / 执行 / 推送

第一层 config.yaml 放所有 endpoint、参数、目标市场。我现在跑了 Amazon 美国 (domain=1) + 沃尔玛美国 (domain=?) + 东南亚 Shopee (domain=201-208) 三类, 共 12 个常用工具, 不是我 79 个全用上, 是按业务挑的。挑的依据: 类目报告 (CategoryRequest) + 关键词反查 (ASINRequestKeywordv2) + 产品详情 (ProductRequest) + 子体销量 (AsinSalesVolume) + 趋势 (CategoryTrend) + 监控任务列表 (KeywordTasks)。这 6 个就能覆盖"日更看哪几个 ASIN 涨了 / 哪几个词掉出首页"。

第二层 runner.py 是主循环。它做的事按顺序是: 1) 读配置, 2) 对每个 endpoint 起一个 worker, 3) worker 调 sf-api.py, 4) 解析返回 JSON, 5) 落盘到 ./data/YYYY-MM-DD/<endpoint>.json, 6) 异常重试三次, 7) 汇总本次跑了多少个、失败多少个, 写进 summary.json。

第三层 notifier.py 是发飞书/企微的。它读 summary.json, 如果失败率 > 20% 就告警 (红色卡片), 否则发正常日报 (绿色卡片)。卡片内容我设计成 今天跑了 N 个 endpoint, 抓回 M 条记录, 失败 K 个, 详见附件, 附件是 ./data/YYYY-MM-DD 整个目录打 zip 传上去。

二、关键代码段 (我故意不写完整, 给思路)

第一段, runner.py 里的 worker 函数。参数是 endpoint、payload、domain、retry_times。我用 subprocess.run 调 python3 scripts/sf-api.py <endpoint> '<json>' --domain <d>, timeout 30 秒, 拿 stdout 解析。如果返回码非 0 或者 JSON 解析失败, 就进重试队列。重试用的是 指数退避, 1 秒、2 秒、4 秒, 最多三次。三次都挂, 这个 endpoint 标红, 不影响其他 endpoint 继续跑。

这一段我想强调的是 解耦: 失败的 endpoint 不要让整个流水线挂掉。我们做选品要的是"广度", 不是"每条都准", 缺一两条数据不影响决策, 但流水线每天断一次就废了。

第二段, 调 MCP 工具的写法。sf-api.py 本身是 CLI 工具的 Python 包装, 它干的事就是剥掉 CLI 的 info: 前缀行, 把真正 JSON 返出来。我没有自己重写 HTTP, 因为 CLI 内部已经处理了 gzip+base64 (Shopee 接口返回是压缩的), 字段名映射, 错误码转换。我只是在 Python 里用 subprocess 调它, 拿到 dict 之后用 pandas 或者纯 dict 操作都行。

注意一个坑: CLI 的 endpoint 名是 PascalCase (例如 ProductRequest、ASINRequestKeywordv2), 不是 README 里的小写。小写那个版本调不通, 我当时被坑了 20 分钟, 翻 issue 才看到。SKILL.md 里专门有一条警告。

第三段, 落盘的部分。我用 pathlib.Path("./data") / today / f"{endpoint}.json" 这种写法, 跨平台不踩坑。文件写完之后用 json.dump(..., ensure_ascii=False, indent=2), 方便我直接 cat 看。批量场景我用 jsonl 追加写, 避免一次写一个巨大的 list 把内存吃满。

三、n8n 怎么排: 三个节点就够

第一个节点是 Cron, 表达式 0 9 * * *, 每天北京时间 9 点触发。注意 n8n 服务如果在海外机器上, 时区要手动配, 我直接用 TZ=Asia/Shanghai 环境变量拉起服务, 简单粗暴。

第二个节点是 HTTP Request, POST 我的 runner.py 暴露的接口, body 是 {"date": "{{ $now.format('yyyy-MM-dd') }}", "endpoints": [...]}。我让 runner 接受 endpoint 列表, 这样 n8n 那边只发任务不下发细节, 后面要改工具不动 n8n 配置。

第三个节点是 飞书机器人, 把 runner 返回的 summary 喂进去, 用飞书的 interactive card 渲染。卡片模板我写好放在 templates/feishu-card.json 里, n8n 用 Function 节点 把 summary 字段映射到模板变量。

这套架构跑了一个月, 每天 9:00:05 准时出日报, 9:00:30 飞书消息到我手机, 我边吃早饭边看。失败了 4 次, 3 次是 n8n 服务自己挂了 (我没配 systemd, 后来加了), 1 次是某个 ASIN 触发了 MCP 限流, 我在 runner 里加了 if response.code == 98: sleep(60) and retry, 之后再没出过问题。

四、我想说的几个非技术问题

第一个, 79 个工具别一次全用。我见过有人一上来就写 79 个 endpoint 的全量调度, 结果每天配额跑完还看不到几条有用数据。我的建议是先做 MVP: 6-8 个最常用的工具, 跑两周, 看自己每天真的看哪些数据, 再加。工具是手段, 不是 KPI。

第二个, 本地落盘比直接喂给 AI 模型靠谱。我之前图省事, 抓回来直接喂给 LLM 解读, 结果上下文一长模型就开始幻觉, 字段名张冠李戴。现在我一律先落 JSON, 解读是另一段, LLM 只看需要的几条, 准确度高很多。

第三个, CI 监控的"静默失败"最可怕。我第一次部署的时候, 跑了一周才发现某天凌晨某个 endpoint 挂了, 之后 n8n 没报错, 飞书没收到 (因为没数据触发告警条件)。后来我在 runner 里加了 "今天如果 0 条数据也发一个空状态卡片", 这种"心跳包"才是真正的 SLA。

第四个, 学会用 CLI 工具的 log 模式。sf-api.py 有 --verbose 和 --dry-run 两种调试模式。--dry-run 打印要发的请求但不真发, 排错特别快; --verbose 把整个 HTTP trace 打出来, 限流/超时问题一眼能看出来。比起 print 大法, 工具自带的 log 模式更结构化。

五、扩展方向 (我还没做, 留个 TODO)

1) 把飞书卡片升级成 "今天该关注的 5 个 ASIN", 让 LLM 读今天的 summary.json 总结出 Top 5, 卡片里直接给链接。我现在还是人工看, 偶尔偷懒。

2) n8n 那边加 失败重试节点, 现在的设计是失败就发红色卡片, 不会自动补跑。可以加一个"失败后 30 分钟再跑一次, 还失败才告警"的逻辑。

3) 我现在的脚本是单机的, 哪天机器挂了 n8n 就触发不了。下一步考虑把 runner 部署到 云函数 (阿里云函数计算或者 AWS Lambda), n8n 触发云函数, 本地机器完全无状态。这套做出来基本就是商业 SaaS 的原型了。

最后说一句, 我不是来卖课的, 我自己就是被这种重复劳动折磨了三年, 才开始琢磨自动化的。如果你也每天被复制粘贴困住, 强烈建议先从 一个脚本 + 一个 Cron 任务 开始, 别一上来就上微服务。跑通了, 再加维度。跑不通, 至少省了买课的钱。

这套东西我前前后后花了大概两周调通, 大部分时间不是写代码, 是踩 MCP 限流、CLI 字段名、n8n 时区这些小坑。代码本身真不难, Python 标准库加 n8n 画布就够。希望这篇能帮你少走几天弯路, 评论区欢迎交流你们的数据自动化踩坑故事。