亚马逊数据采集 API：从接入到实操（SP-API 为主，合规高效）

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

作为全球领先的跨境电商平台，亚马逊的数据采集接口已从早期的 MWS API 升级至全新的 SP-API（Selling Partner API）。自2023年起，MWS API 停止对新用户开放，并逐步退出维护阶段。当前，官方唯一推荐且合规的数据接入方式为 SP-API，该接口支持获取商品信息、销量统计、用户评价、库存状态等关键数据，全面覆盖北美、欧洲、日本等全球各主要站点。

本文将系统性地介绍亚马逊 SP-API 的接入流程、核心接口功能解析、代码实现示例以及常见问题规避策略，兼顾初学者与进阶开发者的实际需求。

一、基础认知：SP-API 与 MWS API 的关键差异（明确版本选择）

对比维度	MWS API（旧版）	SP-API（新版）
当前状态	不再接受新用户接入，逐步停用中	官方主推，持续更新维护
认证机制	开发者ID + 卖家授权Token	IAM角色授权 + OAuth 2.0协议
安全性	较低，存在明文Token传输风险	更高，采用AWS IAM加密授权机制
接口覆盖范围	仅限基础的商品、订单和库存数据	涵盖MWS全部功能，新增广告、评论及合规类数据接口
适配站点	部分区域支持	支持所有亚马逊全球站点
接入复杂度	相对简单	稍高，需配置AWS IAM角色

结论：无论你是新注册用户还是仍在使用 MWS 的老用户，都应优先选用 SP-API。MWS 用户需尽快完成迁移，因亚马逊已关闭新接口申请通道。

二、SP-API 接入前的准备工作（四大核心步骤）

1. 所需资质与账号准备

• 亚马逊卖家账号：建议使用专业卖家账户，个人卖家在部分接口上权限受限；
• AWS 账号：用于创建和管理 IAM 角色，免费注册即可；
• 亚马逊开发者账号：用于应用注册与API调用授权。

资质要求：需具备企业或个体工商户身份，且主体信息必须与卖家账号一致；同时完成开发者账号与卖家账号之间的关联验证（包括企业与税务信息核验）。

2. 注册与配置流程详解

（1）注册 AWS 账号并创建 IAM 角色

1. 完成 AWS 账号注册并进行实名认证；
2. 
3. 登录 AWS 控制台，搜索「IAM」服务，进入「角色」→「创建角色」；
4. 选择信任实体为「AWS 服务」，关联服务选择「Selling Partner API」或直接搜索关键词「sellingpartnerapi」；
5. 附加所需权限策略——可选择全权限策略，或根据业务需要配置细粒度权限；

6. AmazonSellingPartnerAPIFullAccess

7. 也可按需分配特定模块权限，如仅访问订单或商品数据；

8. AmazonSellingPartnerAPIProductAccess

9. 角色创建成功后，请记录下生成的「Role ARN」；

10. arn:aws:iam::账号ID:role/角色名

11. 此ARN将在后续开发者账号配置时使用。

（2）注册亚马逊开发者账号并绑定卖家账号

1. 访问亚马逊开发者平台完成注册，需验证邮箱与手机号；
2. 进入「Selling Partner API」→「Develop Apps」→「Create a new app」；
3. 填写应用基本信息（名称、描述、隐私政策URL），并在「AWS IAM Role ARN」字段中填入上一步获取的 Role ARN；
4. 提交审核后，系统会向卖家邮箱发送授权请求邮件；
5. 点击邮件中的链接完成卖家账号与开发者账号的绑定操作；
6. 审核通过后，平台将提供以下核心凭证：

7. LWA Client ID

8. LWA Client Secret

9. 以上两项是实现 OAuth 2.0 授权的关键参数。

（3）获取访问令牌（Access Token）

每次调用 SP-API 都需通过 OAuth 2.0 获取临时访问令牌（有效期为1小时），具体流程如下：

1. 构造授权请求 URL（替换相应占位符）；

2. https://sellercentral.amazon.com/apps/authorize/consent?client_id=LWA_CLIENT_ID&scope=sellingpartnerapi::migration&response_type=code

3. 使用目标卖家账号登录该链接，同意授权后页面跳转至预设的「回调URL」，并携带一个 code 参数；

4. code

