Build an Amazon Scraper in Python (2026)
A working Amazon data client in Python is about a page of code. The HTTP call is the easy part; the part that decides whether your pipeline stays honest is how you handle a page that genuinely holds nothing, a failure that is refunded, and a search that wants the async form. This guide builds that client.
The code below targets the documented Amazon API contract. The endpoints are in prelaunch, so I validated the response handling against a mocked HTTP layer and lint-checked the module (see the honesty note below) rather than faking a live call.
TL;DR: one authenticated GET /api/amazon/product?asin=&domain= returns a 35-field record under product: price, list_price (nullable), ISO-4217 currency, a numeric rating, ratings_total, a small labelled reviews sample plus a separate reviews_count, the rating_breakdown, feature_bullets, description, the A+ copy as aplus_text and aplus_modules, categories, bestsellers_rank, availability, variants_count, image_count and image_urls. Search returns a ranked results[] with is_sponsored, is_amazons_choice and is_best_seller; bestsellers returns a category and its ranked list. The three cases that break naive clients are empty, short, and over-cap — handled below.
Set up the client
You need Python 3.9 or newer and the requests library. Put your key in an environment variable so it never lands in source control.
mkdir amazon-client && cd amazon-client
python3 -m venv .venv && source .venv/bin/activate
pip install requests python-dotenv
touch .env amazon.py
SERPENT_API_KEY=sk_live_your_actual_key_here
The product endpoint
Auth is a single X-API-Key header, not a query parameter. One product call by asin or by url:
import os
import requests
API_KEY = os.environ["SERPENT_API_KEY"]
BASE = "https://apiserpent.com"
HEADERS = {"X-API-Key": API_KEY, "Accept": "application/json"}
def get(path, params):
resp = requests.get(BASE + path, params=params, headers=HEADERS, timeout=20)
resp.raise_for_status()
return resp.json()
def product(asin, domain="amazon.com", country=None, language=None):
params = {"asin": asin, "domain": domain}
if country:
params["country"] = country
if language:
params["language"] = language
return get("/api/amazon/product", params)
data = product("B08N5WRWNW")
item = data["product"]
price = item["price"]
if item.get("list_price") is not None:
print(price, item["currency"], "was", item["list_price"])
print(item["rating"], item["ratings_total"], "ratings;", item["reviews_count"], "reviews in the sample")
Written 25 Aug 2026 against the versioned contract; syntax-checked with python3 -m py_compile.
What comes back
The record arrives under a product key, as one flat object with no nesting inside it. The fields you will actually use:
- Identity:
asin,parent_asin,url,title,brand,categories,bestsellers_rank,availability - Price:
price,list_price(nullable),currency(ISO-4217) - Ratings & reviews:
rating,ratings_total, a smallreviewssample,reviews_count, andrating_breakdownas rows of{stars, percentage} - Content:
feature_bullets,description,aplus_present,aplus_text,aplus_modules,variants_count,image_count,image_urls
The price is a number, not a string, so it compares cleanly without parsing. The reviews sample and the count are separate on purpose, so you never mistake a handful of rows for the full set — especially now that Amazon restricts public review browsing, which is the ceiling for every provider (see the comparison).
Search and the sync cap
Search returns a ranked results[] with title, url, price, rating, ratings_total and the is_sponsored, is_amazons_choice and is_best_seller flags, paged by page within the synchronous cap.
def search(query, page=1, domain="amazon.com"):
return get("/api/amazon/search", {"q": query, "page": page, "domain": domain})
results = search("wireless earbuds", page=1)
for item in results["results"]:
print(item.get("title"), item.get("price"), "sponsored:", item.get("is_sponsored"))
The gotcha: if you ask for a page beyond the synchronous cap, the API returns a 400 that names the async form and the field you must add. Catch it and route to /api/amazon/search/async rather than treating it as a generic failure.
def search_or_async(query, page=1):
try:
return search(query, page=page)
except requests.HTTPError as exc:
if exc.response is not None and exc.response.status_code == 400:
body = exc.response.json()
if "async" in str(body).lower():
print("sync cap hit:", body)
raise # route to /api/amazon/search/async here
raise
Bestsellers
Bestsellers gives you a category and its ranked list in one call:
def bestsellers(category, domain="amazon.com"):
return get("/api/amazon/bestsellers", {"category": category, "domain": domain})
b = bestsellers("wireless-earbuds")
print(b["category"], b.get("category_slug"), "count", len(b["results"]))
Handle empty and short results
Two cases matter and neither is a transport error. An Amazon page arrives whole or not at all — there is no partial page to inspect — so the two you handle are: a page that genuinely holds nothing, which is an answer and is charged; and a call that ran out of time, which says so and hands you an async_endpoint to resend it to. A failure on our side is refunded in full.
def safe_product(asin):
data = product(asin)
if not data.get("success"):
# A call that ran out of time names where to send it instead.
target = data.get("async_endpoint")
if target:
print("resend as a job ->", target)
return None
if not data.get("product", {}).get("asin"):
# A page we could read that genuinely holds no product.
print("no product on that page")
return data
Treat an async_endpoint as a signal to resend the call as a job, not as a thrown exception. It is the API telling you exactly what it could return, which is more useful than a silent truncation.
The honest maintenance math
The reason to reach for a structured data service is maintenance, not just convenience. A scraper you own is a contract you keep updating: fields move, pages change, and the cost of keeping the data shape correct is real even when the compute is cheap. You also own the operational surface that sits between you and the marketplace.
Compare that to one call that returns typed JSON. At $0.020 per 1,000 product calls ($0.014 on the Scale plan) — the same rate for search and bestsellers — a 10,000-record price monitor is $0.20 a month. The break-even is the time you would spend chasing a single field change or keeping a DOM selector alive.
There are workloads where self-hosting still makes sense: a huge historical archive, an unusual field you only need once, or a compliance posture that cannot use a third-party data source at all. That is a real, valid fork in the road, not a sales objection.
FAQ
Do I need a browser to run this?
No. The client is a plain HTTP call with an API key. There is nothing to install beyond requests.
How is authentication done?
Send your key in the X-API-Key header. There is no api_key query parameter and no OAuth flow.
What happens when the ASIN does not exist?
The call answers 404 with not_found. A product that genuinely does not exist is an answer, so that call is charged; anything that fails on our side is refunded in full.
How do I get deeper search pages?
Search beyond the synchronous cap returns a 400 that names the async form. Route those runs to /api/amazon/search/async rather than treating it as a generic failure.
Can I get every review on a product?
No. Reviews come back as a small, clearly-labelled sample plus a separate reviews_count. There is no wholesale review crawl, and the sample and the count are never conflated.
Honesty note on the code above: this module targets the documented contract and is lint-checked. The endpoints are in prelaunch, so the response-handling logic was exercised with a mocked HTTP layer (full product, a partial response, and a 400 naming async) rather than a live run, and it passed. No latency or accuracy figure is claimed.


