Build an auto listing pipeline that doesn't just publish products — it publishes them at the right price, with the right keywords, informed by current market data.
TL;DR
Amazon auto listing via SP-API is approachable for developers who haven't done it before. The hard parts are:
- Product Type field schemas vary wildly — always fetch the schema before building payloads
- Use PATCH (not PUT) for price updates — it's faster, targeted, and lower risk
- Static auto listing is a start, not a finish — wire a real-time data source in from day one
What We're Building
A complete Amazon auto listing system with:
- SP-API OAuth authentication with automatic token refresh
- Listings Items API for real-time single-SKU listing creation
- Feeds API for async batch listing (hundreds of SKUs)
- Pangolinfo Scrape API integration for live competitor pricing
- 30-minute repricing cycle that updates prices based on live market data
Prerequisites
- Amazon Professional Seller account
- SP-API app registration in Developer Central (with
ListingsandFeedsrole permissions) - LWA Refresh Token from the authorization workflow
- Pangolinfo API key (trial at tool.pangolinfo.com)
- Python 3.9+
Install dependencies:
pip install requests aiohttp
Part 1: Authentication
"""
sp_auth.py — Amazon SP-API LWA authentication
Handles automatic Access Token refresh
"""
import requests
from datetime import datetime, timedelta
from threading import Lock
class SPAPIAuth:
TOKEN_URL = "https://api.amazon.com/auth/o2/token"
def __init__(self, client_id: str, client_secret: str, refresh_token: str):
self.client_id = client_id
self.client_secret = client_secret
self.refresh_token = refresh_token
self._token = None
self._expires_at = datetime.min
self._lock = Lock()
@property
def access_token(self) -> str:
"""Returns a valid access token, refreshing automatically if expired."""
with self._lock:
if datetime.now() >= self._expires_at:
self._refresh()
return self._token
def _refresh(self):
resp = requests.post(self.TOKEN_URL, data={
"grant_type": "refresh_token",
"refresh_token": self.refresh_token,
"client_id": self.client_id,
"client_secret": self.client_secret
}, timeout=10)
resp.raise_for_status()
data = resp.json()
self._token = data["access_token"]
self._expires_at = datetime.now() + timedelta(seconds=data["expires_in"] - 60)
Part 2: Listings Items API (Real-Time)
"""
listings_api.py — SP-API Listings Items API wrapper
"""
import requests
from typing import Optional, List
from datetime import datetime
class ListingsItemsAPI:
BASE = "https://sellingpartnerapi-na.amazon.com"
def __init__(self, auth: SPAPIAuth, seller_id: str, marketplace_id: str = "ATVPDKIKX0DER"):
self.auth = auth
self.seller_id = seller_id
self.marketplace_id = marketplace_id
def _headers(self) -> dict:
return {
"x-amz-access-token": self.auth.access_token,
"x-amz-date": datetime.utcnow().strftime("%Y%m%dT%H%M%SZ"),
"Content-Type": "application/json"
}
def get_product_type_schema(self, product_type: str) -> dict:
"""
IMPORTANT: Always fetch the schema before building listing payloads.
Each product type has unique required fields — don't skip this step.
Returns the JSON Schema for the specified product type.
"""
resp = requests.get(
f"{self.BASE}/definitions/2020-09-01/productTypes/{product_type}",
headers=self._headers(),
params={
"marketplaceIds": self.marketplace_id,
"requirements": "LISTING",
"locale": "en_US"
}
)
resp.raise_for_status()
schema = resp.json()
required = schema.get("schema", {}).get("required", [])
print(f"Product type '{product_type}' requires {len(required)} fields: {required[:8]}...")
return schema
def create_listing(
self,
seller_sku: str,
product_type: str,
attributes: dict
) -> dict:
"""
Create or fully update a listing via PUT.
Use for new listings or when updating many fields simultaneously.
"""
url = f"{self.BASE}/listings/2021-08-01/items/{self.seller_id}/{seller_sku}"
resp = requests.put(
url,
headers=self._headers(),
params={"marketplaceIds": self.marketplace_id},
json={
"productType": product_type,
"requirements": "LISTING",
"attributes": attributes
},
timeout=30
)
result = resp.json()
status = result.get("status", "UNKNOWN")
if status == "ACCEPTED":
print(f"✓ {seller_sku}: Created successfully")
else:
for issue in result.get("issues", []):
print(f"✗ {seller_sku} [{issue.get('severity')}] {issue.get('attributeName', '')}: {issue.get('message')}")
return result
def update_price(
self,
seller_sku: str,
product_type: str,
new_price: float
) -> dict:
"""
Update price only via PATCH.
Why PATCH instead of PUT for pricing?
- Faster processing (fewer fields to validate)
- Minimal payload = lower risk of triggering listing re-review
- Appropriate for high-frequency repricing (every 30 min)
"""
url = f"{self.BASE}/listings/2021-08-01/items/{self.seller_id}/{seller_sku}"
resp = requests.patch(
url,
headers=self._headers(),
params={"marketplaceIds": self.marketplace_id},
json={
"productType": product_type,
"patches": [{
"op": "replace",
"path": "/attributes/purchasable_offer",
"value": [{
"marketplace_id": self.marketplace_id,
"currency": "USD",
"our_price": [{"schedule": [{"value_with_tax": new_price}]}]
}]
}]
},
timeout=15
)
result = resp.json()
if result.get("status") == "ACCEPTED":
print(f"✓ {seller_sku}: Price updated to ${new_price:.2f}")
return result
def build_standard_attributes(
self,
title: str,
brand: str,
bullet_points: List[str],
description: str,
keywords: List[str],
price: float,
quantity: int,
image_url: str
) -> dict:
"""
Build a listing attributes dict covering most common product types.
Note: You'll need to add category-specific required fields based on
the schema returned by get_product_type_schema().
"""
return {
"item_name": [{"value": title[:200], "language_tag": "en_US"}],
"brand": [{"value": brand}],
"bullet_point": [
{"value": bp[:500], "language_tag": "en_US"}
for bp in bullet_points[:5]
],
"product_description": [{"value": description[:2000], "language_tag": "en_US"}],
"generic_keyword": [{"value": " ".join(keywords[:50])}],
"fulfillment_availability": [{
"fulfillment_channel_code": "DEFAULT",
"quantity": quantity
}],
"purchasable_offer": [{
"marketplace_id": self.marketplace_id,
"currency": "USD",
"our_price": [{"schedule": [{"value_with_tax": price}]}]
}],
"main_product_image_locator": [{"media_location": image_url}]
}
Part 3: Real-Time Pricing with Pangolinfo
"""
live_pricing.py — Real-time competitor price monitoring and repricing
Uses Pangolinfo Scrape API to fetch live Amazon product data
"""
import requests
from typing import Optional
class LivePricingEngine:
PANGOLINFO_URL = "https://api.pangolinfo.com/v1/scrape"
def __init__(self, pangolinfo_api_key: str, listings_api: ListingsItemsAPI):
self.pangolinfo_headers = {
"Authorization": f"Bearer {pangolinfo_api_key}",
"Content-Type": "application/json"
}
self.listings = listings_api
def fetch_live_price(self, asin: str) -> Optional[float]:
"""
Fetch real-time price for an Amazon ASIN via Pangolinfo Scrape API.
Returns the current Buybox price, or None if unavailable.
Pangolinfo's SP ad position capture rate: 98% — useful for
understanding how competitively a keyword is being bid on.
"""
try:
resp = requests.post(
self.PANGOLINFO_URL,
headers=self.pangolinfo_headers,
json={
"url": f"https://www.amazon.com/dp/{asin}",
"platform": "amazon",
"data_type": "product_detail",
"marketplace": "US",
"render": True,
"extract_ads": True # Also capture SP ad positions
},
timeout=30
)
resp.raise_for_status()
result = resp.json().get("result", {})
price_data = result.get("price", {})
current_price = price_data.get("current") if isinstance(price_data, dict) else price_data
if current_price:
print(f" Live price for {asin}: ${current_price:.2f} "
f"(Buybox: {result.get('buybox', {}).get('seller_name', 'N/A')})")
return current_price
except Exception as e:
print(f" Failed to fetch price for {asin}: {e}")
return None
def calculate_target_price(
self,
competitor_price: float,
unit_cost: float,
undercut: float = 0.02, # Price this % below competitor
min_margin: float = 0.15 # Never price below this margin
) -> float:
"""Calculate optimal price given competitor data and margin constraints."""
floor = unit_cost / (1 - min_margin)
target = round(competitor_price * (1 - undercut), 2)
return max(target, floor)
def reprice_product(
self,
seller_sku: str,
product_type: str,
competitor_asin: str,
unit_cost: float,
current_price: float
) -> bool:
"""
Check competitor price and update our price if needed.
Returns True if price was updated.
"""
comp_price = self.fetch_live_price(competitor_asin)
if not comp_price:
return False
new_price = self.calculate_target_price(comp_price, unit_cost)
# Only update if price changes by more than $0.01 (avoid unnecessary API calls)
if abs(new_price - current_price) <= 0.01:
print(f" {seller_sku}: No price change needed (${current_price:.2f})")
return False
result = self.listings.update_price(seller_sku, product_type, new_price)
return result.get("status") == "ACCEPTED"
def run_repricing_cycle(self, products: list) -> dict:
"""
Full repricing cycle for a list of monitored products.
Call this every 30 minutes for active price competition.
products: [{"sku": str, "product_type": str, "competitor_asin": str,
"cost": float, "current_price": float}, ...]
"""
updated = 0
skipped = 0
failed = 0
for product in products:
success = self.reprice_product(
seller_sku=product["sku"],
product_type=product["product_type"],
competitor_asin=product["competitor_asin"],
unit_cost=product["cost"],
current_price=product["current_price"]
)
if success:
updated += 1
elif success is False:
skipped += 1
else:
failed += 1
print(f"\nRepricing cycle complete: {updated} updated, {skipped} unchanged, {failed} failed")
return {"updated": updated, "skipped": skipped, "failed": failed}
Complete Integration Example
# main.py — Full auto listing + repricing example
auth = SPAPIAuth(
client_id="YOUR_CLIENT_ID",
client_secret="YOUR_CLIENT_SECRET",
refresh_token="YOUR_REFRESH_TOKEN"
)
listings = ListingsItemsAPI(auth, seller_id="YOUR_SELLER_ID")
pricing_engine = LivePricingEngine("YOUR_PANGOLINFO_KEY", listings)
# Step 1: Check what fields this product type actually requires
schema = listings.get_product_type_schema("WATER_BOTTLE")
# Step 2: Build listing attributes
attrs = listings.build_standard_attributes(
title="Premium 32oz Stainless Steel Water Bottle — Vacuum Insulated, BPA Free",
brand="YourBrand",
bullet_points=[
"STAYS COLD 24H / HOT 12H: Triple-wall vacuum insulation outperforms single-wall alternatives.",
"BPA-FREE 18/8 STAINLESS STEEL: Zero plastic taste, LFGB and FDA certified.",
"LEAK-PROOF LID: 360° silicone seal tested at all angles including inverted.",
"FITS CUP HOLDERS: 2.95\" diameter fits standard car cup holders and most bag pockets.",
"LIFETIME WARRANTY: No-questions replacement guarantee on every bottle."
],
description="For people who refuse to compromise on hydration quality...",
keywords=["insulated water bottle", "stainless steel water bottle", "bpa free water bottle",
"32oz water bottle", "vacuum insulated bottle", "leak proof water bottle"],
price=32.99, # Will be overwritten by live competitor data below
quantity=300,
image_url="https://cdn.yourbrand.com/wb-32oz-hero.jpg"
)
# Step 3: Get live competitor price and override before submitting
live_comp_price = pricing_engine.fetch_live_price("B08MAINCOMPETITOR")
if live_comp_price:
attrs["purchasable_offer"] = [{
"marketplace_id": "ATVPDKIKX0DER",
"currency": "USD",
"our_price": [{"schedule": [{"value_with_tax":
pricing_engine.calculate_target_price(live_comp_price, unit_cost=9.20)
}]}]
}]
# Step 4: Submit to Amazon
result = listings.create_listing("BRAND-WB-32OZ", "WATER_BOTTLE", attrs)
# Step 5: Set up 30-minute repricing cycle (use cron or scheduler)
monitored = [{
"sku": "BRAND-WB-32OZ",
"product_type": "WATER_BOTTLE",
"competitor_asin": "B08MAINCOMPETITOR",
"cost": 9.20,
"current_price": 31.65
}]
# Run this every 30 minutes:
pricing_engine.run_repricing_cycle(monitored)
Common Issues
INVALID on submission despite correct syntax:
Almost always a missing required field for the specific product type. Run get_product_type_schema() and check the required array.
Price update accepted but not reflected in Seller Central:
Price changes can take 10-30 minutes to propagate. Check the listing status via GET /listings/2021-08-01/items/{sellerId}/{sellerSku}.
Rate limit errors (429):
Listings Items API: 5 req/s burst, 0.5 req/s sustained. For bulk operations, use Feeds API instead of looping Listings Items PUT.
Feed stuck in IN_PROGRESS for hours:
Check GET /feeds/2021-06-30/feeds/{feedId} — if processingStatus stays IN_PROGRESS beyond 2 hours, there may be a malformed message in the feed causing processing to stall.
Resources
- Pangolinfo Scrape API — Amazon real-time data
- Pangolinfo Docs — API reference
- SP-API Listings Items API Reference
- SP-API Feeds API Reference
- Amazon Developer Central — App registration
Questions about the SP-API integration or the Pangolinfo data layer? Drop them in the comments.
Tags: #python #api #amazon #ecommerce #automation #tutorial #webdev
Top comments (0)