5. 该授权码有效期仅为5分钟，需立即处理；
6. 使用该 code 换取 Access Token 和 Refresh Token（其中 Refresh Token 有效期长达一年，可用于刷新失效的 Access Token）；
7. 请求地址：

8. https://api.amazon.com/auth/o2/token

9. 发送 POST 请求，参数如下：

10. grant_type=authorization_code&code=授权码&client_id=LWA_CLIENT_ID&client_secret=LWA_CLIENT_SECRET

11. 成功响应示例如下（请妥善保存返回结果中的 access_token 和 refresh_token）：

12. refresh_token

13. {
      "access_token": "Atza|...",  // 访问令牌（1小时有效）
      "refresh_token": "Atzr|...", // 刷新令牌（1年有效）
      "token_type": "bearer",
      "expires_in": 3600
    }

3. 开发工具与依赖环境搭建

• 推荐开发语言：Python（生态完善，有成熟SDK支持）、Java、Node.js 等均可；
• Python 核心依赖库：

• boto3

  —— AWS 官方 SDK，用于发起 SP-API 请求；

• requests-oauthlib

  —— 处理 OAuth 2.0 认证流程；

• python-dotenv

  —— 管理配置文件，避免敏感信息硬编码。

安装所需依赖包：

bash
pip install boto3 requests-oauthlib python-dotenv
    

三、SP-API 核心接口分类解析（按业务场景划分）

SP-API 将接口按功能划分为多个模块，以下是电商运营中最常用的高频数据采集接口：

业务场景	核心接口	可采集数据内容	调用前提条件
商品基础信息	catalog-items-v2022-04-01	商品标题、SKU、ASIN、所属类目、品牌、规格参数、主图链接等	只需完成卖家账号关联，无需额外权限
商品价格与库存	listings-items-v2021-08-01	当前售价、促销价、可用库存数量、配送时效等	卖家账号需对该商品具有销售权限
销量与订单数据	orders-v0	订单编号、买家信息、支付金额、发货状态、物流单号等	需为专业卖家账号，并申请 orders 权限
用户评价数据	reviews-v2021-01-01	客户评分、评论内容、评论时间、是否带图、有用性投票等	需单独申请评论数据访问权限

商品评价信息采集包括评价内容、评分、追加评论、晒图情况以及评价时间等数据。

需申请相应权限方可获取，且仅支持采集自身店铺商品的用户评价。对于竞品评价数据，必须通过其他合规途径另行获取。

reviews

针对竞品监控（ASIN），可采集的信息涵盖目标商品的价格变动、配送方式及卖家相关信息。

该功能需要单独申请接口权限，并提供具体的竞品 ASIN 列表才能调用。

product-pricing-v2022-05-01

product_pricing

市场趋势相关数据如商品销量走势、销售额变化和转化率等指标，仅限专业卖家账号访问。

此类数据同样仅支持查看与自身商品相关的统计结果，无法查询非自有商品的市场表现。

sales-performance-v2022-04-01

重要说明事项

接口版本要求： SP-API 接口具有明确的版本号（例如

v2022-04-01

），建议始终使用最新版接口；亚马逊会定期停用旧版本，请及时升级以确保服务正常。

请求频率限制（限流规则）： 不同接口有不同的调用频率上限。例如，

catalog-items

接口默认每秒最多允许 5 次请求，超出将返回

429 Too Many Requests

错误码。

区域端点配置： 各大亚马逊站点对应不同的 API 端点地址，跨区域调用会导致失败。主要站点对应的接口端点如下：

站点区域	接口端点
北美（美国、加拿大、墨西哥）	https://sellingpartnerapi-na.amazon.com
欧洲（英国、德国、法国、意大利、西班牙）	https://sellingpartnerapi-eu.amazon.com
日本	https://sellingpartnerapi-jp.amazon.com
澳大利亚	https://sellingpartnerapi-au.amazon.com

四、Python 实践操作：SP-API 调用核心示例

以下演示两个典型场景的实际代码实现：采集商品基础信息（使用 Catalog Items API）与获取自身商品的用户评价（使用 Reviews API）。代码包含 Token 自动刷新、请求签名处理和响应解析完整流程。

1. 配置文件设置（.env）——安全存储凭证

创建名为

.env

