Amazon Selling Partner API(SP-API)はAmazon出品者向けの統一APIだが、「LWA(Login with Amazon)認証」「リージョン別エンドポイント」「注文/カタログ/在庫/レポート/財務の5大リソース」「ページネーションNextToken」といった独自概念が多く、初見では実装に詰まるポイントが多い。本記事ではrenueが社内EC分析基盤に組み込んでいるAmazonSPClientの実装コードをもとに、SP-API統合の本番品質パターンを解説する。
SP-APIの全体像: 何が独自で何が標準か
SP-APIはRESTful設計だが、以下の点で一般的なSaaS APIと異なる。
- 3リージョン分離: NA(北米)/EU(欧州)/FE(極東)でエンドポイントが完全分離
- LWA認証: Login with AmazonのOAuth 2.0 refresh_token flow — アクセストークンは1時間で失効
- Marketplace ID必須: ほぼ全リクエストで出品先マーケットプレイスIDを指定
- NextToken方式: ページネーションがpage番号ではなく不透明な継続トークン
- x-amz-access-tokenヘッダ: Authorizationヘッダではなく独自ヘッダ名
- レポートは非同期: 大量データはCreate→Poll→Downloadの3段階
日本の出品者が知っておくべき値
| 項目 | 値 |
|---|---|
| 日本のMarketplace ID | A1VC38T7YXB528 |
| 極東(FE)エンドポイント | https://sellingpartnerapi-fe.amazon.com |
| LWAトークンエンドポイント | https://api.amazon.com/auth/o2/token |
| アクセストークン有効期限 | 3600秒(1時間) |
| 1ページの最大件数(Orders) | 100 |
| 1ページの最大件数(Catalog) | 20 |
レイヤー1: リージョン別エンドポイント設計
SP-APIで最初にやるべきはリージョン別エンドポイントを定数化することだ。一度実装すれば欧米展開時の追加コストがほぼゼロになる。
SP_API_ENDPOINTS = {
"na": "https://sellingpartnerapi-na.amazon.com", # 北米
"eu": "https://sellingpartnerapi-eu.amazon.com", # ヨーロッパ
"fe": "https://sellingpartnerapi-fe.amazon.com", # 極東(日本含む)
}
TOKEN_URL = "https://api.amazon.com/auth/o2/token"
class AmazonSPClient:
def __init__(
self,
access_token: Optional[str] = None,
refresh_token: Optional[str] = None,
client_id: Optional[str] = None,
client_secret: Optional[str] = None,
marketplace_id: Optional[str] = None,
region: str = "fe", # 日本がデフォルト
):
self.marketplace_id = marketplace_id or "A1VC38T7YXB528" # 日本
self.region = region
self.base_url = SP_API_ENDPOINTS.get(region, SP_API_ENDPOINTS["fe"])
region="fe"をデフォルトにしているのは日本市場中心の社内利用を想定しているため。グローバル出品を扱う場合はregionパラメータを必須化する設計に変える。
レイヤー2: LWAリフレッシュトークンフロー
SP-APIの認証はOAuth 2.0 refresh_token flowそのもの。既存のLWA refresh_tokenを使って毎回アクセストークンを取得する。
def refresh_amazon_access_token(
client_id: str,
client_secret: str,
refresh_token: str,
) -> Dict[str, Any]:
payload = {
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": client_id,
"client_secret": client_secret,
}
try:
response = requests.post(TOKEN_URL, data=payload, timeout=30)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
logger.error("Failed to refresh Amazon access token: %s", e)
raise AmazonSPAPIError(
status_code=401,
message=f"Failed to refresh access token: {str(e)}",
)
初期化時の自動フォールバック
クライアント初期化時にaccess_tokenが未指定なら自動的にrefreshする設計にすると、呼び出し側がシンプルになる。
self.access_token = access_token or getattr(env_setting, "AMAZON_SP_ACCESS_TOKEN", None)
if not self.access_token:
if self.refresh_token and self.client_id and self.client_secret:
self._refresh_access_token()
else:
raise ValueError("Amazon SP access token or refresh credentials are required.")
開発時は短命アクセストークンを直接渡す、本番はrefresh_tokenのみ環境変数に入れる、という柔軟な運用ができる。
レイヤー3: x-amz-access-token ヘッダ
SP-APIはAuthorization ヘッダではなく x-amz-access-tokenという独自ヘッダにトークンを載せる。この仕様を知らないとひたすら401で詰まる。
def _headers(self) -> Dict[str, str]:
return {
"x-amz-access-token": self.access_token,
"Content-Type": "application/json",
}
通常のREST APIでありがちなAuthorization: Bearer {token}を書いてしまうと認証が通らない。ヘッダ名だけはSP-API固有と覚えておく。
レイヤー4: 統一requestメソッド + エラー階層
Orders/Catalog/Inventory/Reports/Financesを横断するrequestメソッドを一つ用意し、各API関数はその薄いラッパーにするのが設計の要点。
class AmazonSPAPIError(Exception):
def __init__(self, status_code: int, message: str, payload=None):
super().__init__(message)
self.status_code = status_code
self.message = message
self.payload = payload or {}
def request(
self,
method: str,
path: str,
params=None,
payload=None,
timeout: int = 30,
) -> Dict[str, Any]:
url = f"{self.base_url}{path}"
try:
response = requests.request(
method.upper(), url,
headers=self._headers(),
params=params,
json=payload,
timeout=timeout,
)
except requests.RequestException as e:
logger.error("Amazon SP-API request failed: %s", e)
raise AmazonSPAPIError(status_code=502, message="Amazon SP-API request failed.")
if not response.ok:
try:
payload_data = response.json()
except ValueError:
payload_data = {"raw": response.text}
error_message = "Amazon SP-API error"
if "errors" in payload_data:
errors = payload_data["errors"]
if errors:
error_message = errors[0].get("message", error_message)
raise AmazonSPAPIError(
status_code=response.status_code,
message=error_message,
payload=payload_data,
)
try:
return response.json()
except ValueError:
return {"raw": response.text}
errorsフィールドの抽出が重要
SP-APIのエラーレスポンスは{"errors": [{"code": "...", "message": "..."}]}形式。単にresponse.textを投げるとデバッグできないので、errors[0].message を必ず抽出する。これだけで本番でのトラブルシューティング時間が10分の1になる。
レイヤー5: 5大リソースの薄いラッパー
Orders API
def get_orders(
self,
created_after: Optional[str] = None,
order_statuses: Optional[List[str]] = None,
max_results: int = 100,
next_token: Optional[str] = None,
) -> Dict[str, Any]:
params = {
"MarketplaceIds": self.marketplace_id,
"MaxResultsPerPage": min(max_results, 100),
}
if created_after:
params["CreatedAfter"] = created_after
if order_statuses:
params["OrderStatuses"] = ",".join(order_statuses)
if next_token:
params["NextToken"] = next_token
return self.request("GET", "/orders/v0/orders", params=params)
OrderStatusesはカンマ区切り(Unshipped,PartiallyShipped)で渡すのがAmazon仕様。["Unshipped", "PartiallyShipped"]をそのまま渡すと配列エラーになる。
Catalog API
def search_catalog_items(
self,
keywords: Optional[str] = None,
identifiers: Optional[List[str]] = None,
identifier_type: str = "ASIN",
max_results: int = 20,
) -> Dict[str, Any]:
params = {
"marketplaceIds": self.marketplace_id,
"pageSize": min(max_results, 20), # Catalogは最大20
}
if keywords:
params["keywords"] = keywords
if identifiers:
params["identifiers"] = ",".join(identifiers)
params["identifiersType"] = identifier_type
return self.request("GET", "/catalog/2022-04-01/items", params=params)
CatalogはpageSize最大20という厳しい制限がある。Orders(100)と混同すると400エラーになる。
Inventory API (FBA)
def get_inventory_summaries(
self,
seller_skus: Optional[List[str]] = None,
next_token: Optional[str] = None,
) -> Dict[str, Any]:
params = {
"granularityType": "Marketplace",
"granularityId": self.marketplace_id,
"marketplaceIds": self.marketplace_id,
}
if seller_skus:
params["sellerSkus"] = ",".join(seller_skus)
if next_token:
params["nextToken"] = next_token
return self.request("GET", "/fba/inventory/v1/summaries", params=params)
Inventory APIはgranularityType/granularityIdという独自概念がある。基本はMarketplace粒度で問題ない。
Reports API(非同期フロー)
Reports APIは大量データ向けの非同期方式で、Create Report → Poll Status → Download Documentの3段階。
def create_report(
self,
report_type: str, # 例: GET_FLAT_FILE_ALL_ORDERS_DATA_BY_ORDER_DATE_GENERAL
data_start_time: Optional[str] = None,
data_end_time: Optional[str] = None,
) -> Dict[str, Any]:
payload = {
"reportType": report_type,
"marketplaceIds": [self.marketplace_id],
}
if data_start_time:
payload["dataStartTime"] = data_start_time
if data_end_time:
payload["dataEndTime"] = data_end_time
return self.request("POST", "/reports/2021-06-30/reports", payload=payload)
def get_report(self, report_id: str) -> Dict[str, Any]:
return self.request("GET", f"/reports/2021-06-30/reports/{report_id}")
def get_report_document(self, report_document_id: str) -> Dict[str, Any]:
return self.request("GET", f"/reports/2021-06-30/documents/{report_document_id}")
非同期フローなので呼び出し側はCeleryタスクやバックグラウンドジョブで組むのが基本。待機ループをサーバー同期コードで書くのはアンチパターン。
NextTokenページネーション: 全件取得パターン
SP-APIはNextToken方式のため、全件取得はwhileループで「NextTokenがnilになるまで」回す。
def get_all_orders(client, created_after: str) -> List[Dict]:
all_orders = []
next_token = None
while True:
if next_token:
response = client.get_orders(next_token=next_token)
else:
response = client.get_orders(created_after=created_after)
payload = response.get("payload", {})
orders = payload.get("Orders", [])
all_orders.extend(orders)
next_token = payload.get("NextToken")
if not next_token:
break
return all_orders
重要なのは初回リクエストではfilterを渡し、2回目以降はNextTokenのみ渡すパターン。Amazonは「NextTokenと他のフィルタを併用するな」という暗黙の仕様があり、両方渡すと400エラーになる。
運用Tips
レート制限(トークンバケット)
SP-APIはエンドポイントごとに「バーストレート/リフィルレート」のトークンバケット方式でレート制限される。OrdersはBurst 20, Refill 0.0167/s(1分あたり1)のように非常に厳しい。レポート優先で取得し、リアルタイム性が必要な部分だけOrdersを叩く設計が推奨。
アクセストークンキャッシュ
アクセストークン有効期限は1時間。毎リクエストで取得するとLWA側にもレート制限がかかる。クライアントインスタンスに_token_expires_atを持たせてキャッシュする改良を入れると本番安定度が上がる。
エラーリトライ
SP-APIは429/500/502/503/504は指数バックオフでリトライすべきとAmazonが公式に推奨している。本記事の実装は基本形なので、本番ではtenacity等でリトライラッパーを被せるとよい。
まとめ
- リージョン別エンドポイント定数化(NA/EU/FE)で多国展開に備える
- LWA refresh_token flowで1時間アクセストークンを自動取得
- x-amz-access-tokenヘッダ(Authorizationヘッダではない)
- 統一requestメソッド + errors[0].message抽出でデバッグ性を確保
- Orders 100 / Catalog 20 のページサイズ違いに注意
- Reports APIは非同期3段階でバックグラウンドジョブ必須
- NextToken併用時はフィルタを渡さない(暗黙仕様)
よくある質問
Q. python-amazon-sp-apiなどのサードパーティSDKは使わない?
SDKは便利だが、更新頻度・バグ・依存関係の重さから、本番運用では自社実装が選ばれることが多い。本記事で示した実装は約360行で全主要APIをカバーしており、保守しやすい。
Q. AWS Signature v4署名は不要?
2023年の仕様変更でAWS Signature v4署名は不要になり、LWAアクセストークンのみで認証できるようになった。古い記事では署名ロジックが書かれているが現在は不要。
Q. グローバル出品対応はどうする?
NA/EU/FEごとにAmazonSPClientインスタンスを作り、各リージョンで並列取得するのが基本。marketplace_idも各国ごとに異なるので環境変数で管理する。