亚马逊 API 接口深度分析及 Python 实现
亚马逊作为全球最大的电商平台，提供了一套全面的 API 体系，用于对接其电商生态。目前主要分为Selling Partner API (SP-API)（新一代）和Marketplace Web Service (MWS)（传统，逐步被替代）。以下重点分析 SP-API 的特性，并实现 Python 调用框架。
一、亚马逊 SP-API 核心特性分析

1. 接口体系与功能域
   SP-API 是亚马逊为卖家提供的现代化 API，覆盖跨境电商全流程，核心功能域包括：
   商品管理：创建 / 更新商品、查询商品详情（如catalogItems接口组）；
   订单管理：获取订单、更新订单状态、处理退款（如orders接口组）；
   库存管理：查询库存、更新库存数量（如inventory接口组）；
   定价管理：设置商品价格、获取竞品价格（如pricing接口组）；
   物流管理：创建物流计划、跟踪物流信息（如fulfillmentInbound接口组）；
   销售分析：获取销售报表、流量数据（如reports接口组）。
2. 认证与安全机制
3. 接口规范与地域特性
   协议与格式：强制 HTTPS，请求 / 响应均为 JSON；
   地域端点：不同地区有独立 API 端点（如北美https://sellingpartnerapi-na.amazon.com、欧洲https://sellingpartnerapi-eu.amazon.com）；
   版本控制：接口版本在 URL 中指定（如/orders/v0/orders）；
   节流机制：按 API 组设置调用限额（如订单接口默认每秒 2 次），超限返回429错误。
4. 错误处理机制
   SP-API 的错误响应包含三层信息：
   HTTP 状态码：4xx客户端错误（如401认证失败）、5xx服务端错误；
   错误类型：ErrorType字段标识错误类别（如InvalidInput、QuotaExceeded）；
   详细信息：Message字段描述具体错误原因，Code字段提供错误码（如InvalidOrderId）。
   二、Python 脚本实现：亚马逊 SP-API 调用框架
   以下实现 SP-API 的通用调用框架，包含 LWA 认证、AWS 签名、请求处理，并以 “获取订单列表” 和 “查询商品详情” 为例演示。
5. 完整脚本实现
   python
   运行
   import requests
   import json
   import time
   import logging
   from datetime import datetime, timedelta
   from requests.exceptions import RequestException
   from boto3.auth import SigV4Auth
   from botocore.awsrequest import AWSRequest
   from typing import Dict, Optional

配置日志

logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)

class AmazonSPAPI:
def init(
self,
client_id: str,
client_secret: str,
aws_access_key: str,
aws_secret_key: str,
region: str = "us-east-1", # 默认为北美地区
marketplace_id: str = "ATVPDKIKX0DER" # 默认为美国市场
):
"""
初始化亚马逊SP-API客户端
:param client_id: LWA Client ID
:param client_secret: LWA Client Secret
:param aws_access_key: AWS Access Key
:param aws_secret_key: AWS Secret Key
:param region: 地区（us-east-1:北美, eu-west-1:欧洲, us-west-2:亚太）
:param marketplace_id: 市场ID（美国:ATVPDKIKX0DER, 英国:A1F83G8C2ARO7P等）
"""
self.client_id = client_id
self.client_secret = client_secret
self.aws_access_key = aws_access_key
self.aws_secret_key = aws_secret_key
self.region = region
self.marketplace_id = marketplace_id

    # 根据地区选择API端点
    self.endpoints = {
        "us-east-1": "https://sellingpartnerapi-na.amazon.com",
        "eu-west-1": "https://sellingpartnerapi-eu.amazon.com",
        "us-west-2": "https://sellingpartnerapi-fe.amazon.com"
    }
    self.base_url = self.endpoints.get(region, self.endpoints["us-east-1"])

    self.access_token = None
    self.token_expiry = None  # access_token过期时间