的环境变量文件，避免将敏感信息硬编码在源码中。

env
# LWA凭证
LWA_CLIENT_ID=你的LWA_CLIENT_ID
LWA_CLIENT_SECRET=你的LWA_CLIENT_SECRET
REFRESH_TOKEN=你的REFRESH_TOKEN

# SP-API端点（北美站点为例）
SP_API_ENDPOINT=https://sellingpartnerapi-na.amazon.com

# AWS IAM凭证（可选，boto3会自动读取~/.aws/credentials）
AWS_ACCESS_KEY_ID=你的AWS_ACCESS_KEY_ID
AWS_SECRET_ACCESS_KEY=你的AWS_SECRET_ACCESS_KEY
  

2. 封装通用 SP-API 调用类（集成 Token 刷新与签名逻辑）

构建基础类用于统一处理身份验证、签名生成和 HTTP 请求发送。

python
运行
import os
import time
import requests
from dotenv import load_dotenv
from oauthlib.oauth2 import BackendApplicationClient
from requests_oauthlib import OAuth2Session

# 加载环境变量
load_dotenv()

class AmazonSPAPIClient:
    def __init__(self):
        self.lwa_client_id = os.getenv("LWA_CLIENT_ID")
        self.lwa_client_secret = os.getenv("LWA_CLIENT_SECRET")
        self.refresh_token = os.getenv("REFRESH_TOKEN")
        self.sp_api_endpoint = os.getenv("SP_API_ENDPOINT")
        self.access_token = None
        self.token_expire_time = 0  # 记录Token过期时间

    def _refresh_access_token(self):
        """刷新Access Token（过期前自动刷新）"""
        current_time = time.time()
        # 若Token未获取或剩余有效期<300秒，刷新Token
        if not self.access_token or (self.token_expire_time - current_time) < 300:
            url = "https://api.amazon.com/auth/o2/token"
            data = {
                "grant_type": "refresh_token",
                "refresh_token": self.refresh_token,
                "client_id": self.lwa_client_id,
                "client_secret": self.lwa_client_secret
            }
            response = requests.post(url, data=data, timeout=10)
            if response.status_code == 200:
                token_data = response.json()
                self.access_token = token_data["access_token"]
                self.token_expire_time = current_time + token_data["expires_in"]
                print("Token刷新成功，有效期至：", time.ctime(self.token_expire_time))
            else:
                raise Exception(f"Token刷新失败：{response.status_code} - {response.text}")
        return self.access_token

    def call_api(self, method, path, params=None, data=None):
        """通用API调用方法（处理Token、签名、请求）"""
        access_token = self._refresh_access_token()
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json",
            "x-amz-access-token": access_token  # 部分接口需额外携带该Header
        }
        url = f"{self.sp_api_endpoint}{path}"
        try:
            if method.upper() == "GET":
                response = requests.get(url, headers=headers, params=params, timeout=15)
            elif method.upper() == "POST":
                response = requests.post(url, headers=headers, json=data, timeout=15)
            else:
                raise ValueError("仅支持GET/POST请求")
            
            # 处理响应（限流时重试）
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 5))
                print(f"触发限流，{retry_after}秒后重试...")
                time.sleep(retry_after)
                return self.call_api(method, path, params, data)
            
            response.raise_for_status()  # 抛出HTTP错误（4xx/5xx）
            return response.json()
        except Exception as e:
            raise Exception(f"API调用失败：{str(e)}")
  

3. 场景一：获取商品基本信息（基于 ASIN 查询）

调用

catalog-items-v2022-04-01

接口，传入 ASIN 可获取商品标题、品牌名称、所属类目等关键属性。

python
运行
def get_product_details(client, asin_list):
    """
    采集商品基础信息
    :param client: SP-API客户端实例
    :param asin_list: ASIN列表（如["B07VGFG35C", "B08X7Z3X7Q"]）
    :return: 商品详情列表
    """
    path = "/catalog/2022-04-01/items"
    params = {
        "asin": ",".join(asin_list),
        "marketplaceIds": "ATVPDKIKX0DER",  # 北美市场ID（美国），其他市场ID见下文
        "includedData": "attributes,identifiers,images"  # 需返回的数据类型
    }
    response = client.call_api("GET", path, params=params)
    product_details = []
    
    # 解析响应数据
    for item in response.get("items", []):
        asin = item["asin"]
        attributes = item.get("attributes", {})
        images = item.get("images", {}).get("primary", [])
        
        product_details.append({
            "asin": asin,
            "title": attributes.get("title", [{"value": "未知"}])[0]["value"],
            "brand": attributes.get("brand", [{"value": "未知"}])[0]["value"],
            "category": attributes.get("productType", [{"value": "未知"}])[0]["value"],
            "main_image_url": images[0]["url"] if images else "无",
            "price": attributes.get("listPrice", [{"value": 0}])[0]["value"]  # 标价
        })
    return product_details

