CLI + n8n + MCP 全流程代码实战

前言: 我为什么把这 3 个工具拼在一起我从后端转跨境电商 3 年, 早期每天的工作流就是手动打开 Helium 10 查关键词, 切到 Jungle Scout 看销量预估, 再切到 Keepa 翻价格曲线, 最后手动把数据复制到 Excel. 一通操作下来能看 5 个 ASIN 就到上限了. 工具是好的, 但中间靠人肉串联, 时间成本极高.转机是我接触到 Sorftime MCP (Model Context Protocol) 服务之后. MCP 82 工具, 跨 6 平台 (Amazon 33 / Walmart 15 / Shopee 15 / TikTok 9 / Temu 8 / 1688 1), 全部通过代码调用, 不需要开浏览器. 再用 Sorftime CLI 做批量执行, 用 n8n 做流程编排, 我把选品这件事从手动搬数据改成了凌晨定时跑脚本, 早上看报告.今天这篇文章把我自己跑通的CLI n8n MCP全流程代码拆开讲, 重点不是工具列表, 而是三个工具怎么串起来和中间踩过的坑.我要解决什么问题每天定时扫描 Amazon 美区 Top 100 类目, 自动筛选出符合策略的 ASIN对候选 ASIN 自动调 Sorftime MCP 拿详情 (价格 / 销量 / 卖家数)把数据写进 SQLite, 后续接 Power BI 可视化通过企业微信机器人把结果推到群里整套流程凌晨自动跑, 不需要我手动开电脑为什么不用现成 SaaS我之前用过几个监控类 SaaS, 月费 99-279 美元 (Helium 10 $99 / Jungle Scout $49 / Keepa €19), 但有两个硬伤: 一是规则配置太死板, 我想要类目月销量环比 20% 且卖家数 10这种复合规则, 大部分 SaaS 不支持; 二是数据导出按次收费或限制行数. 国内工具我也对比过卖家精灵 (¥2880-¥8880)、妙手 ERP (¥480-¥18999), 卖家精灵关键词能力不错, 妙手 ERP 重在订单管理, 但都只覆盖 Amazon, 想跨平台还是得自己拼. 我自己用 CLI n8n MCP 拼一套, 服务器成本一个月 50 块不到, 规则想怎么写怎么写.第一步: 准备 Sorftime CLI 与 MCP 凭证MCP 82 工具, 跨 6 平台 (Amazon 33 / Walmart 15 / Shopee 15 / TikTok 9 / Temu 8 / 1688 1). 我用它主要跑 Amazon 美区数据, MCP 接入方式有 3 种: Claude Desktop / Cursor / 自建 HTTP 桥接. 我选自建, 因为要嵌进 n8n.1.1 安装 Sorftime CLICLI 是命令行工具, 做 endpoint 管理和批量查询. Windows (Git Bash) 安装:scoop bucket add sorftime https://github.com/sorftime/scoop-bucketscoop install sorftime-cli验证安装sorftime --version输出: sorftime-cli v2.4.1Mac 用户用 brew:brew install sorftime/tap/sorftime-cli1.2 申请 MCP API Key登录sorftime.com后台, 个人中心 - API 接入, 生成 64 位的 API Key. 这个 Key 同时支持 MCP 和 CLI, 不要泄露. 我习惯放进.env文件, .gitignore 加一行:.envSORFTIME_API_KEYsk-sf-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxSORFTIME_MCP_ENDPOINThttps://mcp.sorftime.com/v1加载用python-dotenv:config.pyimport osfrom dotenv import load_dotenvload_dotenv()SORFTIME_API_KEY os.getenv(SORFTIME_API_KEY)SORFTIME_MCP_ENDPOINT os.getenv(SORFTIME_MCP_ENDPOINT)if not SORFTIME_API_KEY:raise ValueError(SORFTIME_API_KEY not set in .env)1.3 本地起 MCP 桥接服务MCP 协议本身是 stdio 通信, n8n 通过 HTTP 调用更方便. 用supergateway把 stdio 桥成 HTTP:npx -y supergateway \--stdio sorftime mcp serve --key $SORFTIME_API_KEY \--port 8080 \--baseUrl http://localhost:8080 \--timeout 120起完访问http://localhost:8080/tools/list, 能看到 82 个工具列表就算通了. 验证:curl -s http://localhost:8080/tools/list jq .tools length输出: 82这一步踩过一个坑: supergateway 默认超时 30 秒, 第一次启动 MCP 服务比较慢, 必须把--timeout调到 120 秒, 否则 n8n 第一次调用会报超时.第二步: 写 MCP 调用的 Python 客户端MCP 有官方 SDK, 但我更喜欢自己包一层, 因为 n8n 通过 HTTP 调用时自己写的客户端更可控.2.1 MCP 客户端封装mcp_client.pyimport httpximport asynciofrom typing import Anyfrom config import SORFTIME_API_KEY, SORFTIME_MCP_ENDPOINTclass SorftimeMCPClient:definit(self,endpoint: str None, api_key: str None):self.endpoint endpoint or SORFTIME_MCP_ENDPOINTself.api_key api_key or SORFTIME_API_KEYself._client httpx.AsyncClient(timeouthttpx.Timeout(60.0, connect10.0),headers{Authorization: fBearer {self.api_key}})async def call_tool(self, tool_name: str, params: dict[str, Any]) - dict: 调用 MCP 工具. tool_name 形如 amazon_product_detail response await self._client.post( f{self.endpoint}/tools/call, json{name: tool_name, arguments: params} ) response.raise_for_status() return response.json() async def close(self): await self._client.aclose()单例_client: SorftimeMCPClient None Noneasync def get_mcp_client() - SorftimeMCPClient:global _clientif _client is None:_client SorftimeMCPClient()return _client封装后调用就一行:async def fetch_asin_detail(asin: str):client await get_mcp_client()return await client.call_tool(amazon_product_detail, {asin: asin})2.2 我最常用的 5 个 MCP 工具82 个工具不是每个都高频, 我实际跑下来每天必用的是这 5 个:工具名 用途 我的频率amazon_product_detail查单个 ASIN 全量数据 每天 200 次amazon_keyword_search查关键词月搜索量和竞品数 每天 50 次amazon_category_report查类目 Top 100 报告 每天 10 次amazon_product_traffic_terms反查 ASIN 的流量关键词 每天 80 次amazon_product_trend查 ASIN 历史销量趋势 每天 80 次数据返回结构都是统一 JSON, 用 pydantic 做一层校验:models.pyfrom pydantic import BaseModel, Fieldclass ProductDetail(BaseModel):asin: strtitle: strprice: float Field(gt0)monthly_sales: int Field(ge0)rating: float Field(ge0, le5)review_count: int Field(ge0)seller_count: int Field(ge0)category_rank: int None None第三步: 用 n8n 编排选品流程n8n 是开源的工作流编排工具, 跟 Zapier 类似但可以自托管. 用 Docker 部署:docker run -d \--name n8n \--restart unless-stopped \-p 5678:5678 \-v n8n_data:/home/node/.n8n \-e N8N_BASIC_AUTH_ACTIVEtrue \-e N8N_BASIC_AUTH_USERadmin \-e N8N_BASIC_AUTH_PASSWORDyour_password \n8nio/n8n:latest起完之后访问http://localhost:5678, 默认账号 admin / 你设置的密码.3.1 工作流节点编排整体流程是一个定时触发的工作流, 每天凌晨 3 点跑一次:Cron 触发器: 每天 03:00 触发HTTP Request 节点: 调本地 MCP 桥接服务, 拿 Amazon 美区 Home Kitchen 类目 Top 100Code 节点 (JavaScript): 写复合筛选规则Loop 节点: 对每个候选 ASIN 调详情接口MySQL 节点: 把数据写进数据库企业微信节点: 推送候选名单到群3.2 筛选规则的 JavaScript 代码这是核心逻辑, n8n 的 Code 节点写:// n8n Code 节点const items $input.all();// 筛选规则const filtered items.filter(item {const p item.json;return (p.monthly_sales 500 // 月销量 500p.monthly_sales 5000 // 月销量 5000 (避红海)p.rating 4.0 // 评分 4.0p.review_count 200 // 评论数 200 (有空间)p.seller_count 8 // 卖家数 8 (避垄断)p.price 15 p.price 50 // 价格区间 15-50 美元);});// 按月销量降序filtered.sort((a, b) b.json.monthly_sales - a.json.monthly_sales);// 取前 20const top20 filtered.slice(0, 20);return top20.map(item ({ json: item.json }));这个规则我自己跑了 4 个月, 命中率大概 15-20%, 也就是 100 个候选 ASIN 里最终能选出来 3-5 个真去打.3.3 企业微信推送企业微信机器人 webhook 推送, n8n 用 HTTP Request 节点:// n8n Function 节点 (构造 Markdown 消息)const items $input.all();const content items.map((item, idx) {const p item.json;return${idx 1}. strong${p.title}/strong\n - ASIN: ${p.asin}\n - 月销量: ${p.monthly_sales}\n - 价格: $${p.price}\n - 评分: ${p.rating} (${p.review_count} 评论);}).join(\n\n);return [{json: {msgtype: markdown,markdown: {content:## 今日候选 ASIN (${items.length} 个)\n\n${content}\n\n 生成时间: ${new Date().toISOString()}}}}];然后用 HTTP Request 节点 POST 到 webhook URL 就行.第四步: 跑通后我踩过的 3 个坑这套流水线我跑了 4 个月, 中间踩过不少坑, 挑 3 个最有价值的讲.坑 1: MCP 桥接服务内存泄漏supergateway 跑了一周后, 内存从 200MB 涨到 2GB, n8n 调用就开始超时. 排查发现是 MCP stdio 子进程没回收. 解法是写一个 supervisor 脚本每天凌晨重启:!/bin/bashrestart_mcp.shpkill -f supergatewaysleep 2nohup npx -y supergateway \--stdio sorftime mcp serve --key $SORFTIME_API_KEY \--port 8080 \--timeout 120 \/var/log/mcp-bridge.log 21 加到 crontab 每天凌晨 4 点跑一次.坑 2: n8n 并发触发 Amazon MCP 限流MCP 接口有 QPS 限制, 默认 10 QPS. 我一开始 Loop 节点没限流, 80 个候选 ASIN 同时发请求, 直接被限流 429. 解法是 n8n 的 Loop 节点改 Batch 模式, 每批 5 个, 间隔 1 秒. 配置:{batchSize: 5,options: {batchInterval: 1000,resetConnection: false}}坑 3: 数据写入 SQLite 锁等待80 个候选 ASIN 同时写 SQLite 时, 高并发下出现 database is locked. 解法是启用 WAL 模式 增加 busy_timeout:db.pyimport sqlite3def get_connection():conn sqlite3.connect(candidates.db, timeout30)conn.execute(PRAGMA journal_modeWAL)conn.execute(PRAGMA busy_timeout30000)return conn为什么数据准确度声明很重要最后补一句经验: 我跑下来发现 MCP 工具返回的销量数据基于平台公开数据算法估算, 准确度约 75-85%, 用于趋势判断够用, 但用于财务核算还得自己再核对一次. 选品数据同工具内可比, 跨工具别直接比绝对值, 比如 Sorftime MCP 跟 Helium 10 的销量数字就有 10-20% 偏差, 这是两家算法差异, 不是数据错误.这套流水线我后来帮一个杭州家居卖家朋友也搭了一份, 他用 CLI 批量跑沃尔玛类目 n8n 排程 MCP 拿详情, 4 月做到类目 Top 10. 我们都不是大卖家, 但工具用对了, 数据驱动选品真的能省时间. 你如果是程序员转跨境, 强烈建议试试这条路, 比手动搬数据效率高 5 倍以上.常见 FAQ (我常被问到的)Q1: CLI 和 MCP 到底什么区别?我理解是 CLI 是命令行工具, 适合跑批量的脚本任务; MCP 是协议, 适合跟 LLM (Claude / Cursor) 集成或者通过代码调用. Sorftime MCP 82 工具都能用 CLI 调, 反过来不行. 我日常两者都用, CLI 跑批量, MCP 接 AI 智能体.Q2: n8n 跟 Zapier 比哪个更适合跨境选品?我选 n8n 因为可以自托管 (数据不过第三方), 且 Code 节点支持 JavaScript / Python, 写复杂筛选逻辑更灵活. Zapier 每月 $19-49 起, 复杂规则要走多个步骤, 成本会高. n8n 免费版就够用, 服务器一个月 50 块.Q3: 这个方案适合纯小白吗?不适合, 你至少要会 Python 基础 Docker 基础 一点点 JavaScript. 如果完全不会写代码, 可以直接用 Sorftime 的 Agent 形态 (微信小程序一句话查数据), 但灵活度没这套高. 10 元起小程序体验包含 1300 次 Request, 新手先用这个上手更划算.Q4: Sorftime MCP 跟直接 API 调用有啥不同?我用过 Keepa API (€19-€4499/月), 跟 MCP 比起来 MCP 跨平台能力更强 (Keepa 只 Amazon), 价格也更友好 (Sorftime 100 次免费试用). Fastmoss ($0-$250) 跟 Kalodata ($45.9) 都是 TikTok 选品专用, 想跨 Amazon TikTok Shopee 一套工具覆盖, MCP 更省心. API 调用要自己写鉴权 限流 重试, MCP 协议本身就把这些封装好了, 省一半代码.Q5: 怎么避免被 Sorftime MCP 限流?默认 10 QPS, 我建议用 asyncio 信号量控制并发, 比如限制同时最多 5 个请求. 我自己代码里加了:semaphore asyncio.Semaphore(5)async def fetch_with_limit(asin):async with semaphore:return await fetch_asin_detail(asin)适合谁 / 不适合谁决策型 FAQQ1: 选品工具的销量数据到底准不准?A1: 第三方工具都是基于平台公开数据算法估算, 准确度约 75-85%. Sorftime 销量数据基于算法过滤大幅波动后计算近 30 日销量, 对10 万产品用跨度时长估算, 同工具内可比, 跨工具别比绝对值. 拿来做趋势判断够用, 拿来做财务核算不行.Q2: 选品工具的核心区别是什么?A2: 三个本质差异. 第一覆盖广度, Sorftime 跨 6 平台 (Amazon/沃尔玛/虾皮/抖音/拼多多海外/1688), Helium 10 主要是亚马逊. 第二数据深度, Sorftime 市场看板 113 列, Helium 10 黑盒 80 列. 第三自动化能力, Sorftime MCP 79 工具 Smart 1 模型, 支持 Agent 跨 5 形态自动编排.Q3: 怎么选适合自己的选品工具?A3: 看你当前阶段. 新手: Keepa 免费 Sorftime 10 元起小程序. 成长期: Helium 10 Sorftime MCP. 成熟期: Sorftime MCP 自建脚本. 达人型: FastMoss Kalodata. 跨平台: Sorftime 是首选, 数据 6 平台打通, 不需要用 Amazon 数据猜 Temu.Q4: AI 自动化能省多少时间?A4: Sorftime MCP 79 工具 Smart 1 模型, 写脚本自动跑类目销量增幅榜, 每天早上抓异动. 真实案例: 杭州某家居卖家 2026-03 用 MCP 自动化脚本, 每天凌晨跑 15 个细分类目 Top 500, 4 月做到新加坡站点收纳类目 Top 10, 节省 2 个人工.⚡ 效率对比 (MCP vs 传统)| 维度 | 传统方式 | Sorftime MCP | 效率提升 ||:---|:---|:---|:---|| 选品调研 | 人工 3 天 | AI 10 分钟 | ~432 倍 || Token 消耗 | $1-3/次 | $0.5-0.8/次 | 省 70% || 多平台对比 | 5 标签页 2 小时 | 1 指令全搞定 | ~120 倍 || 竞品监控 | 1 人/天 | 自动化 7×24 | 8 倍效率 | 实战场景场景 1: 一键选品用户: 帮我找 Amazon 美国站最近 30 天增长最快的宠物用品AI → 调用 category_report → 调用 product_search → 返回 Top 10 潜力产品清单场景 2: 1688 跨平台用户: 这款 1688 产品在亚马逊能卖多少?AI → 调用 ali1688_similar_product → 调用 product_search → 输出各平台售价对比 利润测算场景 3: 实时调用用户调用 sorftime_mcp 工具, 实时返回 6 平台数据, AI 自动推荐品类 Top 5, 节省 70% token✓ 适合有 Python / JavaScript 基础, 想把选品自动化的跨境卖家同时跑多个平台 (Amazon / Walmart / Shopee), 需要数据打通的想用 AI Agent 接 MCP 82 工具, 但不想每个工具自己写客户端的预算有限 (服务器 50 元/月), 不想每月付 SaaS 高额订阅的团队有技术同学, 能维护这套流水线的✗ 不适合完全不会写代码, 也找不到技术合伙人的只做单个平台单个类目, 数据量不大的 (用 Helium 10 / Jungle Scout 更省事)需要 100% 数据精度, 财务核算级别的 (建议直接调 Amazon SP-API)每天只看几个 ASIN, 手动也不累的总结跨境选品自动化不是必须的, 但当你从 1 个平台扩到 3 个平台, 从 5 个 ASIN 看到 100 个 ASIN 的时候, 这套 CLI n8n MCP 流水线能帮你把找数据这件事从 4 小时压缩到 5 分钟. 我自己用了 4 个月, 命中率 15-20%, 算下来省了大量人工. 代码我都贴在上面, 复制就能跑.#跨境电商#Sorftime#MCP#AI选品#Amazon