def _get_access_token(self) -> Optional[str]:
    """通过LWA获取access_token（有效期1小时）"""
    # 检查token是否有效，有效则复用
    if self.access_token and self.token_expiry and datetime.utcnow() < self.token_expiry:
        return self.access_token

    # 调用LWA接口获取新token
    url = "https://api.amazon.com/auth/o2/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": self.client_id,
        "client_secret": self.client_secret,
        "scope": "sellingpartnerapi::notifications"
    }

    try:
        response = requests.post(url, data=payload, timeout=10)
        response.raise_for_status()
        result = response.json()

        self.access_token = result.get("access_token")
        expires_in = result.get("expires_in", 3600)  # 默认3600秒
        self.token_expiry = datetime.utcnow() + timedelta(seconds=expires_in - 60)  # 提前60秒过期
        logging.info("获取access_token成功")
        return self.access_token

    except RequestException as e:
        logging.error(f"获取access_token失败：{str(e)}")
        return None

def _sign_request(self, method: str, path: str, params: Dict = None, data: Dict = None) -> Dict:
    """使用AWS Signature V4签名请求"""
    url = f"{self.base_url}{path}"

    # 构建AWS请求对象
    request = AWSRequest(
        method=method,
        url=url,
        params=params,
        data=json.dumps(data) if data else None
    )

    # 添加必要的请求头
    request.headers.update({
        "Host": self.base_url.replace("https://", ""),
        "Content-Type": "application/json",
        "x-amz-access-token": self._get_access_token(),
        "x-amz-date": datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
    })

    # 签名请求
    sigv4 = SigV4Auth(self.aws_access_key, self.aws_secret_key, "sellingpartnerapi", self.region)
    sigv4.add_auth(request)

    return {
        "url": url,
        "headers": dict(request.headers),
        "params": params,
        "data": json.dumps(data) if data else None
    }

def call(self, method: str, path: str, params: Dict = None, data: Dict = None) -> Optional[Dict]:
    """
    通用API调用方法
    :param method: HTTP方法（GET/POST/PUT等）
    :param path: API路径（如/orders/v0/orders）
    :param params: 查询参数
    :param data: 请求体数据
    :return: 接口返回的业务数据或None
    """
    # 1. 获取签名后的请求信息
    signed_request = self._sign_request(method, path, params, data)
    if not signed_request:
        return None

    # 2. 发送请求
    try:
        response = requests.request(
            method=method,
            url=signed_request["url"],
            headers=signed_request["headers"],
            params=signed_request["params"],
            data=signed_request["data"],
            timeout=30
        )

        # 3. 处理响应
        logging.info(f"API调用：{method} {path}，状态码：{response.status_code}")

        # 解析响应（部分成功响应可能为空）
        response_data = response.json() if response.text else {}

        # 4. 处理错误
        if response.status_code >= 400:
            error_msg = response_data.get("errors", [{}])[0].get("message", "未知错误")
            error_code = response_data.get("errors", [{}])[0].get("code", "UnknownError")
            logging.error(f"API错误：{error_msg}（错误码：{error_code}，状态码：{response.status_code}）")
            return None

        return response_data

    except RequestException as e:
        logging.error(f"请求异常：{str(e)}，路径：{path}")
        return None
    except json.JSONDecodeError:
        logging.error(f"响应格式错误：{response.text}，路径：{path}")
        return None

def get_orders(self, start_date: datetime, end_date: datetime, page_size: int = 100) -> Optional[Dict]:
    """
    获取订单列表（接口：/orders/v0/orders）
    :param start_date: 订单创建开始时间（UTC）
    :param end_date: 订单创建结束时间（UTC）
    :param page_size: 每页条数（最大100）
    :return: 订单列表数据
    """
    path = "/orders/v0/orders"
    # 时间格式需为ISO 8601（UTC）
    params = {
        "CreatedAfter": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "CreatedBefore": end_date.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "MarketplaceIds": self.marketplace_id,
        "MaxResultsPerPage": page_size
    }
    return self.call("GET", path, params=params)

def get_catalog_item(self, asin: str) -> Optional[Dict]:
    """
    查询商品详情（接口：/catalog/v0/items/{asin}）
    :param asin: 商品ASIN（亚马逊标准识别号）
    :return: 商品详情数据
    """
    path = f"/catalog/v0/items/{asin}"
    params = {
        "MarketplaceId": self.marketplace_id,
        "IncludeData": "attributes,salesRank,identifiers"  # 需要返回的数据类型
    }
    return self.call("GET", path, params=params)