# 调用示例
if __name__ == "__main__":
    client = AmazonSPAPIClient()
    # 目标ASIN列表（美国站点商品）
    asin_list = ["B07VGFG35C", "B08X7Z3X7Q"]
    product_data = get_product_details(client, asin_list)
    for product in product_data:
        print(f"ASIN：{product['asin']}")
        print(f"标题：{product['title']}")
        print(f"品牌：{product['brand']}")
        print(f"主图URL：{product['main_image_url']}")
        print("-" * 50)
  

4. 场景二：提取自身商品的用户评价

通过

reviews-v2021-01-01

接口拉取本店商品的客户评价数据（不支持直接采集竞品评价，需另行合规申请权限）。

python
运行
def get_product_reviews(client, asin, page_size=10):
    """
    采集自身商品评价
    :param client: SP-API客户端实例
    :param asin: 商品ASIN
    :param page_size: 每页评价数（最大100）
    :return: 评价列表
    """
    path = "/reviews/2021-01-01/reviews"
    params = {
        "asin": asin,
        "marketplaceIds": "ATVPDKIKX0DER",
        "pageSize": page_size,
        "sortOrder": "NEWEST"  # 排序：NEWEST（最新）、HELPFUL（最有帮助）
    }
    response = client.call_api("GET", path, params=params)
    reviews = []
    
    # 解析评价数据
    for review in response.get("reviews", []):
        reviews.append({
            "review_id": review["reviewId"],
            "asin": review["asin"],
            "rating": review["rating"],  # 评分（1-5）
            "review_text": review.get("reviewText", ""),
            "reviewer_name": review.get("reviewerName", "匿名用户"),
            "review_date": review["reviewDate"],
            "verified_purchase": review.get("verifiedPurchase", False)  # 是否为真实购买
        })
    return reviews

# 调用示例
if __name__ == "__main__":
    client = AmazonSPAPIClient()
    asin = "B07VGFG35C"
    reviews = get_product_reviews(client, asin, page_size=20)
    print(f"ASIN {asin} 的评价（共{len(reviews)}条）：")
    for idx, review in enumerate(reviews, 1):
        print(f"\n{idx}. 评分：{review['rating']}星 | 日期：{review['review_date']}")
        print(f"用户：{review['reviewer_name']} | 真实购买：{review['verified_purchase']}")
        print(f"评价内容：{review['review_text']}")
  

5. 场景三：获取竞品价格信息（基于 ASIN 查询）

使用

product-pricing-v2022-05-01

接口，可查询指定竞品 ASIN 当前售价、物流选项等公开信息。

python
运行
def get_competitor_price(client, asin_list):
    """
    采集竞品价格信息
    :param client: SP-API客户端实例
    :param asin_list: 竞品ASIN列表
    :return: 价格列表
    """
    path = "/products/pricing/v2022-05-01/offer-prices"
    params = {
        "asin": ",".join(asin_list),
        "marketplaceIds": "ATVPDKIKX0DER",
        "itemCondition": "New"  # 商品状态：New（新品）、Used（二手）
    }
    response = client.call_api("GET", path, params=params)
    price_data = []
    
    for item in response.get("offerPrices", []):
        asin = item["asin"]
        # 提取最低售价（新品）
        lowest_price = item.get("lowestPrices", [{}])[0]
        price_data.append({
            "asin": asin,
            "currency": lowest_price.get("currency", "USD"),
            "amount": lowest_price.get("amount", 0),
            "shipping": lowest_price.get("shipping", {}).get("amount", 0),
            "total_price": lowest_price.get("amount", 0) + lowest_price.get("shipping", {}).get("amount", 0),
            "offer_count": item.get("offerCount", 0)  # 在售卖家数量
        })
    return price_data

