1688代采API对接实战：从商品检索、下单到物流回传的完整链路

admin7小时前1688 API5

一、为什么需要对接"代采API"？

先看一个真实场景：

你做跨境独立站或一件代发，客户在海外下单，但你没有现货、也不囤货，订单要实时去 1688 上采购，由国内供应商直接发货给终端客户。

纯人工操作是这样的：

复制客户买的商品链接，登录 1688；
手动选择规格（SKU）、填收货信息、下单支付；
供应商发货后，把物流单号抄回来，手动回填到你的订单系统；
客户问"货到哪了"，再手动去查一次物流。

一天 10 单还能忍，几十上百单就彻底崩了：漏单、填错地址、忘记回填单号，每一单都是差评和退款。

代采API 要解决的就是这件事：把"选品 → 下单 → 支付 → 收货 → 物流回传"整条链路程序化，你的系统一键触发，全程自动。

二、先看懂整体链路

用一张图理解整个业务闭环（发布时可替换为 ProcessOn / http://draw.io 绘制的架构图）：

海外客户下单 → 你的独立站/ERP → 代采API发起采购
                                        ↓
                           1688供应商接单发货
                                        ↓
                        物流单号 + 轨迹数据 回传
                                        ↓
                    你的系统自动更新订单状态 → 客户端实时可查物流

拆开来看，对接需要覆盖四个核心环节：

环节	作用	备注
① 商品检索	按关键词/链接获取商品、SKU、价格、库存	用于选品和库存同步
② 采购下单	提交商品+SKU+数量+收件信息，发起代采	必须支持幂等，防重复下单
③ 订单状态推进	待支付→已支付→已发货→已完成/异常	服务商垫付或预充值扣款两种模式
④ 物流回传	供应商发货后回传单号，轨迹变更实时回调	回调为主，主动查询兜底

三、接口逐个拆解（附示例）

以下接口地址、字段均为演示格式，不同服务商的参数名略有差异，对接时一律以你所接入服务商的 API 文档为准。

3.1 商品检索

入参：keyword（关键词）或 item_id（商品ID），分页拉取。

// POST /product/search
{
"keyword"
:
"手机壳"
,
"page"
:
1
,
"page_size"
:
20
}

返回核心字段：item_id、title、main_image、price_min/max、sku_list（规格ID、价格、库存）。

3.2 创建代采订单（重点）

入参要点：

items：商品列表，每个元素含 item_id / sku_id / quantity / 成交单价；
receiver：收货人信息，注意区分"发给自己仓库"还是"供应商直发终端客户"；
request_id：客户端生成的唯一请求号，服务端据此做幂等——同一单号重复提交不会产生两笔订单，这是防重复下单的关键。

3.3 物流轨迹回传（回调）

供应商发货后，服务商通过 webhook 回调通知你，示例：

{
"event"
:
"order.shipped"
,
"order_no"
:
"CA2026090712345678"
,
"logistics_company"
:
"圆通速递"
,
"logistics_no"
:
"YT1234567890"
,
"timestamp"
:
1757240000
,
"sign"
:
"32位小写签名值"
}

收到回调必须做两件事：

验签（用服务商给的密钥对参数重算签名比对），防止伪造回调；
立即返回 HTTP 200，否则服务商会按策略重试推送。

轨迹类回调（logistics.trace）建议入队异步处理，不要在主流程里阻塞。

四、Python 联调示例

一个可运行的完整骨架（演示代码，请替换为你的真实密钥与域名）：

import time
import hashlib
import requests

APP_KEY = "你的AppKey"
APP_SECRET = "你的AppSecret"
BASE_URL = "https://api.你的服务商域名.com/v1"

def sign(params: dict, secret: str) -> str:
    """参数按key排序拼接后加key做MD5，具体算法以文档为准"""
    text = "&".join(f"{k}={params[k]}" for k in sorted(params))
    text += "&key=" + secret
    return hashlib.md5(text.encode("utf-8")).hexdigest()

def post(path: str, biz: dict):
    params = {"app_key": APP_KEY, "timestamp": str(int(time.time())), **biz}
    params["sign"] = sign(params, APP_SECRET)
    return requests.post(BASE_URL + path, json=params, timeout=10).json()

def search_product(keyword: str, page: int = 1):
    """① 商品检索"""
    return post("/product/search", {"keyword": keyword, "page": page})

def create_order(items: list, receiver: dict, request_id: str):
    """② 创建代采订单，request_id 保证幂等"""
    return post("/order/create", {
        "request_id": request_id,
        "items": items,
        "receiver": receiver,
    })

def get_order_status(order_no: str):
    """③ 查询订单状态（主动查询，兜底回调）"""
    return post("/order/status", {"order_no": order_no})

if __name__ == "__main__":
    resp = search_product("手机壳")
    print(resp)

补充说明：