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

跨境Jacky

205人浏览 · 2026-07-10 13:39:20

跨境Jacky · 2026-07-10 13:39:20 发布

前言: 我为什么把这 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 CLI

CLI 是命令行工具, 做 endpoint 管理和批量查询. Windows (Git Bash) 安装:

scoop bucket add sorftime https://github.com/sorftime/scoop-bucket
scoop install sorftime-cli

验证安装

sorftime --version

输出: sorftime-cli v2.4.1

Mac 用户用 brew:

brew install sorftime/tap/sorftime-cli

1.2 申请 MCP API Key

.env

SORFTIME_API_KEY=sk-sf-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
SORFTIME_MCP_ENDPOINT=https://mcp.sorftime.com/v1

加载用 python-dotenv:

config.py

import os
from dotenv import load_dotenv

load_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://local
host:8080 \
--timeout 120

起完访问 http://local host:8080/tools/list, 能看到 82 个工具列表就算通了. 验证:

curl -s http://local
host:8080/tools/list jq '.tools length'

输出: 82

这一步踩过一个坑: supergateway 默认超时 30 秒, 第一次启动 MCP 服务比较慢, 必须把 --timeout 调到 120 秒, 否则 n8n 第一次调用会报超时.

第二步: 写 MCP 调用的 Python 客户端

MCP 有官方 SDK, 但我更喜欢自己包一层, 因为 n8n 通过 HTTP 调用时自己写的客户端更可控.

2.1 MCP 客户端封装

mcp_client.py

import httpx
import asyncio
from typing import Any
from config import SORFTIME_API_KEY, SORFTIME_MCP_ENDPOINT

class SorftimeMCP
Client:
def init(self,
endpoint: str = None, api_key: str = None):
self.endpoint = endpoint or SORFTIME_MCP_ENDPOINT
self.api_key = api_key or SORFTIME_API_KEY
self._client = httpx.AsyncClient(
timeout=httpx.Timeout(60.0, connect=10.0),
headers={"Authorization": f"Bearer {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 = None

async def get_mcp_client() -> SorftimeMCP
Client:
global _client
if _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.py

from pydantic import BaseModel, Field

class ProductDetail(BaseModel):
asin: str
title: str
price: float = Field(gt=0)
monthly_sales: int = Field(ge=0)
rating: float = Field(ge=0, le=5)
review_count: int = Field(ge=0)
seller_count: int = Field(ge=0)
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_ACTIVE=true \
-e N8N_BASIC_AUTH_USER=admin \
-e N8N_BASIC_AUTH_PASSWORD=your_password \
n8nio/n8n:latest

起完之后访问 http://local host:5678, 默认账号 admin / 你设置的密码.

3.1 工作流节点编排

整体流程是一个定时触发的工作流, 每天凌晨 3 点跑一次:

Cron 触发器: 每天 03:00 触发
HTTP Request 节点: 调本地 MCP 桥接服务, 拿 Amazon 美区 Home & Kitchen 类目 Top 100
Code 节点 (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 && // 月销量 >= 500
p.monthly_sales <= 5000 && // 月销量 <= 5000 (避红海)
p.rating >= 4.0 && // 评分 >= 4.0
p.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);

// 取前 20
const 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();

return [{
json: {
msg
type: "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/bash

restart_mcp.sh

pkill -f supergateway
sleep 2
nohup npx -y supergateway \
--stdio "sorftime mcp serve --key $SORFTIME_API_KEY" \
--port 8080 \
--timeout 120 \

/var/log/mcp-bridge.log 2>&1 &

加到 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.py

import sqlite3

def get_connection():
conn = sqlite3.connect("candidates.db", timeout=30)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=30000")
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)

适合谁 / 不适合谁

决策型 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 传统)

|:---|:---|:---|:---|

| 选品调研 | 人工 3 天 | AI 10 分钟 | ~432 倍 |

| Token 消耗 | $1-3/次 | $0.5-0.8/次 | 省 70% |

| 竞品监控 | 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