# 调用示例
if __name__ == "__main__":
    client = AmazonSPAPIClient()
    competitor_asins = ["B08X7Z3X7Q", "B07VGFG35C"]
    price_data = get_competitor_price(client, competitor_asins)
    for price in price_data:
        print(f"ASIN：{price['asin']}")
        print(f"最低售价：{price['currency']} {price['amount']}")
        print(f"运费：{price['currency']} {price['shipping']}")
        print(f"总价：{price['currency']} {price['total_price']}")
        print(f"在售卖家数：{price['offer_count']}")
        print("-" * 50)
  

五、常见问题避坑指南（SP-API 使用注意事项）

1. 权限异常处理

当接口返回

403 Forbidden

时，应检查以下几点：

• 确认 IAM 角色已正确配置所需权限（如价格采集需具备

  product_pricing

  权限）；
• 核实卖家账户与开发者账户已完成绑定，未关联将导致权限不足；
• 部分高风险接口（如

  orders

  ）需提交人工审核申请，须在亚马逊开发者平台完成审批流程。

2. 应对限流机制与实现重试策略

SP-API 对各类接口设有严格限流规则（如

catalog-items

接口默认每秒 5 次），超限后会返回

429

。

应对方案包括：

• 在代码中加入自动重试机制（参考示例中的

  call_api

  方法），依据响应头中的

  Retry-After

  字段进行延时重发；
• 主动控制请求频率（建议控制在每秒 1–2 次以内），防止触发限流；
• 批量查询时优先使用支持多参数的接口（如

  asin

  参数支持逗号分隔多个 ASIN），减少总请求数量。

3. 区域与 marketplaceIds 匹配关系

不同站点需使用对应的 marketplaceIds，否则可能引发

400 Bad Request

错误。

站点	marketplaceIds
美国	ATVPDKIKX0DER
加拿大	A2EUQ1WTGCTBG2
英国	A1F83G8C2ARO7P
德国	A1PA6795UKMFR9
日本	A1VC38T7YXB528

marketplaceIds

4. Token 生命周期管理

• Access Token 有效期为 1 小时，需在过期前自动刷新（示例中已实现）；
• Refresh Token 有效期长达 1 年，务必妥善保存，一旦丢失需重新授权；
• 避免每次请求都刷新 Token，仅在即将过期时执行刷新操作，降低风险。

5. 合规性要求

• 严禁采集买家个人敏感信息（如手机号、收货地址），所有评价内容需做匿名化处理；
• 所获数据仅可用于自身业务运营或市场分析，不得用于恶意竞争行为（如恶意降价、虚假宣传等）；
• 严格遵守《亚马逊开发者协议》，禁止滥用 API 进行大规模无关 ASIN 数据抓取。

六、进阶优化策略：提升数据采集效率与系统稳定性

1. 批量采集性能优化

• 充分利用接口提供的批量参数功能（如

  asin

  支持一次传入多个 ASIN），显著减少请求数量；
• 处理分页数据时，当接口返回

  nextToken

  标识，需循环调用直至获取全部结果（适用于评价列表等长数据）。

2. 增强异常处理能力

• 捕获常见异常类型（如网络超时、JSON 解析失败等），并引入日志记录模块（推荐使用

  logging

  ）便于排查问题；
• 对关键操作（如订单数据同步）增加本地备份或数据库持久化机制，防止意外数据丢失。

3. 数据存储与深度分析

• 将采集到的数据存入结构化数据库（如 MySQL 或 PostgreSQL），便于后续检索、聚合与可视化展示；
• 结合数据分析方法（如价格波动趋势分析、用户评价情感分析），深入挖掘商业洞察，辅助决策制定。

总结

亚马逊 SP-API 是官方唯一认可的数据采集通道，具备数据精准、服务稳定、合法合规三大核心优势，适用于长期监控商品动态、销售表现和用户反馈。

成功接入的关键步骤包括：完成 AWS IAM 权限配置、完成开发者账号授权绑定、实现 Token 的自动化管理，并根据实际业务需求选择合适的接口进行调用。

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏0 回帖

关键词：数据采集 API 亚马逊 Application performance

相关内容：亚马逊采集合规