示例调用

if name == "main":

# 替换为你的实际参数（从亚马逊开发者平台获取）
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"
AWS_ACCESS_KEY = "your_aws_access_key"
AWS_SECRET_KEY = "your_aws_secret_key"
REGION = "us-east-1"  # 北美地区
MARKETPLACE_ID = "ATVPDKIKX0DER"  # 美国市场

# 初始化API客户端
amazon_api = AmazonSPAPI(
    client_id=CLIENT_ID,
    client_secret=CLIENT_SECRET,
    aws_access_key=AWS_ACCESS_KEY,
    aws_secret_key=AWS_SECRET_KEY,
    region=REGION,
    marketplace_id=MARKETPLACE_ID
)

# 1. 获取最近1天的订单（UTC时间）
end_date = datetime.utcnow()
start_date = end_date - timedelta(days=1)
orders = amazon_api.get_orders(start_date, end_date, page_size=10)
if orders:
    print(f"订单总数：{len(orders.get('Orders', []))}")
    for order in orders.get('Orders', [])[:3]:  # 打印前3条订单
        print(f"订单号：{order.get('AmazonOrderId')}，金额：{order.get('OrderTotal', {}).get('Amount')} {order.get('OrderTotal', {}).get('CurrencyCode')}")

# 2. 查询商品详情（替换为实际ASIN）
asin = "B07VGRJDFY"  # 示例ASIN（亚马逊自营商品）
catalog_item = amazon_api.get_catalog_item(asin)
if catalog_item:
    item_attrs = catalog_item.get('Items', [{}])[0].get('Attributes', {})
    print(f"\n商品标题：{item_attrs.get('Title')}")
    print(f"品牌：{item_attrs.get('Brand')}")
    print(f"售价：{item_attrs.get('ListPrice', {}).get('Amount')} {item_attrs.get('ListPrice', {}).get('CurrencyCode')}")

三、关键技术点解析

1. 双层认证的实现
   LWA 认证：通过client_credentials模式获取access_token，注意 token 有效期为 1 小时，需在过期前自动刷新；
   AWS 签名：使用boto3库的SigV4Auth实现签名，需正确设置服务名（sellingpartnerapi）和地区，否则签名无效。
2. 地域与市场 ID 映射
   亚马逊全球有多个市场（如美国、英国、日本等），需正确匹配region和marketplace_id：
   地区 区域代码 市场 ID（示例）
   北美 us-east-1 美国：ATVPDKIKX0DER
   欧洲 eu-west-1 英国：A1F83G8C2ARO7P
   亚太 us-west-2 日本：A1VC38T7YXB528
3. 时间格式处理
   SP-API 要求所有时间参数使用UTC 时区的 ISO 8601 格式（如2023-08-01T12:00:00Z），需注意：
   避免使用本地时间（如北京时间需转换为 UTC）；
   时间戳需精确到秒，末尾加Z表示 UTC。
4. 节流与重试策略
   当收到429错误（请求超限）时，需根据Retry-After响应头的建议时间重试；
   实现指数退避重试（如 1 秒→2 秒→4 秒），避免频繁请求加重服务器负担。
   四、扩展与实战建议
   分页处理：SP-API 采用分页返回大量数据（如订单列表），需通过NextToken参数实现翻页；
   批量操作：使用reports接口组创建批量报表，高效获取大量数据（如批量商品信息）；
   异常监控：集成 Sentry 等工具监控 API 错误，重点关注5xx服务端错误和401认证错误；
   合规性：严格遵守亚马逊 API 使用规范，禁止爬取未授权数据，避免账号被封禁。
   通过上述框架，可实现与亚马逊 SP-API 的稳定对接，适用于跨境电商 ERP 系统、多平台管理工具等场景。实际开发中需参考亚马逊 SP-API 官方文档，根据具体业务需求扩展接口封装。