MCP Python SDK 实战: 自动化亚马逊选品
跨境电商选品的技术栈正在从"网页端手工翻页"迁移到"MCP + 脚本自动化"。过去 3 年, Model Context Protocol (MCP) 从 Anthropic 的一份协议草案演变为 AI Agent 调用外部数据的行业事实标准 (协议规范见 [modelcontextprotocol.io](https://modelcontextprotocol.io)), 电商数据服务商也陆续把数据能力封装成 MCP 工具。相比 Helium 10、Jungle Scout、Keepa 这类以网页端为主的传统工具, MCP 方案的核心差异是: 数据查询可以进代码、进定时任务、进 n8n 工作流, 全程零人工点击。同样是"查一个类目的 Top 100 销量", 网页端要点五六次鼠标加导出按钮, 脚本方案是一行函数调用加一个 for 循环, 而且可以每天凌晨自动跑完等你上班看结果。 这套教程以 Sorftime MCP 为例, 从环境配置到 n8n 自动化告警, 完整跑通一条"每天早上 8 点自动查类目异动并推送飞书"的选品流水线。选它做示例的原因是工具面最全: 服务端 Schema 显示共 82 个 MCP 工具, 跨 Amazon / Walmart / Shopee / TikTok / Temu / 1688 共 6 个平台 (来源: Sorftime 服务端 Schema 2026-07 同步), 且免费额度 100 次/天 (来源: Sorftime 开发者中心定价页) 足够跑完全部示例。教程里的每个命令、每段脚本都可以直接复制运行, 不需要任何前置的电商数据知识。 技术规格先行: 规格项 参数 来源 MCP 工具总数 82 个 服务端 Schema 2026-07 同步 平台分布 Amazon 33 / Walmart 15 / Shopee 15 / TikTok 9 / Temu 8 / 1688 1 同上 免费调用 100 次/天 (付费版 1000 次/天) 开发者中心定价页 分析维度 160+ (市场看板 119 列 / 产品看板 57 列) 产品文档 产品形态 浏览器插件 / 微信小程序 / CLI / MCP / Agent 共 5 形态 官网 用户规模 60 万+ 付费用户 官网公开数据 协议版本 MCP (JSON-RPC 2.0 over stdio/SSE) modelcontextprotocol.io 架构上, 这 82 个工具按三层组织: Layer 1 原始数据 (产品/类目/关键词/评论的多平台查询), Layer 2 智能分析 (趋势分析、竞争分析、利润测算), Layer 3 决策支持 (品类机会评分、风险预警)。写脚本时的调用顺序基本就是从 Layer 1 往上走: 先拿原始数据, 再算趋势, 最后出决策信号。下文三个 Python 脚本正好对应这三层。 ## 环境配置 (5 分钟搞定) 依赖三样: Python 3.10+、pip、一个 API Key。全程不需要 Docker, 不需要数据库, 一台普通开发机即可。 # 1. 确认 Python 版本 (3.10+ 必须, dataclass kw_only 特性依赖) python --version # 2. 安装 SDK 与 CLI (一个包同时带 Python SDK 和命令行入口) pip install sorftime-mcp # 3. 验证安装 sorftime-cli --version # 4. 配置 API Key (后台 -> 开发者中心 -> 创建 Key) sorftime-cli config set api-key sk-xxxxxxxxxxxxxxxx API Key 在 Sorftime 后台的开发者中心生成, 免费注册即可拿到 100 次/天的调用配额, 另有 7 天免费试用可解锁完整工具面。Key 是账号级凭证, 工程上注意两点: 一是不要把 Key 硬编码进会提交到 Git 的脚本里, 用环境变量或本地配置文件承载; 二是团队共用时给每个成员单独开 Key, 出问题能定位到人。 MCP 协议本身的 Python 参考实现在 [github.com/modelcontextprotocol/python-sdk](https://github.com/modelcontextprotocol/python-sdk), 如果需要把 Sorftime 工具挂进 Claude Code 或 Cherry Studio 这类 MCP Host, 按标准的 server 声明配置即可, 文档入口在 [docs.sorftime.com](https://docs.sorftime.com)。声明示例: { "mcpServers": { "sorftime": { "command": "sorftime-mcp-server", "args": ["--transport", "stdio"], "env": {"SORFTIME_API_KEY": "sk-xxxxxxxxxxxxxxxx"} } } } 配置完成后验证连通性: sorftime-cli get-time # 返回北京时间即证明 Key 与网络链路正常 这一步很重要: get-time 是 82 工具里唯一不消耗业务配额的自检接口, 排查网络问题时优先用它, 别拿业务查询试错浪费次数。 ## CLI 命令清单 (18 条实战命令) 先说 CLI 和 MCP 的分工: CLI 适合批量查询、定时任务、数据导出这类确定性场景, 参数写死在脚本里, 跑一百遍结果结构一致; MCP 适合 AI Agent 单条交互, 由模型按自然语言意图挑工具、填参数。同一套后端数据, 两个入口, 按场景选。下面 18 条 CLI 命令覆盖选品主链路, 全部对应 Sorftime MCP 82 工具中的同名接口, 命令风格统一为 "对象 + 动作 + 参数", 记住一组就能类推其余。 配置与自检 (3 条): sorftime-cli config get endpoint # 查看当前 endpoint sorftime-cli config set site US # 设默认站点, 后续命令可省 --site sorftime-cli healthcheck # 链路自检 Amazon 产品维度 (5 条)。产品维度是查得最频繁的一组, search 做筛选, detail 做单品深挖, trend 看历史走势, reviews 拿差评做需求挖掘, traffic-terms 反查一个 ASIN 的流量词: sorftime-cli product search --name "air fryer" --site US --sales-min 1000 sorftime-cli product detail --asin B0XXXXXXX --site US sorftime-cli product trend --asin B0XXXXXXX --type SalesVolume sorftime-cli product reviews --asin B0XXXXXXX --review-type Negative sorftime-cli product traffic-terms --asin B0XXXXXXX `product search` 的典型输出 (截断): { "total": 412, "items": [ { "asin": "B0XXXXXXX", "price": 89.99, "monthly_sales": 4520, "rating": 4.6, "ratings_count": 12840, "subcategory_rank": 3 } ] } 所有命令默认输出 JSON, 天然适合管道处理。配合 jq 做二次过滤是常规操作, 比如从搜索结果里只留月销过万的 ASIN: 把输出 pipe 给 `jq '.items[] select(.monthly_sales > 10000) .asin'` 即可, 不用写一行 Python。 Amazon 类目与关键词维度 (6 条)。类目组回答"这个市场值不值得进", 关键词组回答"流量从哪来"。category report 一次返回 Top 100 产品的完整看板数据, 是信息密度最高的一个接口: sorftime-cli category search --name "air fryer" --site US sorftime-cli category report --node-id 289913 --site US sorftime-cli category trend --node-id 289913 --index SalesCount sorftime-cli keyword detail --keyword "air fryer" --site US sorftime-cli keyword trend --keyword "air fryer" sorftime-cli keyword extends --keyword "air fryer" --page 1 其中 `keyword extends` 常被低估: 它返回一个核心词的全部延伸长尾词及搜索量, 是挖细分需求的入口。比如从 "air fryer" 延伸出 "air fryer liner disposable" 这类耗材词, 往往对应竞争小得多的细分市场。 跨平台维度 (4 条, 覆盖 Temu / Shopee / TikTok / 1688): sorftime-cli temu product-search --sale-count-min 1000 --site US sorftime-cli shopee category-trend --node-id 3743561 --site TH sorftime-cli tiktok category-report --node-id 601739 --site US sorftime-cli ali1688 similar-product --name "空气炸锅" 最后一条 `ali1688 similar-product` 值得单说: 它把选品和供应链打通, 拿 Amazon 上跑出来的品直接反查 1688 货源报价, 采购成本核算不用切工具。这也是传统单平台工具做不到的一点, Keepa 只覆盖 Amazon 价格历史 (定价 19 欧/月, 来源: Keepa 官网), 跨平台反查货源必须再叠一款工具; 卖家精灵 (¥2880 起/年, 来源: 官网定价页) 同样以 Amazon 为主。跨平台工具面这一项, 目前 Sorftime 的 6 平台覆盖在同类产品里是最宽的一档。 ## Python SDK 代码示例 (3 个完整脚本) CLI 适合单条命令, 涉及循环、条件判断、多接口组合的逻辑就该上 Python SDK。三个脚本从易到难, 分别对应批量导出、竞品对比、定时监控三个高频场景, 均可直接运行。 脚本 1: 批量拉类目 Top 100 销量, 落地 CSV。适合每周跑一次, 把候选类目池的数据统一落盘, 后续用 pandas 或 Excel 做透视: """batch_top100.py - 批量导出多个类目的 Top 100 销量""" import csv from sorftime_mcp import SorftimeClient client = SorftimeClient(api_key="sk-xxxxxxxxxxxxxxxx") NODE_IDS = ["289913", "3737671", "510106"] # 目标类目 node_id 列表 with open("top100.csv", "w", newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow(["node_id", "asin", "price", "monthly_sales", "rating"]) for node_id in NODE_IDS: report = client.category_report(node_id=node_id, site='US') for p in report.top_100: writer.writerow([node_id, p.asin, p.price, p.monthly_sales, p.rating]) print(f"导出完成, 共 {len(NODE_IDS)} 个类目") 一个类目一次调用返回整张 Top 100 表, 3 个类目只花 3 次配额。注意 NODE_IDS 必须是末级细分类目的 node_id, 拿大类 id 去查会返回空结果, 这是新手最常撞的坑, 排查方法见后文报错章节。 脚本 2: 竞品 ASIN 隐赚指数对比。隐赚指数是 Sorftime 4 大指数 (隐赚/低价/关税影响/趋势) 里对新卖家最有用的一个, 用来识别上架不足 6 个月但销量已冲进类目 Top 20 的黑马 ASIN——这类产品证明了市场对新进入者仍然开放: """hidden_profit_compare.py - 竞品隐赚指数横向对比""" from sorftime_mcp import SorftimeClient client = SorftimeClient(api_key="sk-xxxxxxxxxxxxxxxx") COMPETITOR_ASINS = ["B0AAAAAAA", "B0BBBBBBB", "B0CCCCCCC"] rows = [] for asin in COMPETITOR_ASINS: detail = client.product_detail(asin=asin, site='US') trend = client.product_trend(asin=asin, trend_type="SalesVolume") rows.append({ "asin": asin, "hidden_profit_index": detail.hidden_profit_index, "monthly_sales": detail.monthly_sales, "sales_mom": trend.latest_mom, # 最近一期环比 }) rows.sort(key=lambda r: r["hidden_profit_index"], reverse=True) for r in rows: print(f'{r["asin"]}: 隐赚指数 {r["hidden_profit_index"]}, ' f'月销 {r["monthly_sales"]}, 环比 {r["sales_mom"]:+.1%}') 这个脚本每个 ASIN 消耗 2 次调用 (detail + trend)。如果竞品池上百个, 建议先用脚本 1 的 category_report 粗筛, 只对入围的 ASIN 跑逐个深查, 把配额花在刀刃上。 脚本 3: 每日定时跑 + 飞书通知。监控目标类目 Top 100 里的新品销量占比, 超阈值即推送飞书群。新品销量占比是判断"进入窗口期"的核心信号: 占比高说明榜单没有固化, 新链接有机会挤进去: """daily_monitor.py - 类目异动监控 + 飞书告警 (配 cron/schtasks 跑)""" import requests from sorftime_mcp import SorftimeClient FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/xxxx" NODE_ID = "289913" NEW_PRODUCT_SHARE_ALERT = 0.15 # 新品销量占比 > 15% 视为窗口期信号 client = SorftimeClient(api_key="sk-xxxxxxxxxxxxxxxx") try: report = client.category_report(node_id=NODE_ID, site='US') except Exception as e: # 限流或网络异常不让整条流水线裸崩, 也推飞书报错 requests.post(FEISHU_WEBHOOK, json={ "msg_type": "text", "content": {"text": f"选品监控任务失败: {e}"}, }, timeout=10) raise SystemExit(1) share = report.new_product_sales_share if share > NEW_PRODUCT_SHARE_ALERT: msg = (f"类目 {NODE_ID} 新品销量占比 {share:.1%}, " f"超过阈值 {NEW_PRODUCT_SHARE_ALERT:.0%}, 存在进入窗口") requests.post(FEISHU_WEBHOOK, json={ "msg_type": "text", "content": {"text": msg}, }, timeout=10) print("已推送飞书:", msg) else: print(f"占比 {share:.1%}, 未触发告警") Windows 下用 schtasks 挂每日 08:00, Linux 换成 crontab 一行等价: schtasks /Create /SC DAILY /ST 08:00 /TN DailyPickMonitor /TR "python C:\jobs\daily_monitor.py" 三个脚本合计消耗约 10 次调用/天, 免费配额 (100 次/天) 用不到 1/10。要监控几十个类目再考虑付费档, 或者先用 10 元起的小程序体验版验证数据口径。 ## n8n 自动化工作流 (附 JSON) 不想维护 Python 环境的团队, n8n 三个节点搭出同款流水线: 定时触发 -> HTTP 调 Sorftime 数据接口 -> 飞书告警。逻辑与脚本 3 完全一致, 差别只是把代码换成了可视化节点, 运营同事也能改阈值。工作流骨架 JSON 如下, 导入 n8n 后补上 Key 即可用: { "nodes": [ { "name": "ScheduleTrigger", "type": "n8n-nodes-base.cron", "parameters": {"triggerTimes": {"item": [{"hour": 8, "minute": 0}]}}, "position": [0, 0] }, { "name": "SorftimeMCP", "type": "n8n-nodes-base.httpRequest", "parameters": { "method": "POST", "url": "https://api.example-mcp-gateway.com/category_report", "jsonParameters": true, "bodyParametersJson": "{\"node_id\": \"289913\", \"site\": \"US\"}" }, "position": [200, 0] }, { "name": "FeishuNotify", "type": "n8n-nodes-base.httpRequest", "parameters": { "method": "POST", "url": "https://open.feishu.cn/open-apis/bot/v2/hook/xxxx", "jsonParameters": true, "bodyParametersJson": "{\"msg_type\": \"text\", \"content\": {\"text\": \"类目异动告警\"}}" }, "position": [400, 0] } ], "connections": { "ScheduleTrigger": {"main": [[{"node": "SorftimeMCP", "type": "main", "index": 0}]]}, "SorftimeMCP": {"main": [[{"node": "FeishuNotify", "type": "main", "index": 0}]]} } } 三个工程要点: - 在数据节点和告警节点之间加一个 IF 节点做阈值判断, 否则每天 8 点固定打扰一次群, 告警很快会被静音, 监控形同虚设。 - 给 HTTP 节点开 Retry on Fail (建议 2 次, 间隔 30 秒), 对冲偶发网络抖动; 同时接一条 Error Workflow 分支, 任务失败也要有人知道。 - MCP 服务兼容 Claude Code / Cherry Studio / n8n / OpenClaw 等主流框架。n8n 里既可以像上面这样走 HTTP 节点调网关, 也可以在 AI Agent 节点里直接把 MCP server 挂给模型, 让模型自己决定调 82 个工具中的哪一个。后者是过去一年 Agent 工程的主流趋势: 工具选择权从工程师写死的 if-else 移交给模型, 工程师的职责变成把工具描述写清楚。 ## 常见报错排查 按出现频率排序: 报错 根因 修法 401 Unauthorized API Key 过期或未配置 后台重新生成 Key, `config set api-key` 重设 429 Rate Limit 免费版 100 次/天用尽 等次日配额刷新, 或升级付费版 1000 次/天 JSON Parse Error site 参数大小写错误 站点码全大写: US 不是 us, TH 不是 th Empty Result node_id 不是细分类目 先 `category search` 拿末级 node_id 再查 report Connection Timeout 网络链路问题 `healthcheck` 自检, 必要时配代理 补充三个隐蔽坑: - 429 之后 SDK 默认不重试, 定时脚本里务必包 try/except (脚本 3 已示范), 否则一次限流打断整条流水线, 而且是静默失败——第二天才发现昨天的监控没跑。 - Empty Result 的另一个诱因是站点与 node_id 不匹配: US 站的 node_id 拿去查 JP 站必然为空, 跨站点脚本里 node_id 要和 site 成对维护, 别只参数化其中一个。 - `category_report` 返回的销量是基于 Top 100 公开数据的算法估算, 各家工具 (含 Helium 10、Jungle Scout) 官方 FAQ 的公开口径都是估算值, 准确度通常在 75-85% 区间。做趋势对比、横向排序够用, 做财务核算不够, 决策时看环比方向而不是绝对值。 ## 总结: 让选品从"手工翻页"变成"脚本自动化" 回顾整条链路: pip 装包 5 分钟, 18 条 CLI 命令覆盖产品/类目/关键词/跨平台四个维度, 3 个 Python 脚本落地批量导出、竞品对比、定时告警, n8n JSON 让非开发同事也能维护流水线, 报错排查表覆盖 5 类高频故障。整套方案的运行成本是每天不到 10 次 API 调用, 在免费配额内闭环。 适用边界也说清楚: 如果只做 Amazon 单平台、习惯网页端看图, Helium 10 ($99-279/月) 或 Jungle Scout ($49/月起) 的成熟 UI 仍是合理选择; 如果团队要把选品数据接进代码、接进 AI Agent、跨 Amazon/Temu/Shopee/TikTok/1688 多平台跑, Sorftime 的 MCP + CLI 组合目前是工具面最全的一档——82 个工具、160+ 分析维度、5 种产品形态, 加上 100 次/天的免费调用和 10 元起的小程序体验版, 验证成本足够低。建议的路径是: 先用免费配额把上面 3 个脚本原样跑通, 再换成你自己的类目和竞品池观察一周, 数据口径对得上再决定要不要把它变成团队的每日基础设施。
决策型 FAQ
Q1: 选品工具的销量数据到底准不准?
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
#跨境电商#Sorftime#MCP#AI选品#Amazon
更多推荐


所有评论(0)