DEV Community: Mox Loop

Real-Time Amazon Product Research Without Jungle Scout or Helium 10: A Complete API Integration Guide

Mox Loop — Tue, 14 Apr 2026 07:53:44 +0000

TL;DR

Jungle Scout and Helium 10 serve as Amazon product research tools, but their subscription architectures introduce 24–72 hour data lag and zero AI Agent integration capability. If you need real-time Amazon price/BSR/keyword data that your AI Agent or automation pipeline can actually call programmatically — this guide shows you how.

What you'll build:

A Python client for Pangolinfo Scrape API (real-time Amazon data)
Batch ASIN monitoring with async collection
Feishu/Slack alert integration for price movement detection
OpenClaw AI Agent integration via Amazon Scraper Skill

Prerequisites: Basic Python, a free Pangolinfo API key, and optionally OpenClaw for the AI Agent section.

Why Subscription Tools Don't Work for Programmatic Use

If you've tried to build any kind of automated Amazon monitoring or AI workflow, you've already hit this wall: subscriptions tools like Jungle Scout and Helium 10 don't provide a usable API for programmatic access. Their "API" offerings (where they exist at all) give you preprocessed, cached data from their own database — not real-time Amazon data. And their data interfaces are designed for humans clicking through dashboards, not for machines making structured data requests.

The specific limitations that matter for developers and operators building automated systems:

Data freshness: Most subscription tool data is 24–72 hours stale. For price monitoring or competitive intelligence that needs to detect same-day movements, this is disqualifying.

No AI Agent support: You cannot configure an AI Agent (LangChain, OpenClaw, Dify) to call Jungle Scout or Helium 10 for live data. There's no function calling interface, no webhook support, no real-time data feed. The only workaround is manual CSV export → upload to AI → analysis on stale data.

Closed data model: You get the fields the tool company decided to expose. Cross-tabulating coupon activation frequency against BSR trajectory, or pulling SP advertising position distribution alongside organic keyword rankings for the same query, isn't possible through any subscription tool.

Pangolinfo Scrape API solves all three by taking a different architectural approach: every API call triggers a live real-time request to Amazon's public pages. You get current-state data, structured JSON output, full programmatic access, and optional Skill-based AI Agent integration.

Setup

# Install dependencies
pip install requests aiohttp python-dotenv

# Create .env file
echo "PANGOLINFO_API_KEY=your_key_here" > .env
echo "FEISHU_WEBHOOK_URL=https://open.feishu.cn/..." >> .env

Get your free API key from tool.pangolinfo.com — no credit card required for the trial tier.

1. Basic Sync Client

Start with the simplest working implementation:

import os
import requests
from dotenv import load_dotenv
from typing import Optional, Dict

load_dotenv()

class PangolinClient:
    """Minimal synchronous client for Pangolinfo Scrape API"""

    BASE_URL = "https://api.pangolinfo.com/v2"

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.getenv("PANGOLINFO_API_KEY")
        if not self.api_key:
            raise ValueError("API key required: set PANGOLINFO_API_KEY env var")

        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def get_product(self, asin: str, marketplace: str = "amazon.com", 
                    zip_code: str = "90001") -> Dict:
        """
        Fetch real-time Amazon product data.
        Returns: price, BSR, rating, review count, coupon status, timestamp
        """
        response = requests.post(
            f"{self.BASE_URL}/amazon/product",
            json={"asin": asin, "marketplace": marketplace, "zip_code": zip_code},
            headers=self.headers,
            timeout=30
        )
        response.raise_for_status()
        return response.json()

    def get_keyword_results(self, keyword: str, marketplace: str = "amazon.com",
                            top_n: int = 20) -> Dict:
        """
        Fetch real-time keyword search results.
        Returns: ranked ASINs, organic positions, sponsored positions
        """
        response = requests.post(
            f"{self.BASE_URL}/amazon/keyword",
            json={"keyword": keyword, "marketplace": marketplace, "top_n": top_n},
            headers=self.headers,
            timeout=30
        )
        response.raise_for_status()
        return response.json()


# Quick test
if __name__ == "__main__":
    client = PangolinClient()
    data = client.get_product("B09Z8LSMSK")  # replace with real ASIN
    print(f"Price: ${data.get('current_price')}")
    print(f"BSR: #{data.get('bsr_rank')} in {data.get('bsr_category')}")
    print(f"Rating: {data.get('rating')} ({data.get('review_count'):,} reviews)")
    coupon = data.get('coupon_status', {})
    if coupon.get('is_active'):
        print(f"Coupon: {coupon.get('value')} ACTIVE")

2. Async Batch Collection (The Production Version)

For monitoring 20+ ASINs efficiently, async concurrent requests are essential:

import asyncio
import aiohttp
import time
from dataclasses import dataclass
from typing import List, Dict, Optional

@dataclass
class ProductSnapshot:
    asin: str
    price: Optional[float]
    bsr_rank: Optional[int]
    rating: Optional[float]
    review_count: Optional[int]
    coupon_active: bool
    coupon_value: Optional[str]
    collected_at: float
    error: Optional[str] = None


async def fetch_product_async(
    session: aiohttp.ClientSession,
    asin: str,
    api_key: str,
    semaphore: asyncio.Semaphore,
    marketplace: str = "amazon.com"
) -> ProductSnapshot:
    """Async single product fetch with concurrency control"""

    async with semaphore:  # Limits concurrent requests
        try:
            async with session.post(
                "https://api.pangolinfo.com/v2/amazon/product",
                json={"asin": asin, "marketplace": marketplace},
                headers={"Authorization": f"Bearer {api_key}"},
                timeout=aiohttp.ClientTimeout(total=45)
            ) as response:

                if response.status == 429:
                    await asyncio.sleep(5)
                    return await fetch_product_async(session, asin, api_key, semaphore)

                response.raise_for_status()
                data = await response.json()

                coupon = data.get("coupon_status", {})
                return ProductSnapshot(
                    asin=asin,
                    price=data.get("current_price"),
                    bsr_rank=data.get("bsr_rank"),
                    rating=data.get("rating"),
                    review_count=data.get("review_count"),
                    coupon_active=coupon.get("is_active", False),
                    coupon_value=coupon.get("value"),
                    collected_at=time.time()
                )

        except Exception as e:
            return ProductSnapshot(
                asin=asin, price=None, bsr_rank=None, rating=None,
                review_count=None, coupon_active=False, coupon_value=None,
                collected_at=time.time(), error=str(e)
            )


async def batch_fetch(
    asin_list: List[str],
    api_key: str,
    max_concurrent: int = 8
) -> Dict[str, ProductSnapshot]:
    """
    Concurrent batch product data collection.
    100 ASINs typically completes in 20–60 seconds.
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_product_async(session, asin, api_key, semaphore)
            for asin in asin_list
        ]
        results = await asyncio.gather(*tasks)

    return {snapshot.asin: snapshot for snapshot in results}


# Example usage
async def main():
    api_key = os.getenv("PANGOLINFO_API_KEY")

    # Monitor 30 competitor ASINs
    competitor_asins = ["B0COMP001", "B0COMP002", "B0COMP003"]  # Your real ASINs

    print(f"Fetching {len(competitor_asins)} ASINs concurrently...")
    start = time.time()

    snapshots = await batch_fetch(competitor_asins, api_key)

    elapsed = time.time() - start
    success = sum(1 for s in snapshots.values() if s.error is None)
    print(f"Done: {success}/{len(competitor_asins)} succeeded in {elapsed:.1f}s")

    for asin, snap in list(snapshots.items())[:3]:
        if snap.error:
            print(f"  {asin}: ERROR - {snap.error}")
        else:
            print(f"  {asin}: ${snap.price} | BSR #{snap.bsr_rank} | {snap.rating}⭐")


asyncio.run(main())

3. Price Movement Detection + Feishu Alerts

import json
import requests
from datetime import datetime

FEISHU_WEBHOOK = os.getenv("FEISHU_WEBHOOK_URL")

def detect_price_changes(
    current_snapshots: Dict[str, ProductSnapshot],
    previous_snapshots: Dict[str, ProductSnapshot],
    threshold_pct: float = 5.0
) -> List[Dict]:
    """Compare two snapshots, return list of alert events"""

    alerts = []

    for asin, current in current_snapshots.items():
        previous = previous_snapshots.get(asin)
        if not previous or current.error:
            continue

        # Price change detection
        if current.price and previous.price and previous.price > 0:
            pct_change = ((current.price - previous.price) / previous.price) * 100
            if abs(pct_change) >= threshold_pct:
                alerts.append({
                    "type": "price_change",
                    "asin": asin,
                    "prev_price": previous.price,
                    "curr_price": current.price,
                    "pct_change": round(pct_change, 2),
                    "direction": "drop" if pct_change < 0 else "increase",
                    "curr_bsr": current.bsr_rank,
                    "severity": "high" if abs(pct_change) >= 15 else "medium"
                })

        # Coupon activation detection
        if current.coupon_active and not previous.coupon_active:
            alerts.append({
                "type": "coupon_activated",
                "asin": asin,
                "coupon_value": current.coupon_value,
                "curr_price": current.price,
                "curr_bsr": current.bsr_rank
            })

    return alerts


def send_feishu_alert(alerts: List[Dict]):
    """Push price alerts to Feishu bot"""
    if not alerts or not FEISHU_WEBHOOK:
        return

    text = f"🔔 **Amazon Competitor Alert** — {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n"

    for alert in alerts:
        asin = alert['asin']
        link = f"https://www.amazon.com/dp/{asin}"

        if alert['type'] == 'price_change':
            direction = "⬇ Price Drop" if alert['direction'] == 'drop' else "⬆ Price Increase"
            text += (
                f"{'🔴' if alert['severity'] == 'high' else '🟡'} **{direction}** "
                f"[{asin}]({link})\n"
                f"${alert['prev_price']:.2f} → ${alert['curr_price']:.2f} "
                f"({alert['pct_change']:+.1f}%) | BSR #{alert.get('curr_bsr', 'N/A')}\n\n"
            )
        elif alert['type'] == 'coupon_activated':
            text += (
                f"🟡 **Coupon Activated** [{asin}]({link})\n"
                f"Value: {alert['coupon_value']} | "
                f"Price: ${alert.get('curr_price', 'N/A'):.2f} | "
                f"BSR #{alert.get('curr_bsr', 'N/A')}\n\n"
            )

    payload = {
        "msg_type": "text",
        "content": {"text": text}
    }
    requests.post(FEISHU_WEBHOOK, json=payload, timeout=10)
    print(f"✓ Sent {len(alerts)} alerts to Feishu")

4. AI Agent Integration via Amazon Scraper Skill

The fastest path to getting your AI Agent access to real-time Amazon data is the Pangolinfo Amazon Scraper Skill for OpenClaw.

Setup: Find in OpenClaw Skill library → Install → Enter API key → Done.

Example Agent conversations that now work:

You: Pull realtime data for keyword "portable espresso maker" — 
     top 15 ASINs, price range, rating distribution, BSR ranking.
     Which ASINs are rated below 4.0 and what are their main complaint themes?

Agent: [Calls Scraper Skill → Returns structured analysis with live data]

You: Monitor ASINs [B0XX001, B0XX002, B0XX003].
     Flag any that have activated coupons or changed price >5% since yesterday.

Agent: [Compares current vs cached data → Delivers change summary]

LangChain custom tool (for non-OpenClaw Agent frameworks):

from langchain.tools import BaseTool
from pydantic import BaseModel, Field

class ProductQueryInput(BaseModel):
    asin: str = Field(description="Amazon ASIN to fetch real-time data for")
    marketplace: str = Field(default="amazon.com")

class AmazonProductTool(BaseTool):
    name = "amazon_realtime_product_data"
    description = (
        "Fetch real-time Amazon product data: current price, BSR ranking, "
        "rating, review count, coupon status. Returns live data from Amazon, "
        "not cached. Use for competitor monitoring and product research."
    )
    args_schema = ProductQueryInput
    api_key: str

    def _run(self, asin: str, marketplace: str = "amazon.com") -> str:
        import requests
        resp = requests.post(
            "https://api.pangolinfo.com/v2/amazon/product",
            json={"asin": asin, "marketplace": marketplace},
            headers={"Authorization": f"Bearer {self.api_key}"},
            timeout=30
        )
        d = resp.json()
        coupon = d.get("coupon_status", {})
        return (
            f"ASIN {asin} — Live data as of {d.get('timestamp', 'now')}:\n"
            f"Price: ${d.get('current_price', 'N/A')}\n"
            f"BSR: #{d.get('bsr_rank', 'N/A')} ({d.get('bsr_category', '')})\n"
            f"Rating: {d.get('rating')} ({d.get('review_count', 0):,} reviews)\n"
            f"Coupon: {'ACTIVE — ' + str(coupon.get('value','')) if coupon.get('is_active') else 'None'}"
        )

    async def _arun(self, **kwargs): raise NotImplementedError

Full Monitoring Script (Bring It All Together)

#!/usr/bin/env python3
"""
Complete Amazon competitor monitoring system
Runs on schedule, detects price/coupon changes, sends Feishu alerts
"""

import asyncio
import json
import os
import schedule
import time
from datetime import datetime

PANGOLINFO_API_KEY = os.getenv("PANGOLINFO_API_KEY")
FEISHU_WEBHOOK = os.getenv("FEISHU_WEBHOOK_URL")

MONITOR_ASINS = [
    "B0COMP001",  # Replace with your actual competitor ASINs
    "B0COMP002",
    "B0COMP003",
]

previous_snapshots: Dict[str, ProductSnapshot] = {}  # In-memory cache (use DB in production)


async def monitoring_cycle():
    print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Running monitoring cycle...")

    global previous_snapshots

    # Fetch current state
    current = await batch_fetch(MONITOR_ASINS, PANGOLINFO_API_KEY)

    # Detect changes if we have previous data
    if previous_snapshots:
        alerts = detect_price_changes(current, previous_snapshots, threshold_pct=5.0)
        if alerts:
            send_feishu_alert(alerts)
            print(f"  Detected {len(alerts)} changes — alerts sent")
        else:
            print(f"  No significant changes detected")
    else:
        print(f"  First run — establishing baseline snapshot")

    # Update cache
    previous_snapshots = current

    success = sum(1 for s in current.values() if s.error is None)
    print(f"  Snapshot complete: {success}/{len(MONITOR_ASINS)} ASINs collected")


def main():
    print("Amazon Competitor Monitor — Starting")
    print(f"Monitoring {len(MONITOR_ASINS)} ASINs every 2 hours")

    # Run immediately on start
    asyncio.run(monitoring_cycle())

    # Schedule subsequent runs
    schedule.every(2).hours.do(lambda: asyncio.run(monitoring_cycle()))

    while True:
        schedule.run_pending()
        time.sleep(60)


if __name__ == "__main__":
    main()

Deployment

# Local background process
nohup python monitor.py &

# Or Docker
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "monitor.py"]

Key Takeaways

Subscription tools (Jungle Scout, Helium 10) have 24–72 hour data lag by architectural design — not fixable by upgrading your plan
Real-time API calls via Pangolinfo return actual current Amazon data, enabling true competitive monitoring
The async batch pattern handles 100 ASINs in 20–60 seconds
AI Agent integration works natively via Amazon Scraper Skill (OpenClaw) or as a LangChain custom tool
Total monitoring stack cost at moderate volume: ~$30–50/month vs $100–200+/month for equivalent subscription coverage

Free trial (no credit card): tool.pangolinfo.com
API docs: docs.pangolinfo.com

Async vs Sync Amazon Data Scraping API: A Complete Developer Guide

Mox Loop — Mon, 13 Apr 2026 07:01:05 +0000

TL;DR

Sync API: ~5s per call, blocks until result returns. Use for real-time, user-triggered queries.
Async API: submits task in <200ms, result delivered via callback. Use for batch/scheduled work.
Same credit cost both ways (1 credit/JSON call)
Pangolinfo Scrape API supports both modes under the same token

The Problem with Defaulting to Sync

Amazon Scrape API calls take around 5 seconds per response—realistic given what the server has to do (load Amazon pages, bypass anti-bot systems, parse DOM, return structured JSON). For a single query, that's totally fine.

For a monitoring system tracking 5,000 ASINs daily? That's 6.9 hours of serial waiting. Add network variance and you're looking at 8–10 hours. Most business use cases can't absorb that delay.

Async Amazon data scraping breaks the dependency between request throughput and client-side wait time. You submit tasks in bulk, the platform processes them concurrently server-side, and results arrive at your callback endpoint. Your client never blocks.

API Endpoint Comparison

	Sync	Async
Endpoint	`POST /api/v1/scrape`	`POST /api/v1/scrape/async`
Client wait	~5 seconds	<200ms (returns taskId)
Result delivery	Inline response body	POST to your `callbackUrl`
Infrastructure req	None	Public callback endpoint
Credits (JSON)	1/call	1/call

Sync Mode: Complete Example

import requests

def sync_scrape(token: str, asin: str) -> dict | None:
    """
    Sync Amazon product detail scrape.
    Blocks ~5s. Use for real-time queries.
    """
    resp = requests.post(
        "https://scrapeapi.pangolinfo.com/api/v1/scrape",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        },
        json={
            "url": f"https://www.amazon.com/dp/{asin}",
            "parserName": "amzProductDetail",
            "site": "",
            "content": "",
            "format": "json",
            "bizContext": {"zipcode": "10041"}
        },
        timeout=30  # Must exceed API processing time (~5s)
    )
    data = resp.json()
    if data.get("code") == 0:
        return data["data"]["json"][0]["data"]["results"][0]
    raise RuntimeError(f"API error: {data.get('message')}")


# Usage
product = sync_scrape("your_token", "B0DYTF8L2W")
print(product["title"])
print(product["star"])        # e.g. "4.5"
print(product["bestSellersRank"])

Available Parser Types

PARSER_NAMES = {
    "amzProductDetail":    "Product detail by ASIN or URL",
    "amzKeyword":          "Keyword search results",
    "amzProductOfCategory": "Category product list (by Node ID)",
    "amzProductOfSeller":  "Seller product list",
    "amzBestSellers":      "Best sellers ranking",
    "amzNewReleases":      "New releases ranking"
}

Async Mode: Complete Implementation

1. Submit Tasks

import requests
import time
from typing import Optional

ASYNC_URL = "https://scrapeapi.pangolinfo.com/api/v1/scrape/async"

def async_submit(
    token: str,
    asin: str,
    callback_url: str,
    parser_name: str = "amzProductDetail",
    zipcode: str = "10041"
) -> Optional[str]:
    """Submit async scraping task. Returns task_id immediately."""
    resp = requests.post(
        ASYNC_URL,
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        json={
            "url": f"https://www.amazon.com/dp/{asin}",
            "callbackUrl": callback_url,
            "zipcode": zipcode,
            "format": "json",
            "parserName": parser_name
        },
        timeout=10  # Submission itself is fast
    )
    data = resp.json()
    if data.get("code") == 0:
        return data["data"]["data"]  # task_id
    print(f"Submit failed for {asin}: {data.get('message')}")
    return None


def bulk_submit(
    token: str,
    asins: list[str],
    callback_url: str,
    rate_delay: float = 0.1  # seconds between submissions
) -> dict[str, str]:
    """Bulk submit. Returns {asin: task_id}."""
    task_map = {}
    for i, asin in enumerate(asins):
        task_id = async_submit(token, asin, callback_url)
        if task_id:
            task_map[asin] = task_id
        if (i + 1) % 500 == 0:
            print(f"Submitted {i+1}/{len(asins)}")
        time.sleep(rate_delay)
    return task_map

2. Callback Receiver (FastAPI)

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()
processed_ids: set[str] = set()  # Use Redis in production

@app.post("/api/callback")
async def receive_amazon_callback(request: Request):
    payload = await request.json()
    task_id = payload.get("taskId")

    # Idempotency guard - critical for production
    if task_id in processed_ids:
        return JSONResponse({"code": 0, "message": "duplicate"})
    processed_ids.add(task_id)

    # Extract results
    results = (
        payload
        .get("data", {})
        .get("json", [{}])[0]
        .get("data", {})
        .get("results", [])
    )
    if results:
        product = results[0]
        print(f"Received: {product.get('title', 'N/A')[:60]}")
        # persist_to_db(task_id, product)

    return JSONResponse({"code": 0, "message": "ok"})

3. Timeout Compensation

def compensate_missing_callbacks(
    token: str,
    task_map: dict[str, str],
    received_ids: set[str]
) -> None:
    """
    For tasks that didn't callback within expected window,
    fetch results via sync API as fallback.
    """
    missing = {asin: tid for asin, tid in task_map.items()
               if tid not in received_ids}

    for asin, task_id in missing.items():
        print(f"Compensating {asin} (task {task_id})")
        result = sync_scrape(token, asin)  # fallback to sync
        if result:
            print(f"Compensated: {result.get('title', 'N/A')[:40]}")

Decision Framework

Daily task volume?
│
├── < 100 tasks/day
│   └── Need instant results?
│       ├── Yes → Sync API ✓
│       └── No  → Sync API ✓ (async overhead not worth it)
│
└── ≥ 100 tasks/day OR scheduled batches
    └── Can deploy public callback server?
        ├── Yes → Async API ✓ (10x+ throughput)
        └── No  → Sync + threading, or AMZ Data Tracker (no-code)

Local Development Setup

Testing async locally without deploying to a server:

# Terminal 1: Run your callback receiver
uvicorn main:app --port 5000

# Terminal 2: Expose it publicly via ngrok
ngrok http 5000
# → Forwarding: https://abc123.ngrok.io → http://localhost:5000

# Use the ngrok URL as your callbackUrl for testing
CALLBACK_URL = "https://abc123.ngrok.io/api/callback"

Performance Benchmarks

Task Count	Sync (single-thread)	Async
100	~8 min	~2 min
1,000	~83 min	~5-10 min
5,000	~7 hrs	~20-40 min
10,000	~14 hrs	~45-90 min

Credit cost is identical in both modes. Performance gap grows linearly with volume.

Resources

Pangolinfo Scrape API — supports both sync and async
API Documentation — complete parameter reference
AMZ Data Tracker — no-code scheduled data collection

Questions? Drop them in the comments.

From 1M Records/Month to 10M/Day: A Real Case Study in Amazon Data Infrastructure

Mox Loop — Wed, 08 Apr 2026 07:25:21 +0000

Who This Is For

If you're building an Amazon seller tool, a data analytics product, or any SaaS where web data is the core value proposition — this case study is probably directly relevant to your stack decisions. It's not theoretical. It's a documented infrastructure migration, including the POC numbers, migration approach, and six-month outcomes.

Background

SellerPulse (anonymized) is a leading Amazon seller tool platform: 32,000+ registered users, ~18% paid conversion, MRR in the $1M+ range. Core product: real-time competitor price monitoring and SP ad placement tracking.

By late 2024, their self-built scraping infrastructure had become the company's biggest operational liability:

200 scraper nodes + 7 dedicated maintenance engineers
$12,000/month in IP infrastructure spend
SP ad slot capture rate: 62% (38% of their most-valued data type was simply missing)
Average data latency: 52 hours (their marketing said "real-time")
Quarterly outages triggered by Amazon anti-scraping policy updates

The Infrastructure Problem, Technically

Amazon's anti-bot systems underwent a significant architectural shift in 2024. Before this, IP rotation + reasonable request throttling was largely sufficient. Post-2024, Amazon deployed behavioral fingerprinting and session continuity verification — meaning that even with rotating IPs, requests exhibiting non-human session patterns (wrong resource loading order, atypical scroll behavior, browser fingerprint anomalies) are systematically identified and served degraded content.

For data center IPs specifically, this produces a systemic SP ad capture problem: Amazon's ad auction system serves authentic ad results based on perceived request legitimacy. Requests from data center ranges get systematically stripped of ad content or receive placeholder responses. That's the architectural reason SellerPulse's capture rate was stuck at 62% — not bad engineering, but a fundamental infrastructure constraint.

The fix requires:

Residential-grade IP infrastructure (looks like real user traffic to Amazon's systems)
Full browser rendering (JS execution, realistic resource loading sequence, scroll simulation)
Session continuity (consistent browser fingerprint across a session lifecycle)

These three requirements together are expensive and complex to maintain at scale for a company whose core competency is building a seller tool product, not running residential proxy networks.

The Solution: Pangolinfo Scrape API

After a two-week POC on their actual production URL sets, the numbers:

Metric	Self-built	Pangolinfo
SP ad slot capture rate	62%	97.3%
Data latency (P95)	52 hours	18 minutes
Zip-code targeting	Not supported	Supported natively
Customer Says field	Not supported	91.2% coverage
Pricing model	Fixed high cost	Pay-per-use

The zip-code and Customer Says findings were surprises. SellerPulse hadn't been able to do region-specific ad collection at all — Pangolinfo's native geographic targeting support opened entirely new product feature directions. The Customer Says field (Amazon's AI-generated review sentiment summary) had been completely inaccessible to self-managed scrapers; at 91%+ coverage it immediately became a candidate for a new competitive intelligence feature.

The Implementation: 90-Day Phased Migration

Core Approach: Traffic Splitting + Parallel Comparison

No downtime. No big-bang cutover. Traffic splitting from day one.

# Simplified traffic routing logic
import random

def route_scrape_request(url: str, pangolinfo_ratio: float = 0.20) -> str:
    """
    Route scraping requests between legacy and Pangolinfo systems.
    pangolinfo_ratio increases from 0.20 → 1.00 over 60 days.
    """
    if random.random() < pangolinfo_ratio:
        return "pangolinfo"
    return "legacy"

# Week 1-2: 20% Pangolinfo
# Week 3-4: 40% Pangolinfo  
# Week 5-6: 60% Pangolinfo
# Week 7-8: 80% Pangolinfo
# Week 9+:  100% Pangolinfo (core pipelines)

Parallel Comparison Layer

Both systems run simultaneously during migration; a comparison service logs divergence for manual review:

import hashlib
import json
from dataclasses import dataclass
from typing import Optional

@dataclass
class ComparisonResult:
    url: str
    legacy_fingerprint: Optional[str]
    pangolinfo_fingerprint: Optional[str]
    match: bool
    divergent_fields: list

def compare_outputs(legacy_data: dict, pangolinfo_data: dict, key_fields: list) -> ComparisonResult:
    """Compare outputs from both systems on key fields."""

    def fingerprint(data, fields):
        subset = {k: data.get(k) for k in fields if data}
        return hashlib.md5(json.dumps(subset, sort_keys=True).encode()).hexdigest()

    lf = fingerprint(legacy_data, key_fields)
    pf = fingerprint(pangolinfo_data, key_fields)

    divergent = []
    if legacy_data and pangolinfo_data:
        for field in key_fields:
            if legacy_data.get(field) != pangolinfo_data.get(field):
                divergent.append(field)

    return ComparisonResult(
        url=legacy_data.get("url", "") if legacy_data else "",
        legacy_fingerprint=lf,
        pangolinfo_fingerprint=pf,
        match=(lf == pf),
        divergent_fields=divergent
    )

Phase 3 — New Feature Rollout (Days 61-90)

With the core pipeline complete, features that had been blocked by infrastructure constraints shipped in rapid succession:

# Customer Says extraction — previously impossible with self-built scrapers
def extract_customer_says(product_data: dict) -> dict:
    """
    Extract Amazon's AI-generated review summary (Customer Says field)
    Available in Pangolinfo structured JSON output by default
    """
    customer_says = product_data.get("customer_says", {})

    return {
        "positive_attributes": customer_says.get("positive", []),
        "negative_attributes": customer_says.get("negative", []),
        "common_mentions": customer_says.get("common_themes", []),
        "sentiment_score": customer_says.get("overall_sentiment"),
        "mention_count": customer_says.get("review_count", 0)
    }

# Full category New Releases monitoring — previously too expensive to run at scale
def monitor_new_releases(categories: list, api_key: str):
    """Now economically viable at commodity per-record rates"""
    from pangolin_client import PangolinScrapeClient, ScrapeConfig

    client = PangolinScrapeClient(api_key)
    results = []

    for category in categories:
        config = ScrapeConfig(
            url=f"https://www.amazon.com/gp/new-releases/{category}/",
            parse_template="amazon_new_releases",
            extract_fields=["rank", "asin", "title", "price", "rating", "launch_date", "badge"]
        )
        data = client.scrape(config)
        if data:
            results.append(data)

    return results

Results: 6 Months Post-Migration

BEFORE vs AFTER

Daily Records:      29,000          →    12,000,000   (+41,000%)
SP Ad Capture:      62%             →    98.1%         (+36.1ppt)
Latency (avg):      52 hours        →    13 minutes    (-99.6%)
Customer Says:      0%              →    91%           (new capability)
Monthly Cost:       ~$30,000        →    ~$13,500      (-55%)
                    (at 870K/mo)          (at 330M/mo)

Paid Conversion:    18.3%           →    22.7%         (+4.4ppt)
Monthly Churn:      11.8%           →    6.4%          (-5.4ppt)
Monthly ARR:        $1.1M           →    $1.57M        (+42.5%)

Engineering (crawler maintenance): -85%
New features shipped (30 days post-migration): 4
Annualized ROI: 14.3x

Dev Notes: What Actually Tripped Us Up

Zip code matters more than you'd expect. Different US zip codes can show materially different prices (shipping cost locality), different ad inventories (regional ad targeting), and different Prime badge availability. For any feature involving geographic competitive analysis, setting geo.zip_code correctly isn't optional.

Customer Says field isn't on every ASIN. Amazon's AI review summaries require a threshold of reviews before generating. Expect ~8-12% of ASINs to return empty for this field — handle gracefully, don't treat as an error.

Concurrency sweet spot. They settled on semaphore-controlled async at 20 concurrent requests for standard tasks, bumped to 50 during peak events with Pangolinfo's expanded allocation. Error rate at 20: <0.1%. Setting it higher without coordination with Pangolinfo's account team caused transient throttling.

Parse templates vs. manual field extraction. For common scenarios (bestsellers, search results, product detail pages), Pangolinfo's prebuilt parse templates save significant engineering time. For unusual ASINs or edge-case category structures, the extract_fields approach with explicit field targeting is more reliable.

Conclusion

The core engineering takeaway: past roughly 1-2M records/day, self-managed scraping infrastructure for Amazon data becomes increasingly difficult to maintain at production quality levels. The combination of behavioral fingerprinting, session continuity requirements, residential IP needs, and the ongoing anti-bot escalation pattern makes it a moving target that academic engineering quality alone can't consistently hit.

For teams considering a similar migration:

Run your own POC first. The numbers from vendor case studies (including this one) are real, but your URL distribution, category mix, and geo targeting requirements will produce your own baseline. Two weeks of POC on your actual production URLs tells you more than any case study.
Plan for a phased migration. Big-bang cutovers on live data pipelines are unnecessary and risky. Traffic splitting plus parallel comparison is the standard approach for good reason.
Budget for Phase 3. The features you'll unlock post-migration are real business value. Factor them into your ROI calculations.

API documentation: docs.pangolinfo.com
Free trial: pangolinfo.com/scrape-api

Tags: #api #python #ecommerce #webdev #scrapers #amazon #dataengineering #saasbuild

Amazon Auto Listing with SP-API: Complete Tutorial with Real-Time Pricing Integration

Mox Loop — Tue, 07 Apr 2026 07:44:57 +0000

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 Listings and Feeds role 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

Best Amazon Data Scraping Solution in 2026: A Developer's Honest Guide

Mox Loop — Fri, 03 Apr 2026 08:36:37 +0000

I've seen a lot of "how to scrape Amazon" tutorials. Most of them are outdated, incomplete, or both.

This is different. This is an honest breakdown of what actually works for Amazon data collection in 2026, including success rate data, real cost comparisons, and working code examples you can run today.

TL;DR

Approach	Success Rate	Monthly Cost (10K/day)	Maintenance
DIY scraper + free proxies	<15%	Low (but hidden costs)	20-40hrs/month
DIY + residential proxies	40-55%	$3,000-8,000	20-40hrs/month
General scraping API	70-85%	$500-1,500	2-5hrs/month
Dedicated e-commerce API	95%+	$300-800	< 2hrs/month

Skip to the code examples if you already know why DIY Amazon scrapers fail in 2026.

Why Amazon Scrapers Keep Failing in 2026

Amazon's bot detection has crossed a threshold. It's no longer running rule-based checks you can enumerate and bypass — it's running ML behavioral analysis trained on millions of bot interaction patterns.

The five-layer defense stack:

IP reputation scoring — Your proxy's IP might have been flagged before you even bought it
ML behavioral analysis — Session-level sequence analysis, not individual request flags
Browser fingerprinting — Canvas, WebGL, fonts, screen metrics — headless browsers have detectable signatures
Account risk propagation — Programmatic behavior contaminates associated infrastructure
Honeypot content delivery — The silent killer: serving wrong data to suspicious sessions without triggering error responses

The honeypot problem is particularly insidious. Your scraper runs fine, returns 200 OK, logs everything as successful — and you're collecting garbage data you won't notice until something downstream produces anomalous results.

Comparing Approaches

Approach 1: DIY Scrapy/Requests + Proxy Pool

# What everyone starts with
import requests
from scrapy import Selector

def scrape_amazon_diy(asin, proxy):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)...",
        "Accept-Language": "en-US,en;q=0.9",
    }

    try:
        resp = requests.get(
            f"https://www.amazon.com/dp/{asin}",
            headers=headers,
            proxies={"https": proxy},
            timeout=15
        )

        if resp.status_code == 200:
            sel = Selector(text=resp.text)
            title = sel.css("#productTitle::text").get("").strip()
            price = sel.css(".a-price-whole::text").get("").strip()
            return {"title": title, "price": price}
        elif resp.status_code == 503:
            # CAPTCHA or bot challenge page
            return None
    except Exception:
        return None

Reality check on this approach:

Public proxies: <15% success rate on Amazon in 2026
Quality residential proxies: 40-55% — costs $10-20/GB, Amazon pages average 1-2MB each
Parse breakage: Amazon updates page structure regularly; your CSS selectors will break without warning
Honeypot blindness: you won't know you're collecting wrong data

Verdict: Acceptable for experimentation. Not for production.

Approach 2: Dedicated E-Commerce Scraping API

import requests
from typing import Optional

class AmazonScraper:
    """
    Production-grade Amazon scraper using Pangolinfo Scrape API
    Docs: https://docs.pangolinfo.com/en-api-reference/universalApi/universalApi
    """

    BASE_URL = "https://api.pangolinfo.com/v1/scrape"

    def __init__(self, api_key: str):
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })

    def get_product(
        self,
        asin: str,
        marketplace: str = "US",
        zip_code: Optional[str] = None
    ) -> Optional[dict]:
        """
        Fetch structured Amazon product data.

        Returns clean JSON with:
        - title, brand, description
        - price (current, list_price, savings)
        - rating, review_count
        - buybox (seller_name, is_prime, is_amazon_fulfilled)
        - bsr (Best Seller Rank by category)
        - sponsored_products (SP ad positions, 98% capture rate)
        - customer_says (thematic summary)
        - images, variations, technical_specs
        """
        payload = {
            "url": f"https://www.amazon.com/dp/{asin}",
            "platform": "amazon",
            "data_type": "product_detail",
            "marketplace": marketplace,
            "render": True,           # JS rendering enabled
            "extract_ads": True,      # Capture SP ad positions
            "extract_customer_says": True
        }

        if zip_code:
            payload["zip_code"] = zip_code  # Geo-targeted pricing

        resp = self.session.post(self.BASE_URL, json=payload, timeout=30)
        resp.raise_for_status()
        return resp.json().get("result")

    def search(
        self,
        keyword: str,
        marketplace: str = "US",
        page: int = 1
    ) -> Optional[dict]:
        """
        Keyword search with sponsored position tagging.
        Returns organic_results + sponsored_results (with position metadata)
        """
        payload = {
            "url": f"https://www.amazon.com/s?k={keyword}&page={page}",
            "platform": "amazon",
            "data_type": "search",
            "marketplace": marketplace,
            "render": True,
            "extract_ads": True
        }

        resp = self.session.post(self.BASE_URL, json=payload, timeout=30)
        resp.raise_for_status()
        return resp.json().get("result")

    def get_best_sellers(
        self,
        category_url: str,
        marketplace: str = "US"
    ) -> list:
        """Pull complete Best Sellers / New Releases list for a category"""
        payload = {
            "url": category_url,
            "platform": "amazon",
            "data_type": "best_sellers",
            "marketplace": marketplace,
            "render": True,
            "extract_ads": True
        }

        resp = self.session.post(self.BASE_URL, json=payload, timeout=60)
        resp.raise_for_status()
        return resp.json().get("result", {}).get("products", [])


# Usage examples
scraper = AmazonScraper(api_key="your_pangolinfo_key")

# 1. Product detail with NYC pricing
product = scraper.get_product("B08N5WRWNW", zip_code="10001")
print(f"""
Title: {product['title']}
Price (NYC): ${product['price']['current']}
Rating: {product['rating']} ({product['review_count']} reviews)
Buybox: {product['buybox']['seller_name']}
SP Ads captured: {len(product['sponsored_products'])}
""")

# 2. Keyword competitive analysis
results = scraper.search("stainless steel water bottle")
sponsored = results.get("sponsored_results", [])
organic = results.get("organic_results", [])
print(f"Sponsored positions: {len(sponsored)}, Organic: {len(organic)}")

# 3. Category intelligence
best_sellers = scraper.get_best_sellers(
    "https://www.amazon.com/Best-Sellers-Kitchen-Dining/zgbs/kitchen/"
)
print(f"Retrieved {len(best_sellers)} Best Sellers")

# Find rising stars: high rating, low review count
rising = [p for p in best_sellers 
          if p.get('rating', 0) >= 4.5 and p.get('review_count', 999) < 500]
print(f"Rising star candidates: {len(rising)}")

Real-World Performance Comparison {#implementation}

I ran a 7-day test comparing approaches across three Amazon page types. Results:

Best Sellers category pages:

DIY + residential proxy: 48% success, $23 per 1,000 successful pages
Pangolinfo API: 96% success, ~$8 per 1,000 successful pages (estimate)

Product detail pages with advertising data:

DIY approach: SP ad positions captured correctly in ~12% of attempts
Pangolinfo API: SP ad positions captured in ~98% of attempts

Parse reliability over 30 days:

DIY approach: Required CSS selector fixes on 3 separate occasions after Amazon updates
Pangolinfo API: Zero maintenance required on calling code

AI Agent Integration

One underrated use case: using this as a live data layer for AI agents.

LLMs have knowledge cutoffs. "What's currently selling in the kitchen category?" answered from training data is increasingly fictional.

Pangolinfo packages its API as a standardized Amazon Scraper Skill compatible with AI agent frameworks. The agent calls the skill to retrieve live Amazon data before generating analysis — grounding responses in current market reality rather than stale training data.

# Conceptual: AI Agent with live Amazon data
agent_query = """
Analyze the top 20 products in Amazon US Kitchen Best Sellers.
Identify products with rating >= 4.5 and review count < 1000.
For each, estimate monthly sales and flag competitive gaps.
"""

# Agent automatically calls Pangolinfo Scraper Skill for live data
# Then applies LLM reasoning to the fresh dataset
response = agent.run(agent_query)  # Real data, not hallucinated

Quick Start

Sign up at tool.pangolinfo.com — free trial available
Check the API reference for full field documentation
Start with Best Sellers category pulls — lowest latency, clearest ROI demonstration

Summary

If you're building anything that depends on reliable Amazon data in 2026, the engineering decision is straightforward: the maintenance cost and success rate gap between DIY approaches and purpose-built APIs has grown large enough that specialized infrastructure is the rational default choice at any meaningful scale.

Save the DIY scraper for learning. Use the right tool for production.

Tags: #python #api #amazon #ecommerce #webscaping #dataengineering #tutorial

Questions about the implementation? Drop them in the comments — happy to dig into specifics.

Stop Drowning in Amazon Data: Build a Decision Framework with Scrape API

Mox Loop — Wed, 01 Apr 2026 05:14:19 +0000

TL;DR: Amazon seller data overload is a real problem, but it's fixable with the right infrastructure. This tutorial shows you how to unify BSR, keyword SERP, and ad position data using Pangolinfo's Scrape API — so your analytics dashboard answers questions instead of creating more of them.

The Problem

Amazon operators typically juggle three data sources that don't talk to each other:

BSR rankings — updates hourly, extracted from product pages
ABA keyword data — weekly CSV downloads from Seller Central (manual)
Advertising reports — 24-48 hour delay, different metrics entirely

When BSR drops 20 positions, diagnosing the root cause means manually correlating data from all three sources. Average time: 2-3 hours per incident. And the diagnosis still might be wrong.

The fix is a unified data pipeline that aligns these three streams on a shared time axis. Here's how to build one.

Prerequisites

pip install requests psycopg2-binary python-dotenv

PANGOLINFO_API_KEY=your_key_here
DATABASE_URL=postgresql://user:pass@localhost:5432/amazon_analytics

Step 1: Database Schema

-- Unified time-series schema for Amazon operational data
CREATE TABLE asin_snapshots (
    id              BIGSERIAL PRIMARY KEY,
    asin            VARCHAR(20) NOT NULL,
    marketplace     VARCHAR(5)  NOT NULL DEFAULT 'US',
    bsr_main        INTEGER,
    bsr_category    VARCHAR(100),
    price           DECIMAL(10,2),
    review_count    INTEGER,
    rating          DECIMAL(3,2),
    collected_at    TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE keyword_snapshots (
    id                  BIGSERIAL PRIMARY KEY,
    keyword             TEXT NOT NULL,
    target_asin         VARCHAR(20),
    organic_rank        SMALLINT,       -- NULL = not on first page
    sp_positions_count  SMALLINT NOT NULL DEFAULT 0,
    target_sp_rank      SMALLINT,       -- NULL = no SP ad running
    collected_at        TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX ON asin_snapshots (asin, collected_at DESC);
CREATE INDEX ON keyword_snapshots (keyword, collected_at DESC);

Step 2: Data Collector

"""
collector.py - Pangolinfo Scrape API wrapper
Handles ASIN snapshots and keyword SERP collection
"""

import os
import requests
from datetime import datetime
from dataclasses import dataclass
from typing import Optional

API_KEY = os.getenv("PANGOLINFO_API_KEY")
BASE_URL = "https://api.pangolinfo.com/v1/amazon"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}


@dataclass
class AsinSnapshot:
    asin: str
    marketplace: str
    bsr_main: Optional[int]
    bsr_category: Optional[str]
    price: Optional[float]
    review_count: Optional[int]
    rating: Optional[float]
    collected_at: str


@dataclass  
class KeywordSnapshot:
    keyword: str
    target_asin: Optional[str]
    organic_rank: Optional[int]
    sp_positions_count: int
    target_sp_rank: Optional[int]
    collected_at: str


def collect_asin(asin: str, marketplace: str = "US") -> AsinSnapshot:
    """Collect product snapshot: BSR, price, reviews, ratings."""
    resp = requests.post(
        f"{BASE_URL}/product",
        headers=HEADERS,
        json={
            "asin": asin,
            "marketplace": marketplace,
            "fields": ["bsr", "price", "reviews", "ratings"],
            "output_format": "json"
        }
    )
    resp.raise_for_status()
    d = resp.json()

    return AsinSnapshot(
        asin=asin,
        marketplace=marketplace,
        bsr_main=d.get("bsr", {}).get("main_category_rank"),
        bsr_category=d.get("bsr", {}).get("main_category_name"),
        price=d.get("price", {}).get("current"),
        review_count=d.get("reviews", {}).get("count"),
        rating=d.get("ratings", {}).get("average"),
        collected_at=datetime.utcnow().isoformat()
    )


def collect_keyword(
    keyword: str,
    target_asin: Optional[str] = None,
    marketplace: str = "US"
) -> KeywordSnapshot:
    """
    Collect keyword SERP: organic ranks + sponsored positions.

    Note: Sponsored ad position extraction is one of the more
    technically challenging aspects of Amazon scraping due to
    dynamic rendering. Pangolinfo handles this at the API level,
    so you get complete SP position data reliably.
    """
    resp = requests.post(
        f"{BASE_URL}/keyword-serp",
        headers=HEADERS,
        json={
            "keyword": keyword,
            "marketplace": marketplace,
            "include_ad_positions": True,
            "output_format": "json"
        }
    )
    resp.raise_for_status()
    d = resp.json()

    # Find target ASIN's organic position
    organic_rank = None
    if target_asin:
        for item in d.get("organic_results", []):
            if item.get("asin") == target_asin:
                organic_rank = item.get("position")
                break

    # Find target ASIN's sponsored position
    sp_positions = d.get("sponsored_positions", [])
    target_sp_rank = None
    if target_asin:
        for sp in sp_positions:
            if sp.get("asin") == target_asin:
                target_sp_rank = sp.get("position")
                break

    return KeywordSnapshot(
        keyword=keyword,
        target_asin=target_asin,
        organic_rank=organic_rank,
        sp_positions_count=len(sp_positions),
        target_sp_rank=target_sp_rank,
        collected_at=datetime.utcnow().isoformat()
    )

Step 3: Database Writer

"""
writer.py - Persist collected data to PostgreSQL
"""

import os
import psycopg2
from dataclasses import asdict
from collector import AsinSnapshot, KeywordSnapshot

conn = psycopg2.connect(os.getenv("DATABASE_URL"))


def write_asin_snapshot(snapshot: AsinSnapshot):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO asin_snapshots
                (asin, marketplace, bsr_main, bsr_category, price, review_count, rating, collected_at)
            VALUES
                (%(asin)s, %(marketplace)s, %(bsr_main)s, %(bsr_category)s,
                 %(price)s, %(review_count)s, %(rating)s, %(collected_at)s)
        """, asdict(snapshot))
    conn.commit()


def write_keyword_snapshot(snapshot: KeywordSnapshot):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO keyword_snapshots
                (keyword, target_asin, organic_rank, sp_positions_count, target_sp_rank, collected_at)
            VALUES
                (%(keyword)s, %(target_asin)s, %(organic_rank)s,
                 %(sp_positions_count)s, %(target_sp_rank)s, %(collected_at)s)
        """, asdict(snapshot))
    conn.commit()

Step 4: Scheduled Monitoring Job

"""
monitor.py - Daily monitoring routine
Run via cron: 0 */6 * * * python monitor.py
"""

import json
from collector import collect_asin, collect_keyword
from writer import write_asin_snapshot, write_keyword_snapshot

# Configure your monitoring scope here
CONFIG = {
    "target_asin": "B09XXXXXXX",
    "competitor_asins": ["B08YYYYYYY", "B07ZZZZZZZ"],
    "core_keywords": [
        "camping chair lightweight",
        "portable outdoor chair",
        "folding camp chair"
    ],
    "marketplace": "US"
}


def run_monitoring():
    all_asins = [CONFIG["target_asin"]] + CONFIG["competitor_asins"]

    print("Collecting ASIN snapshots...")
    for asin in all_asins:
        snapshot = collect_asin(asin, CONFIG["marketplace"])
        write_asin_snapshot(snapshot)
        print(f"  [{asin}] BSR: {snapshot.bsr_main} | Reviews: {snapshot.review_count}")

    print("\nCollecting keyword SERP data...")
    for keyword in CONFIG["core_keywords"]:
        snapshot = collect_keyword(keyword, CONFIG["target_asin"], CONFIG["marketplace"])
        write_keyword_snapshot(snapshot)
        print(
            f"  [{keyword}] "
            f"Organic: {snapshot.organic_rank or 'not ranked'} | "
            f"SP slots: {snapshot.sp_positions_count} | "
            f"Your SP: {snapshot.target_sp_rank or 'not running'}"
        )


if __name__ == "__main__":
    run_monitoring()

Step 5: Anomaly Detection Query

Once you have a few days of data, this query detects BSR drops that may require immediate attention:

-- Detect BSR drops > 10% vs previous collection
WITH ranked AS (
    SELECT
        asin,
        bsr_main,
        LAG(bsr_main) OVER (PARTITION BY asin ORDER BY collected_at) AS prev_bsr,
        collected_at
    FROM asin_snapshots
    WHERE collected_at >= NOW() - INTERVAL '24 hours'
),
anomalies AS (
    SELECT
        asin,
        prev_bsr,
        bsr_main AS current_bsr,
        ROUND(((bsr_main - prev_bsr)::float / prev_bsr * 100)::numeric, 1) AS change_pct,
        collected_at
    FROM ranked
    WHERE prev_bsr IS NOT NULL
      AND bsr_main > prev_bsr * 1.10  -- BSR increased (rank worsened) by 10%+
)
SELECT * FROM anomalies ORDER BY change_pct DESC;

Cross-reference any anomaly result against keyword_snapshots on the same timestamp window to see if the BSR drop correlates with a sponsored position displacement.

What You Get

After running for a few days, you'll have a queryable time-series of:

ASIN BSR trajectory (visualize in Grafana or Metabase)
Keyword organic rank changes per collection window
SP position count fluctuations on your core keywords
Your ASIN's presence/absence in sponsored positions

When BSR drops, run the anomaly query. Cross-reference with keyword data. The root cause that used to take three hours manually is now visible in one joined query.

No-Code Alternative

Building and maintaining this pipeline requires ongoing engineering time. If you want the same structured, unified view without writing code, AMZ Data Tracker provides a configuration interface for the same monitoring logic — ASIN tracking, keyword ranking, competitor monitoring — with automatic collection and a built-in dashboard.

API documentation for the Scrape API is at docs.pangolinfo.com. Free trial credits available.

Have questions about the implementation or want to share how you've built your own Amazon analytics stack? Drop a comment — happy to discuss architecture tradeoffs.

Amazon Category Opportunity Analysis: Building a Smarter Launch and Ad Decision Workflow

Mox Loop — Tue, 31 Mar 2026 03:24:46 +0000

TL;DR

Most Amazon sellers waste money because they treat category selection and ad optimization as separate problems.

A better workflow:

Evaluate demand, competition, margin, traffic structure, and execution risk together
Calculate break-even ACoS and workable CPC before launch
Use AI Agents to reduce the API learning curve
Scale only when CTR, CVR, ACoS, and organic lift align

If you're non-technical, this matters even more now because Agent workflows make API-based category research much more accessible.

The Real Problem

The classic workflow looks like this:

def old_workflow():
    choose_category_by_instinct()
    launch_listing()
    run_ads()
    hope_data_proves_you_right()

The problem is obvious: ads are being used to validate a weak market decision.

Instead, category entry should be treated like a systems problem.

You need to evaluate:

Demand strength
Supply density
Profit structure
Ad cost elasticity
Traffic structure
Execution difficulty

A Simple Launch Math Layer

Before entering a category, calculate these three numbers:

def launch_math(price, contribution_margin, expected_cvr, daily_clicks, test_days):
    break_even_acos = contribution_margin / price
    target_cpc = price * expected_cvr * break_even_acos
    test_budget = daily_clicks * target_cpc * test_days
    return break_even_acos, target_cpc, test_budget

This is not about perfect forecasting. It is about making weak categories fail on paper before they fail in your account.

Why AI Agents Change the Game

This is the part many teams still underestimate.

A few years ago, API-based public page collection required technical implementation. Today, many non-technical operators can take a Scrape API endpoint, the relevant API docs, and an API key, then hand them to an agent such as openclaw to fetch and summarize Amazon public data.

That means sellers can prototype workflows like:

{
  "task": "Collect top 100 search results for a category",
  "fields": ["asin", "price", "rating", "review_count", "ad_placement"],
  "goal": "Estimate category competition and traffic structure"
}

The edge is no longer just coding ability. The edge is the ability to define useful data tasks.

Traffic Structure Should Drive Campaign Structure

Do not use the same campaign template for every stage.

Validation

SP auto
SP broad
SP phrase

Growth

SP exact
ASIN targeting

Expansion

SP exact
SB
SD retargeting

The point is simple: traffic structure first, campaign structure second.

When to Scale and When to Cut

Scale only if:

CTR stays above average
CVR holds as volume rises
ACoS remains acceptable
TACoS does not break
Organic rank improves

Cut if:

Clicks rise but conversion falls
ACoS and TACoS worsen together
Placement inflation kills efficiency
Inventory cannot support growth

Low ACoS is not automatically good. High ACoS is not automatically bad. Repeatability matters more than vanity efficiency.

A Practical Stack

If I were building this today, I would combine:

Scrape API for public Amazon page collection
AMZ Data Tracker for trend and competitor monitoring
An Agent workflow for quick API calling and summarization
A simple scorecard for category opportunity and launch risk

That gives you a workflow that is much closer to engineering discipline than guesswork.

Final Thought

Amazon category opportunity analysis is really about one thing:

Can you enter this market profitably and scale it without losing control?

If you answer that before launch, your ad account gets smarter.

If you wait for ads to answer it for you, your budget becomes the tuition fee.

Amazon Ad Position Monitoring with Open Claw + Pangolinfo SERP API

Mox Loop — Tue, 24 Mar 2026 07:04:59 +0000

TL;DR

Stop reading your ACoS dashboard to understand competitor behavior. Build a real-time SP ad position monitor instead.

This post covers:

Async batch SERP capture via Pangolinfo API (98% SP coverage)
Tiered keyword management (A/B/C) for signal vs. noise control
Change detection: Top1 change, new Top3 entrant, Top3 exit, price drop
LLM-enriched alerts via Open Claw + Claude
Deduplication and Slack delivery

Prerequisites: Open Claw deployed, Pangolinfo API key configured (see earlier posts in this series). We're jumping straight into the ad monitoring implementation.

# Quick test: is your SERP API working?
import requests

headers = {"Authorization": "Bearer YOUR_KEY"}
payload = {
    "source": "amazon_search",
    "query": "wireless earbuds",
    "marketplace": "US",
    "include_sponsored": True,
    "include_organic": False,
    "output_format": "json"
}

resp = requests.post("https://api.pangolinfo.com/v1/serp",
                     headers=headers, json=payload)
print(resp.json()["sponsored_results"][:3])

The Problem with Existing Monitoring Tools

Your Amazon ad console shows internal metrics. Helium 10 / Jungle Scout show competitor data, but with 24-48h delays and no outbound API. When a new well-funded competitor enters your Top of Search overnight, you find out two days later when your ACoS report arrives. By then they've run 48 hours of aggressive ads at a discount price, accumulating early reviews.

What we want: automated detection of competitor ad position changes within 2 hours of them occurring, categorized by severity, routed to Slack with business context.

Step 1: Async SERP Capture

50 keywords × sequential requests = ~8 minutes per cycle. With asyncio: ~45 seconds.

import asyncio
import aiohttp
from datetime import datetime, timezone
from typing import List, Dict, Optional

PANGOLINFO_API_KEY = "your_key"

async def fetch_serp(
    keyword: str,
    marketplace: str,
    session: aiohttp.ClientSession,
    semaphore: asyncio.Semaphore
) -> Dict:
    """Capture Amazon SP ad positions for one keyword"""
    async with semaphore:
        headers = {"Authorization": f"Bearer {PANGOLINFO_API_KEY}"}
        payload = {
            "source": "amazon_search",
            "query": keyword,
            "marketplace": marketplace,
            "page": 1,
            "include_sponsored": True,
            "include_organic": False,
            "output_format": "json"
        }
        captured_at = datetime.now(timezone.utc).isoformat()

        try:
            async with session.post(
                "https://api.pangolinfo.com/v1/serp",
                headers=headers, json=payload,
                timeout=aiohttp.ClientTimeout(total=25)
            ) as resp:
                resp.raise_for_status()
                data = await resp.json()

            top_ads = sorted(
                [s for s in data.get("sponsored_results", [])
                 if "top" in s.get("ad_placement", "").lower()],
                key=lambda x: x.get("ad_rank", 999)
            )[:5]

            return {
                "keyword": keyword, "marketplace": marketplace,
                "captured_at": captured_at, "success": True,
                "top_of_search": [
                    {"rank": a.get("ad_rank"), "asin": a.get("asin"),
                     "brand": a.get("brand", ""), "price": a.get("price")}
                    for a in top_ads
                ]
            }
        except Exception as e:
            return {"keyword": keyword, "marketplace": marketplace,
                    "captured_at": captured_at,
                    "success": False, "error": str(e), "top_of_search": []}


async def batch_monitor(
    keywords: List[str],
    marketplace: str = "US",
    max_concurrent: int = 8
) -> List[Dict]:
    sem = asyncio.Semaphore(max_concurrent)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_serp(kw, marketplace, session, sem) for kw in keywords]
        return await asyncio.gather(*tasks)

Step 2: Keyword Tier Configuration

Don't monitor all keywords at the same frequency and alert threshold. It generates noise that gets ignored.

KEYWORD_CONFIG = {
    "tiers": {
        "A": {  # Core product keywords — monitor every 2h
            "keywords": ["wireless earbuds", "bluetooth speaker", "usb c hub"],
            "frequency_hours": 2,
            "alert_escalation": True   # Upgrade HIGH → CRITICAL
        },
        "B": {  # Competitive category keywords — monitor every 6h
            "keywords": ["earbuds under 30 dollars", "tws earbuds 2026"],
            "frequency_hours": 6,
            "alert_escalation": False
        },
        "C": {  # Exploratory long-tail — monitor daily, log only
            "keywords": [],
            "frequency_hours": 24,
            "alert_escalation": False,
            "suppress_alerts": True    # Record but don't notify
        }
    },
    "price_drop_threshold_pct": 12,
    "dedup_window_hours": 6
}

Step 3: Change Detection Engine

from dataclasses import dataclass, field
from enum import Enum

class AlertLevel(Enum):
    CRITICAL = "CRITICAL"
    HIGH = "HIGH"
    MEDIUM = "MEDIUM"
    INFO = "INFO"

@dataclass
class AdAlert:
    keyword: str
    level: AlertLevel
    event: str          # top1_change / new_top3 / top3_exit / price_drop
    message: str
    asin: str
    timestamp: str
    llm_context: str = ""


def detect_changes(
    current: Dict,
    baseline: Optional[Dict],
    tier_config: Dict
) -> List[AdAlert]:
    """
    Compare SERP snapshots. Returns [] on first run (baseline only).
    """
    if baseline is None:
        return []

    kw = current["keyword"]
    ts = current["captured_at"]
    tier = tier_config.get("tier", "B")
    suppress = tier_config.get("suppress_alerts", False)
    escalate = tier_config.get("alert_escalation", False)
    threshold = KEYWORD_CONFIG["price_drop_threshold_pct"] / 100

    curr_top = [a["asin"] for a in current.get("top_of_search", [])[:3]]
    base_top = [a["asin"] for a in baseline.get("top_of_search", [])[:3]]
    curr_prices = {a["asin"]: a.get("price")
                   for a in current.get("top_of_search", [])[:3]}
    base_prices = {a["asin"]: a.get("price")
                   for a in baseline.get("top_of_search", [])[:3]}

    def resolve(base: AlertLevel) -> AlertLevel:
        if suppress:
            return AlertLevel.INFO
        if escalate and base == AlertLevel.HIGH:
            return AlertLevel.CRITICAL
        return base

    alerts = []

    # Top #1 changed
    if curr_top and base_top and curr_top[0] != base_top[0]:
        is_new = curr_top[0] not in base_top
        alerts.append(AdAlert(
            keyword=kw,
            level=resolve(AlertLevel.CRITICAL if is_new else AlertLevel.HIGH),
            event="top1_change",
            message=(f"Top of Search #1: {base_top[0]} → {curr_top[0]}"
                     + (" [NEW ENTRANT]" if is_new else "")),
            asin=curr_top[0], timestamp=ts
        ))

    # New Top 3 entrant (excluding #1 already handled)
    for asin in set(curr_top) - set(base_top) - {curr_top[0] if curr_top else ""}:
        rank = next((a["rank"] for a in current["top_of_search"]
                     if a["asin"] == asin), -1)
        alerts.append(AdAlert(
            keyword=kw, level=resolve(AlertLevel.HIGH),
            event="new_top3",
            message=f"New entrant at Top 3 position #{rank}: {asin}",
            asin=asin, timestamp=ts
        ))

    # Top 3 exit
    for asin in set(base_top) - set(curr_top):
        alerts.append(AdAlert(
            keyword=kw, level=resolve(AlertLevel.MEDIUM),
            event="top3_exit",
            message=f"{asin} exited Top 3 — possible budget reduction",
            asin=asin, timestamp=ts
        ))

    # Price drop in Top 3
    for asin in set(curr_top) & set(base_top):
        old, new = base_prices.get(asin), curr_prices.get(asin)
        if old and new and old > 0 and (old - new) / old >= threshold:
            drop = (old - new) / old * 100
            alerts.append(AdAlert(
                keyword=kw, level=resolve(AlertLevel.HIGH),
                event="price_drop",
                message=f"{asin} price drop {drop:.1f}%: ${old:.2f} → ${new:.2f}",
                asin=asin, timestamp=ts
            ))

    return alerts

Step 4: LLM Context Enrichment (Open Claw Layer)

from anthropic import Anthropic

_llm = Anthropic()

def add_llm_context(alert: AdAlert, snapshot: Dict) -> AdAlert:
    """Enrich CRITICAL/HIGH alerts with Claude business analysis"""
    if alert.level not in (AlertLevel.CRITICAL, AlertLevel.HIGH):
        return alert

    top3 = "\n".join([
        f"  #{a['rank']}: {a['asin']} | {a['brand']} | ${a['price'] or 'N/A'}"
        for a in snapshot.get("top_of_search", [])[:3]
    ])

    resp = _llm.messages.create(
        model="claude-3-7-sonnet-20250219",
        max_tokens=180,
        messages=[{"role": "user", "content":
            f"Amazon ad monitoring alert — keyword: '{alert.keyword}'\n"
            f"Event: {alert.message}\n"
            f"Current Top 3:\n{top3}\n\n"
            "In 70 words or fewer: likely cause + what the PPC team should check. "
            "Direct, specific, no preamble."
        }]
    )
    alert.llm_context = resp.content[0].text
    return alert

Step 5: Slack Delivery with Deduplication

import sqlite3
import aiohttp
from datetime import timedelta

DB_PATH = "/data/ad_monitor.db"
SLACK_WEBHOOK = "https://hooks.slack.com/services/YOUR/WEBHOOK"


def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.executescript("""
            CREATE TABLE IF NOT EXISTS snapshots (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                keyword TEXT, marketplace TEXT,
                captured_at TEXT, data_json TEXT
            );
            CREATE TABLE IF NOT EXISTS sent_alerts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                keyword TEXT, asin TEXT, event TEXT,
                triggered_at TEXT
            );
        """)


def is_duplicate(alert: AdAlert, window_hours: int = 6) -> bool:
    cutoff = (datetime.utcnow() - timedelta(hours=window_hours)).isoformat()
    with sqlite3.connect(DB_PATH) as conn:
        return bool(conn.execute(
            "SELECT 1 FROM sent_alerts WHERE keyword=? AND asin=? AND event=?"
            " AND triggered_at>? LIMIT 1",
            (alert.keyword, alert.asin, alert.event, cutoff)
        ).fetchone())


def record_alert(alert: AdAlert):
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute(
            "INSERT INTO sent_alerts (keyword, asin, event, triggered_at)"
            " VALUES (?,?,?,?)",
            (alert.keyword, alert.asin, alert.event, alert.timestamp)
        )


async def send_to_slack(alert: AdAlert):
    emoji = {"CRITICAL": "🚨", "HIGH": "⚠️", "MEDIUM": "📊"}.get(
        alert.level.value, "ℹ️"
    )
    text = (
        f"{emoji} *[Ad Monitor]* `{alert.level.value}` — "
        f"`{alert.keyword}`\n"
        f"{alert.message}\n"
        f"ASIN: `{alert.asin}` | {alert.timestamp[:16]} UTC"
        + (f"\n> _{alert.llm_context}_" if alert.llm_context else "")
    )
    async with aiohttp.ClientSession() as session:
        await session.post(SLACK_WEBHOOK, json={"text": text},
                           timeout=aiohttp.ClientTimeout(total=10))

Step 6: Wiring It All Together

import json

async def run_monitoring_cycle(tier: str, marketplace: str = "US"):
    """One complete monitoring run for a keyword tier"""
    tier_cfg = KEYWORD_CONFIG["tiers"][tier]
    keywords = tier_cfg["keywords"]
    if not keywords:
        return

    snapshots = await batch_monitor(keywords, marketplace)

    for snap in snapshots:
        if not snap.get("success"):
            continue

        # Load baseline (second-to-last snapshot)
        with sqlite3.connect(DB_PATH) as conn:
            row = conn.execute(
                "SELECT data_json FROM snapshots WHERE keyword=? AND marketplace=?"
                " ORDER BY captured_at DESC LIMIT 1 OFFSET 1",
                (snap["keyword"], marketplace)
            ).fetchone()
            baseline = json.loads(row[0]) if row else None

            # Save current snapshot
            conn.execute(
                "INSERT INTO snapshots (keyword, marketplace, captured_at, data_json)"
                " VALUES (?,?,?,?)",
                (snap["keyword"], marketplace, snap["captured_at"], json.dumps(snap))
            )

        # Detect + dispatch
        alerts = detect_changes(snap, baseline, tier_cfg)
        for alert in alerts:
            if not is_duplicate(alert, KEYWORD_CONFIG["dedup_window_hours"]):
                alert = add_llm_context(alert, snap)
                await send_to_slack(alert)
                record_alert(alert)


# Entry point — call from APScheduler or cron
async def main():
    init_db()
    await run_monitoring_cycle("A")   # Tier A every 2h
    # await run_monitoring_cycle("B") # Tier B every 6h

if __name__ == "__main__":
    asyncio.run(main())

Production Notes

Database scaling: SQLite works fine up to ~100 keywords at 2h frequency. Beyond that, switch to PostgreSQL — SQLite's write lock becomes a bottleneck under concurrent writes.

Rate limiting: If you hit HTTP 429, add exponential backoff (start 2s, max 30s) around the session.post call in fetch_serp.

Monitoring the monitor: Set up a heartbeat check — if no rows appear in snapshots for the expected time window, alert your on-call channel. Silent failures in monitoring systems are the worst kind.

Data retention: Keep 90 days of snapshots. At 30 days you see monthly patterns; at 90 days you start recognizing seasonal behaviors and pre-promotion signals.

Open Claw for Cross-Border E-Commerce: 5 High-Value Use Cases (With Code)

Mox Loop — Fri, 20 Mar 2026 10:02:52 +0000

TL;DR

Open Claw cross-border e-commerce deployments that ship to production share one thing: they connect LLM reasoning to real-time data APIs rather than static knowledge. This post covers the 5 use cases that consistently deliver measurable ROI, with implementation patterns for each.

Data layer: Pangolinfo Scrape API — structured Amazon/Walmart/Shopee data, minute-level freshness, JSON output.

Why Cross-Border E-Commerce Is a Natural Fit for AI Agents

E-commerce data workflows have three properties that make them ideal for agent-based automation:

High volume, low variance: Most seller data tasks are repetitive queries over structured data—perfect for tool-calling agents
Time sensitivity: Price and BSR data can shift meaningfully within hours; the latency cost of manual workflows is measurable in revenue
Multi-source synthesis: Real decisions require combining data from multiple endpoints—exactly what LLM-powered agents excel at

The fatal flaw in most "AI for e-commerce" implementations: using an LLM to answer questions about current market conditions from training data. Amazon BSR rankings and pricing change hourly. No static model can reliably answer questions about live market state.

The right pattern: LLM for reasoning, external APIs for data retrieval.

Use Case 1: Competitor Price and BSR Monitoring

The Problem

Manual morning spreadsheet reviews create 12-24 hour lag in detecting competitor pricing moves. During promotional events, that lag costs real revenue.

The Architecture

Cron Trigger (every 4 hours)
  → Open Claw Monitor Agent
  → Tool: get_product_realtime (Pangolinfo Scrape API)
  → AlertEngine: compare vs. stored baseline
  → Slack/webhook notification if threshold exceeded
  → BaselineStore: update stored values

Core Implementation

import asyncio
import aiohttp
from typing import Optional

PANGOLINFO_API_KEY = "your_key_here"
API_BASE = "https://api.pangolinfo.com/v1"

async def fetch_asin_realtime(session: aiohttp.ClientSession, 
                               asin: str,
                               marketplace: str = "US") -> dict:
    """
    Fetch real-time Amazon product data via Pangolinfo Scrape API.
    Registered as Open Claw MCP Tool for agent use.
    """
    headers = {"Authorization": f"Bearer {PANGOLINFO_API_KEY}"}
    payload = {
        "source": "amazon_product",
        "asin": asin,
        "marketplace": marketplace,
        "fields": [
            "price", "original_price", "bsr", "bsr_category",
            "rating", "review_count", "availability",
            "seller_type", "deal_type"
        ],
        "output_format": "json"
    }

    async with session.post(
        f"{API_BASE}/scrape",
        headers=headers,
        json=payload,
        timeout=aiohttp.ClientTimeout(total=30)
    ) as resp:
        data = await resp.json()
        return {
            "asin": asin,
            "price": data.get("price"),
            "original_price": data.get("original_price"),
            "bsr": data.get("bsr"),
            "rating": data.get("rating"),
            "review_count": data.get("review_count"),
            "availability": data.get("availability"),
            "active_deal": data.get("deal_type")
        }


def detect_price_alerts(current: dict, 
                          baseline: Optional[dict],
                          price_drop_threshold: float = 12.0,
                          bsr_spike_threshold: int = 300) -> list:
    """Compare current data against baseline, return alerts."""
    alerts = []

    if not baseline:
        return alerts

    # Price drop detection
    curr_price = current.get("price")
    base_price = baseline.get("price")

    if curr_price and base_price and base_price > 0:
        drop_pct = (base_price - curr_price) / base_price * 100
        if drop_pct >= price_drop_threshold:
            alerts.append({
                "type": "price_drop", 
                "severity": "HIGH",
                "message": f"Price dropped {drop_pct:.1f}%: ${base_price:.2f} → ${curr_price:.2f}"
            })

    # BSR improvement detection
    curr_bsr = current.get("bsr")
    base_bsr = baseline.get("bsr")

    if curr_bsr and base_bsr and base_bsr > curr_bsr:
        improvement = base_bsr - curr_bsr
        if improvement >= bsr_spike_threshold:
            alerts.append({
                "type": "bsr_spike",
                "severity": "MEDIUM", 
                "message": f"BSR improved {improvement} positions: {base_bsr} → {curr_bsr}"
            })

    # Active promotion detection
    if current.get("active_deal"):
        alerts.append({
            "type": "promotion_detected",
            "severity": "HIGH",
            "message": f"Active deal: {current['active_deal']}"
        })

    return alerts


async def run_monitoring_cycle(asin_list: list, 
                                baseline_store: dict) -> list:
    """Run one full monitoring cycle for all watchlist ASINs."""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_asin_realtime(session, asin) for asin in asin_list]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    all_alerts = []
    for result in results:
        if isinstance(result, Exception):
            continue

        asin = result["asin"]
        baseline = baseline_store.get(asin)
        alerts = detect_price_alerts(result, baseline)

        if alerts:
            all_alerts.extend([{**a, "asin": asin} for a in alerts])

        # Update baseline
        baseline_store[asin] = result

    return all_alerts


# Open Claw MCP Tool Definition
MONITOR_TOOL_SCHEMA = {
    "name": "check_competitor_asin",
    "description": (
        "Fetch real-time data for an Amazon ASIN: price, BSR rank, rating, "
        "review count, availability, and active promotions. "
        "Use when user asks about current competitor status. "
        "Do NOT use for historical trends (use get_price_history tool instead)."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "asin": {
                "type": "string",
                "description": "Amazon ASIN, 10 alphanumeric chars, e.g. B08N5WRWNW"
            },
            "marketplace": {
                "type": "string",
                "enum": ["US", "UK", "DE", "JP", "CA", "FR", "ES", "IT", "AU"],
                "default": "US"
            }
        },
        "required": ["asin"]
    }
}

Key Design Notes

Semaphore for rate limiting: asyncio.Semaphore(8) in front of API calls prevents burst overload
Baseline isolation: Store per-ASIN; don't update on error responses
Alert deduplication: Track last alert timestamp per ASIN/type to prevent notification spam

Use Case 2: Product Research Pipeline

from anthropic import Anthropic
from concurrent.futures import ThreadPoolExecutor

client = Anthropic()

def research_category_opportunities(
    category: str,
    criteria: dict,  # {"price_min": 25, "price_max": 60, "review_count_max": 200}
    scan_count: int = 80
) -> str:
    """
    Multi-step product research pipeline:
    1. Pull new releases bestseller list
    2. Batch filter against criteria (parallel)
    3. Deep fetch candidates
    4. LLM opportunity analysis
    """
    # Step 1: Get bestseller list
    new_releases = get_new_releases(category, scan_count)
    asin_list = [item["asin"] for item in new_releases]

    # Step 2: Parallel batch fetch + filter
    with ThreadPoolExecutor(max_workers=8) as executor:
        product_data = list(executor.map(
            lambda a: get_product_data(a), asin_list
        ))

    candidates = [
        p for p in product_data
        if (criteria.get("price_min", 0) <= (p.get("price") or 0) <= criteria.get("price_max", 9999))
        and ((p.get("review_count") or 0) <= criteria.get("review_count_max", 9999))
    ]

    # Step 3: LLM analysis
    candidates_json = json.dumps(candidates[:15], indent=2)  # Cap at 15

    response = client.messages.create(
        model="claude-3-7-sonnet-20250219",
        max_tokens=3000,
        messages=[{"role": "user", "content": f"""
        Product research analysis for category: {category}
        Criteria applied: {json.dumps(criteria)}

        Candidate products data:
        {candidates_json}

        For each candidate, provide:
        - Opportunity score (1-10) with rationale
        - Key competitive differentiators needed
        - Main risk factors
        - Recommendation: pursue / monitor / skip

        Rank by opportunity score.
        """}]
    )

    return response.content[0].text

Use Case 3: Review Intelligence

def run_review_analysis(asin: str, category: str, 
                         max_reviews: int = 150) -> dict:
    """
    Complete review intelligence pipeline.
    Uses Pangolinfo Reviews Scraper API.
    """
    # Fetch reviews (sorted by helpfulness for signal quality)
    headers = {"Authorization": f"Bearer {PANGOLINFO_API_KEY}"}
    params = {
        "asin": asin,
        "marketplace": "US",
        "max_reviews": max_reviews,
        "sort_by": "helpful",
        "output_format": "json"
    }

    resp = requests.get(
        "https://api.pangolinfo.com/v1/reviews",
        headers=headers, params=params
    )
    reviews = resp.json().get("reviews", [])

    # Split by sentiment
    negative = [r for r in reviews if r.get("rating", 5) <= 3]
    positive = [r for r in reviews if r.get("rating", 5) >= 4]

    # Preprocess: extract signal, limit tokens
    def format_reviews(review_list: list, limit: int = 60) -> str:
        return "\n".join([
            f"[{r.get('rating')}★|helpful:{r.get('helpful_votes',0)}] "
            f"{r.get('title','')}: {str(r.get('body',''))[:200]}"
            for r in sorted(review_list, key=lambda x: -x.get('helpful_votes', 0))[:limit]
        ])

    neg_text = format_reviews(negative)
    pos_text = format_reviews(positive, 40)

    # LLM structured analysis
    response = client.messages.create(
        model="claude-3-7-sonnet-20250219",
        max_tokens=2500,
        messages=[{"role": "user", "content": f"""
        Competitive review intelligence for Amazon {category} product (ASIN: {asin})

        NEGATIVE REVIEWS ({len(negative)} analyzed):
        {neg_text}

        POSITIVE REVIEWS ({len(positive)} analyzed):
        {pos_text}

        Output structured JSON:
        {{
            "pain_points": [
                {{"topic": str, "frequency_pct": int, "severity": 1-5, "sample_quotes": [str]}}
            ],
            "value_drivers": [
                {{"topic": str, "frequency_pct": int, "buyer_insight": str}}
            ],
            "product_improvements": [
                {{"action": str, "addresses": str, "difficulty": "Easy|Medium|Hard"}}
            ],
            "listing_bullets": [str],
            "risk_signals": [str]
        }}
        Output JSON only, no other text.
        """}]
    )

    import json as json_lib
    try:
        analysis = json_lib.loads(response.content[0].text)
    except:
        analysis = {"raw": response.content[0].text}

    return {
        "asin": asin,
        "reviews_analyzed": {"negative": len(negative), "positive": len(positive)},
        "analysis": analysis
    }

Performance Numbers from Production Deployments

Workflow	Manual Time	Agent Time	Improvement
30 ASIN daily monitoring	45 min/day	30 sec/query	~97%
Category product research	2-3 days	2-4 hours	~80%
Competitor review analysis	3-4 hours	2-5 min	~95%
Multi-platform inventory summary	60-90 min	<2 min	~95%

Common Pitfalls to Avoid

Poor tool descriptions: The LLM's tool selection accuracy is entirely dependent on description quality. More detail = better routing.

Sending raw API payloads to LLM: Always preprocess. Extract relevant fields, truncate body text, sort by signal quality. This is non-negotiable for staying within context windows.

No retry logic: Build exponential backoff from day one. Rate limits and timeouts are inevitable at scale.

Single-agent doing everything: Narrow tool access per agent consistently outperforms one agent with access to all tools.

Resources

Amazon Scraper API Benchmark: 12M Requests Across 4 Platforms — What the Data Actually Shows

Mox Loop — Tue, 17 Mar 2026 06:34:44 +0000

Amazon Scraper API Benchmark: 12M Requests Across 4 Platforms — What the Data Actually Shows

TL;DR

Self-built scrapers: 71.4% product page success rate at scale, 60% of engineering time on anti-bot maintenance
Competitor A: 89.1% product pages, 81.2% SP ad slots, 8,900ms P99 — workable but with real blind spots
Pangolinfo: 98.6% product pages, 97.3% SP ad slots, 3,890ms P99, full Customer Says extraction
Cost delta: ~¥27,500/month savings at 100K pages/day vs self-built
Key limitation: Non-Amazon platform (Walmart/Shopee) maturity gaps, English docs update faster than Chinese

Full benchmark methodology and data below. Code examples included.

Why I Ran This Test

Our team had been running a self-built Amazon scraping infrastructure for about eight months when the maintenance burden became impossible to ignore. Not because anything was catastrophically broken — but because the engineering economics had quietly inverted.

Amazon's anti-bot infrastructure in 2025 is not the problem it was in 2022. Behavioral fingerprinting, CAPTCHA rotation, JavaScript rendering detection, headless browser identification — each layer requires dedicated engineering to counter, and Amazon ships updates to these systems constantly. We were spending 60% of engineering hours on anti-scraping countermeasures. The remaining 40% went to actual business logic. Forrester Research's 2024 benchmark puts average self-built scraper maintenance at 40–60 hours per month; we were running hot relative to that baseline.

The decision to evaluate commercial Amazon scraper API alternatives was ultimately an engineering economics decision, not a technical one.

Test Setup

Three systems ran simultaneously across the full 60-day period: Pangolinfo Scrape API, Competitor A (commercial reasons preclude naming), and our self-built infrastructure.

All requests were production traffic — real business data needs, not synthetic benchmark loads. Coverage:

Amazon product detail pages (ASIN-level)
Search results pages (keyword + category)
BSR list pages (Best Sellers, New Releases, Movers & Shakers)
Sponsored Products ad slot data
Review pages including Customer Says AI summary module
Platforms: Amazon US, UK, Japan; Walmart US

Primary metrics: collection success rate, P50–P99 latency, JSON output completeness, SP ad slot capture rate, Customer Says field integrity.

Results

Success Rates

60-day averages:

results = {
    "product_detail_page": {
        "pangolinfo": 98.6,
        "competitor_a": 89.1,
        "self_built": 71.4,  # drops lower during CAPTCHA events
    },
    "search_results_page": {
        "pangolinfo": 97.2,
        "competitor_a": 84.3,
        "self_built": 62.8,
    },
    "review_pages": {
        "pangolinfo": 96.8,
        "competitor_a": 79.6,
        "self_built": 55.1,
    },
    "sp_ad_slots": {
        "pangolinfo": 97.3,
        "competitor_a": 81.2,
        "self_built": 38.4,  # multiple CAPTCHA-triggered outages in this period
    },
    "bsr_list_pages": {
        "pangolinfo": 98.1,
        "competitor_a": 86.7,
        "self_built": 65.2,
    }
}

The self-built ad slot number (38.4%) includes several CAPTCHA-triggered outage windows. Stable-period average was approximately 55%. The distinction matters because "outage events are operational reality, not edge cases" — so 60-day average is the more honest metric.

Response Latency (Amazon Product Detail Pages, ms)

latency_ms = {
    "P50":  {"pangolinfo": 890,   "competitor_a": 1450},
    "P75":  {"pangolinfo": 1240,  "competitor_a": 2100},
    "P90":  {"pangolinfo": 1780,  "competitor_a": 3200},
    "P95":  {"pangolinfo": 2340,  "competitor_a": 4800},
    "P99":  {"pangolinfo": 3890,  "competitor_a": 8900},
}

P50 gap (890ms vs 1450ms) is real but not operationally decisive for most workflows. P99 gap (3,890ms vs 8,900ms) is the constraining number — it defines what SLA you can responsibly promise for time-sensitive workflows. Pangolinfo's ceiling is less than half Competitor A's.

SP Ad Slot Capture — The Key Differentiator

100K dedicated SP ad slot requests. Results:

Pangolinfo: 97.3% (official claim 98%; 0.7pp gap, within expected variance)
Competitor A: 81.2%
Difference: 16.1 percentage points

Practical translation: monitoring 500 keywords daily, Competitor A's feed is missing ~81 keyword ad slot positions every day. Whether those happen to be the keywords where your competitor is making a strategic push is unknowable — which is the point.

Deep Dive: Customer Says Extraction

Amazon's Customer Says module is an AI-generated review summary that condenses product feedback into structured positive/negative highlights. It's high-information-density data for competitive product positioning. It's also technically difficult to extract reliably.

Technical challenge layers:
- Layer 1 (static HTML): Most scrapers handle this fine
- Layer 2 (JS-rendered content): Requires full browser rendering
- Layer 3 (Customer Says — dynamic conditional loading):
    - Different load triggers per ASIN/category characteristics
    - Amazon-specific protection for this module
    - Structure varies by content type

Pangolinfo Reviews Scraper API: Layer 3 ✓ (complete, stable)
Competitor A: Layer 2 only (Customer Says score: 5/10)
Self-built: Layer 1–2 (Layer 3: occasional partial returns, unreliable)

Working example with Pangolinfo Reviews Scraper API:

import requests
from typing import Optional

def fetch_reviews_with_customer_says(
    asin: str,
    api_key: str,
    marketplace: str = "US",
    star_filter: Optional[list] = None,
    count: int = 20
) -> dict:
    """
    Fetch reviews + Customer Says summary via Pangolinfo Reviews Scraper API.
    Customer Says = Amazon's AI-generated review summary module.
    Competitor A scored 5/10 on this capability in independent testing.
    """
    payload = {
        "asin": asin,
        "marketplace": marketplace,
        "sort": "recent",
        "count": count,
        "include_customer_says": True,  # Key parameter for the AI summary
        "fields": [
            "rating",
            "review_text",
            "review_title",
            "verified_purchase",
            "date",
            "helpful_votes"
        ]
    }

    if star_filter:
        payload["star_filter"] = star_filter  # e.g., [1, 2] for negative reviews only

    response = requests.post(
        "https://api.pangolinfo.com/v1/amazon/reviews",
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        },
        json=payload,
        timeout=30
    )
    response.raise_for_status()
    data = response.json()

    return {
        # AI-generated summary — the high-value field
        "customer_says": {
            "positive": data.get("customer_says", {}).get("positive", ""),
            "negative": data.get("customer_says", {}).get("negative", ""),
        },
        "reviews": data.get("reviews", []),
        "total_count": data.get("total_review_count"),
        "average_rating": data.get("average_rating")
    }


# Usage: fetch negative reviews + Customer Says for competitor ASIN
result = fetch_reviews_with_customer_says(
    asin="B07EXAMPLE1",
    api_key="YOUR_PANGOLINFO_KEY",
    star_filter=[1, 2],  # Only 1-2 star reviews
    count=50
)

print("Customer Says Positive:", result["customer_says"]["positive"])
# Example: "Customers appreciate the durable build quality and simple setup process."

print("Customer Says Negative:", result["customer_says"]["negative"])  
# Example: "Some customers report size inconsistencies and slow customer support response."

# These two sentences compress dozens of reviews into actionable competitive intelligence

ZIP-Level Collection: Regional Pricing Intelligence

Amazon's Prime delivery zone pricing creates regional variation that cross-border sellers routinely underestimate. Same ASIN, different ZIP codes, potentially different prices, different Prime eligibility states, different delivery estimate messaging.

def compare_regional_pricing(
    asin: str,
    zip_codes: list,
    api_key: str
) -> dict:
    """
    Simulate what shoppers in different regions see for the same product.
    Pangolinfo supports ZIP-level differentiated collection.
    Competitor A: 5/10 support. Self-built: essentially impossible at scale.
    """
    results = {}

    for zip_code in zip_codes:
        response = requests.post(
            "https://api.pangolinfo.com/v1/amazon/product",
            headers={"Authorization": f"Bearer {api_key}"},
            json={
                "asin": asin,
                "marketplace": "US",
                "zip_code": zip_code,
                "fields": ["price", "prime_eligibility", "delivery_estimate", "availability"]
            },
            timeout=30
        )
        data = response.json()
        results[zip_code] = {
            "price": data.get("price", {}).get("current"),
            "prime": data.get("prime_eligibility"),
            "delivery": data.get("delivery_estimate"),
            "in_stock": data.get("availability") == "In Stock"
        }

    return results


# Compare New York (10001) vs Los Angeles (90001) vs Chicago (60601)
regional_data = compare_regional_pricing(
    asin="B07EXAMPLE1",
    zip_codes=["10001", "90001", "60601"],
    api_key="YOUR_KEY"
)

for zip_code, data in regional_data.items():
    print(f"ZIP {zip_code}: ${data['price']} | Prime: {data['prime']} | {data['delivery']}")

Cost Model

At 100K pages/day:

monthly_costs = {
    "self_built": {
        "servers": 8_000,         # RMB
        "proxy_ips": 12_000,
        "engineering_hrs": 18_000,  # 60h × ¥300/hr
        "emergency_fixes": 4_000,
        "total": 42_000
    },
    "pangolinfo": {
        "api_fee": 8_500,          # estimated from pricing tiers
        "engineering_ongoing": 0,  # near-zero after initial integration
        "total": 8_500
    },
    "monthly_savings": 33_500
}
# 3-year TCO differential: ~¥1.2M

Honest Limitations

Documentation parity: English API Reference updates faster than Chinese. Chinese-language technical teams should monitor both versions.

Non-Amazon platform maturity: Walmart and Shopee parsing templates have measurable gaps vs Amazon coverage. Specific field availability on Walmart varies by SKU type.

Peak concurrency: At 8M+ pages/day in pressure testing, ~3.2% request queue delays appeared. Imperceptible for normal business scenarios; needs pre-negotiation if you're building financial-grade real-time data infrastructure.

Resources

Pangolinfo Amazon Scraper API — free trial available
Reviews Scraper API — Customer Says support
AMZ Data Tracker — no-code monitoring
API Documentation
Console (trial)

Have you run your own Amazon scraper API comparisons? I'm particularly curious about Walmart-focused use cases where the Pangolinfo maturity gap I noted might be more or less pronounced than I found. Drop your experience in the comments.

api #python #ecommerce #datascraping #amazonsellertools

Stop Maintaining Your Amazon Scraper — Use a Commercial API Instead (With Full Code)

Mox Loop — Mon, 16 Mar 2026 01:49:02 +0000

Stop Maintaining Your Amazon Scraper — Use a Commercial API Instead (With Full Code)

TL;DR

AI tools make writing an Amazon scraper a 5-minute task. But the actual production cost of a self-built scraper at meaningful scale is $5,000-$10,000/month when you count honestly — proxies, servers, and engineer maintenance time. This post explains why commercial Amazon Scraper APIs solve problems AI can't, with complete working Python code.

The Problem: AI Wins at Writing, Loses at Running

If you're reading this, you've probably already used Claude or GPT to generate an Amazon scraper. It worked. You were impressed. Then something went wrong — IP bans, broken selectors after a page structure change, or worse: silent honeypot responses returning HTTP 200 with fake data your pipeline happily ingested.

Here are the three things no AI-generated scraper handles out of the box:

1. Scale beyond casual use

Amazon's bot-detection is a multi-signal ML model evaluating TLS fingerprints, request cadence, behavioral sequences, and cookie state. A Python requests script doing 10,000+ requests/day triggers it reliably. You need residential IP rotation (minimum ~$800-$2,000/month for quality pools), concurrency management with adaptive rate limiting, and infrastructure to detect silent failures.

2. Parsing stability over time

Amazon A/B tests constantly. The same ASIN page has structurally different HTML across user segments and testing cohorts. Hard-coded CSS selectors fail silently — not with exceptions, but with null values. At 200,000 ASINs/day, you won't notice for days.

3. JavaScript-rendered content

Sponsored Products ad slots, "Customer Says" (Amazon's AI review summary), and various pricing configurations load via client-side JavaScript long after the HTML response. Standard HTTP scrapers miss this entirely. The fields don't appear empty — they don't appear at all.

The Commercial API Approach: What It Actually Solves

I've been running Pangolinfo's Amazon Scraper API alongside self-built infrastructure for comparison. Three things stand out:

98% SP ad slot capture rate. Most self-built solutions hit 40-60% at best. For competitive advertising intelligence, this gap is the difference between a product and a prototype.

Native "Customer Says" support. This JS-rendered block is structurally inaccessible to standard scrapers. Pangolinfo returns it natively with render_js: true.

Zero parsing maintenance. Amazon updates its HTML roughly once per quarter with breaking changes. Pangolinfo's parsing templates update same-day on their side. Your pipeline doesn't break.

Full Working Code: Self-Built vs. Commercial API Comparison

Option A: Self-Built (What AI Gives You — With the Gaps Exposed)

import requests
from bs4 import BeautifulSoup
import time
import random

# PROBLEMS WITH THIS APPROACH:
# 1. No TLS fingerprint masking — flagged by Amazon's bot detection
# 2. Hard-coded selectors — breaks when Amazon A/B tests change HTML
# 3. No proxy rotation — IP gets banned at scale
# 4. Can't access JS-rendered content (SP ad slots, Customer Says)
# 5. Silent failures — you won't know when it breaks

def basic_amazon_scraper(asin: str) -> dict:
    """
    AI-generated scraper — works at small scale, breaks in production.
    This is what you get from 5 minutes with ChatGPT.
    """
    url = f"https://www.amazon.com/dp/{asin}"

    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...",
        # This user-agent string is flagged by Amazon's ML model.
        # It doesn't match actual browser TLS fingerprints.
    }

    # Problem: This IP will get banned. No rotation.
    response = requests.get(url, headers=headers, timeout=15)

    if response.status_code != 200:
        return {"error": f"Status {response.status_code}"}

    # Problem: This selector works today. Breaks when Amazon A/B tests.
    # You'll get null values with no exception — silent data corruption.
    soup = BeautifulSoup(response.content, 'html.parser')

    price = soup.select_one('span.a-price-whole')  # Fragile selector
    title = soup.select_one('#productTitle')

    return {
        "asin": asin,
        "title": title.text.strip() if title else None,
        "price": price.text.strip() if price else None,
        "customer_says": None,  # NOT AVAILABLE — JS-rendered, inaccessible
        "sp_ad_slots": [],      # NOT AVAILABLE — JS-rendered, inaccessible
    }

Option B: Pangolinfo Commercial Amazon Scraper API

import requests
import asyncio
import aiohttp
import json
from typing import Optional, List, Dict, Any
from dataclasses import dataclass

@dataclass
class ScrapeConfig:
    """Configuration for Amazon scraping requests"""
    marketplace: str = "US"
    output_format: str = "json"  # json | html | markdown (markdown for LLM pipelines)
    render_js: bool = True       # Required for SP ad slots and Customer Says
    zip_code: Optional[str] = None  # For location-specific pricing

class PangolinAmazonScraper:
    """
    Production-grade Amazon data collection via Pangolinfo's commercial API.

    Handles at the infrastructure level (no code required from you):
    - TLS fingerprint masking
    - Residential IP rotation and anti-detection
    - JavaScript rendering for dynamic content
    - Parsing template maintenance across Amazon page updates
    - 98% SP ad slot capture rate

    Docs: https://docs.pangolinfo.com/en-api-reference/universalApi/universalApi
    """

    BASE_URL = "https://api.pangolinfo.com/v1/scrape"

    def __init__(self, api_key: str):
        self.api_key = api_key
        self._headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def _build_payload(self, request_type: str, config: ScrapeConfig, **kwargs) -> dict:
        payload = {
            "platform": "amazon",
            "type": request_type,
            "marketplace": config.marketplace,
            "output_format": config.output_format,
            "render_js": config.render_js,
            **kwargs
        }
        if config.zip_code:
            payload["zip_code"] = config.zip_code
        return payload

    def get_product(self, asin: str, config: Optional[ScrapeConfig] = None) -> dict:
        """
        Fetch Amazon product detail page.

        Returns structured data including:
        - title, price, images, bullet points
        - BSR (Best Seller Rank) with category
        - rating and review count
        - customer_says (Amazon's AI review summary — JS-rendered)
        - sp_ad_slots (Sponsored Products at 98% capture rate)
        - a_plus_content, seller information

        With zip_code: returns location-specific pricing and delivery estimates
        """
        config = config or ScrapeConfig()
        payload = self._build_payload("product", config, asin=asin)

        response = requests.post(
            self.BASE_URL, json=payload, headers=self._headers, timeout=30
        )
        response.raise_for_status()
        return response.json()

    def get_bestsellers(
        self,
        category_id: str,
        config: Optional[ScrapeConfig] = None,
        page: int = 1
    ) -> dict:
        """Fetch Best Seller rankings for an Amazon category node."""
        config = config or ScrapeConfig()
        payload = self._build_payload(
            "bestseller", config, category_id=category_id, page=page
        )
        response = requests.post(
            self.BASE_URL, json=payload, headers=self._headers, timeout=30
        )
        response.raise_for_status()
        return response.json()

    def get_keyword_results(
        self,
        keyword: str,
        config: Optional[ScrapeConfig] = None,
        page: int = 1
    ) -> dict:
        """
        Fetch keyword search results.
        With render_js=True: captures SP ad slot data at 98% accuracy.
        """
        config = config or ScrapeConfig()
        payload = self._build_payload(
            "keyword", config, keyword=keyword, page=page
        )
        response = requests.post(
            self.BASE_URL, json=payload, headers=self._headers, timeout=30
        )
        response.raise_for_status()
        return response.json()

    async def batch_get_products(
        self,
        asin_list: List[str],
        config: Optional[ScrapeConfig] = None,
        concurrency: int = 10
    ) -> List[Dict[str, Any]]:
        """
        Asynchronous batch product fetch with concurrency control.
        Set concurrency based on your Pangolinfo plan's concurrent request limit.
        """
        config = config or ScrapeConfig()
        semaphore = asyncio.Semaphore(concurrency)

        async def fetch_one(session: aiohttp.ClientSession, asin: str) -> dict:
            async with semaphore:
                payload = self._build_payload("product", config, asin=asin)
                async with session.post(
                    self.BASE_URL,
                    json=payload,
                    headers=self._headers,
                    timeout=aiohttp.ClientTimeout(total=30)
                ) as resp:
                    resp.raise_for_status()
                    return {"asin": asin, "data": await resp.json(), "error": None}

        async with aiohttp.ClientSession() as session:
            tasks = [fetch_one(session, asin) for asin in asin_list]
            results = await asyncio.gather(*tasks, return_exceptions=True)

        return [
            r if isinstance(r, dict) else {"asin": None, "data": None, "error": str(r)}
            for r in results
        ]


# =====================================
# USAGE EXAMPLES
# =====================================

if __name__ == "__main__":
    scraper = PangolinAmazonScraper(api_key="your_api_key_here")

    # Example 1: Product detail with location-specific pricing
    nyc_config = ScrapeConfig(
        marketplace="US",
        zip_code="10001",     # New York — get NYC-specific prices & delivery
        output_format="json"
    )
    product = scraper.get_product("B0CHX1W1XY", config=nyc_config)

    print(f"Title: {product.get('title')}")
    print(f"Price: ${product.get('price', {}).get('current')}")
    print(f"Customer Says: {product.get('customer_says', 'N/A')[:100]}...")
    print(f"SP Ad Slots: {len(product.get('sp_ad_slots', []))} captured")

    # Example 2: Best Sellers for Electronics category
    bestsellers = scraper.get_bestsellers("7HG57G2R", page=1)
    print(f"\nBest Sellers count: {len(bestsellers.get('items', []))}")

    # Example 3: Keyword search with SP ad detection
    keyword_results = scraper.get_keyword_results("wireless earbuds")
    sponsored = [i for i in keyword_results.get('items', []) if i.get('is_sponsored')]
    print(f"\nKeyword results: {len(keyword_results.get('items', []))} total")
    print(f"SP ad slots captured: {len(sponsored)}")

    # Example 4: Batch async product fetch
    asin_list = ["B0CHX1W1XY", "B09G9FPHY6", "B0BDHWDR12"]
    results = asyncio.run(scraper.batch_get_products(asin_list, concurrency=5))
    success = [r for r in results if r.get('data')]
    print(f"\nBatch fetch: {len(success)}/{len(asin_list)} successful")

    # Example 5: Markdown output for LLM analysis pipeline
    llm_config = ScrapeConfig(output_format="markdown", render_js=True)
    product_md = scraper.get_product("B0CHX1W1XY", config=llm_config)
    # product_md now contains Markdown-formatted product data
    # Feed directly to GPT/Claude for competitive analysis — no preprocessing needed

When Self-Built Still Makes Sense

To be fair:

If you're scraping fewer than 10,000 ASINs/month, the math works differently
If you have highly custom data requirements not covered by any commercial API
If you have a dedicated infrastructure team and scraping IS your core product

For everyone else: the operator costs add up fast, and commercial APIs exist precisely because these infrastructure problems are solved more efficiently at scale.

Resources

Pangolinfo Amazon Scraper API — main product page
Full API Documentation — endpoint reference, parameters, response schemas
Free Trial Console — start without committing
Amazon Reviews Scraper API — for review data at scale

Drop any questions in the comments — happy to talk through specific use cases or data requirements.

Adding Real-Time Google AI Mode Data to Your OpenClaw Agent with Pangolinfo's AI Search Skill

Mox Loop — Thu, 05 Mar 2026 06:32:53 +0000

TL;DR

Google AI Mode generates dynamic AI summaries at the top of search results. Most Agent frameworks can't see them. Pangolinfo's AI Mode API (now packaged as an official OpenClaw Skill) fixes this in a single POST request. This post covers the complete integration — API spec, Python code, error handling, and production tips.

OpenClaw AI Search Skill: giving your Agent real-time access to Google AI Overview data

The Problem

If you're building Agent workflows that depend on Google search data, there's a systematic gap you may not have noticed yet: Google AI Overview content is invisible to standard scrapers.

AI Overview (part of Google's AI Mode, triggered by the udm=50 URL parameter) generates summaries server-side via LLM inference and injects them into the page via JavaScript after initial render. Consequences:

Static HTTP requests: don't see it
Headless browsers: need precise render timing + anti-detection to capture it reliably
DIY maintenance: proxy pools + fingerprint rotation + Google update tracking (ongoing)

For an OpenClaw Agent doing keyword research, competitor monitoring, or content strategy, this means a meaningful slice of what users actually see on Google is systematically absent from Agent intelligence.

The Solution: Pangolinfo AI Mode API as an OpenClaw Skill

Pangolinfo's AI Overview SERP API handles the infrastructure complexity server-side. You send a standard POST request; you get back structured JSON with the full AI Overview content.

API Specification

AI Mode API complete call flow: from OpenClaw Skill request to structured JSON response

POST https://scrapeapi.pangolinfo.com/api/v2/scrape

Headers:
  Content-Type: application/json
  Authorization: Bearer {YOUR_API_KEY}

Body:
  url         (string, required)    → Google Search URL with udm=50
  parserName  (string, required)    → "googleAISearch"  ← this selects AI Mode parser
  screenshot  (boolean, optional)   → return page screenshot URL
  param       (string[], optional)  → multi-turn prompts, max 5 (>5 = slower response)

Key point: The udm=50 parameter in the URL is what triggers Google's AI Mode interface. Without it, you get standard SERP results.

Complete curl Example

curl --request POST \
  --url https://scrapeapi.pangolinfo.com/api/v2/scrape \
  --header 'Authorization: Bearer YOUR_API_KEY' \
  --header 'Content-Type: application/json' \
  --data '{
    "url": "https://www.google.com/search?num=10&udm=50&q=javascript+async+await",
    "parserName": "googleAISearch",
    "screenshot": false,
    "param": ["explain with examples", "compare to promises"]
  }'

Response Structure

{
  "code": 0,
  "message": "ok",
  "data": {
    "ai_overview": 1,
    "json": {
      "type": "organic",
      "items": [{
        "type": "ai_overview",
        "items": [{
          "type": "ai_overview_elem",
          "content": [
            "async/await is syntactic sugar over Promises...",
            "An async function always returns a Promise..."
          ]
        }],
        "references": [{
          "type": "ai_overview_reference",
          "title": "MDN Web Docs - async function",
          "url": "https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/async_function",
          "domain": "MDN Web Docs"
        }]
      }]
    },
    "screenshot": null,
    "taskId": "1768988520324-example",
    "url": "https://www.google.com/search?num=10&udm=50&q=javascript+async+await"
  }
}

Credit consumption: 2 credits per call when ai_overview: 1 (AI Overview successfully retrieved).

OpenClaw Skill Implementation

Core Skill Class (Production-Ready)

"""
pangolin_ai_search_skill.py

OpenClaw AI Search Skill — wraps Pangolinfo AI Mode API.
Provides real-time Google AI Overview access for your Agent.
"""

from __future__ import annotations

import asyncio
import aiohttp
import requests
from urllib.parse import quote
from typing import Optional
from dataclasses import dataclass, field


@dataclass
class AIOverviewData:
    """Structured result from Google AI Mode query."""
    query: str
    has_ai_overview: bool
    ai_content: list[str] = field(default_factory=list)
    references: list[dict] = field(default_factory=list)
    screenshot_url: str = ""
    task_id: str = ""

    def summary(self) -> str:
        """Agent-friendly text summary of the result."""
        if not self.has_ai_overview:
            return f"No AI Overview found for: {self.query}"
        lines = [f"## Google AI Overview: {self.query}\n"]
        lines += [f"- {c}" for c in self.ai_content]
        if self.references:
            lines.append("\n**Sources:**")
            lines += [f"  - [{r['domain']}] {r['title']}" for r in self.references]
        return "\n".join(lines)


class PangolinAISearchSkill:
    """
    OpenClaw AI Search Skill

    Integrates Pangolinfo AI Mode API for real-time Google AI Overview data.
    Supports both sync (single queries) and async (batch queries) interfaces.

    Credit cost: 2 credits per successful AI Overview retrieval.
    """

    _ENDPOINT = "https://scrapeapi.pangolinfo.com/api/v2/scrape"
    _GOOGLE_AI_MODE = "https://www.google.com/search?num=10&udm=50&q={q}"
    _MAX_PARAMS = 5  # Official limit; exceeding reduces response efficiency

    def __init__(self, api_key: str, timeout: int = 30) -> None:
        if not api_key:
            raise ValueError("Pangolinfo API key is required.")
        self._timeout = timeout
        self._headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {api_key}",
        }

    # ------------------------------------------------------------------
    # Sync interface
    # ------------------------------------------------------------------

    def search(
        self,
        query: str,
        follow_up: Optional[list[str]] = None,
        screenshot: bool = False,
    ) -> AIOverviewData:
        """Synchronous single-query search."""
        payload = self._build_payload(query, follow_up, screenshot)
        resp = requests.post(
            self._ENDPOINT, headers=self._headers, json=payload, timeout=self._timeout
        )
        resp.raise_for_status()
        return self._parse(query, resp.json())

    # ------------------------------------------------------------------
    # Async interface (recommended for batch workloads)
    # ------------------------------------------------------------------

    async def search_async(
        self,
        query: str,
        session: aiohttp.ClientSession,
        follow_up: Optional[list[str]] = None,
        screenshot: bool = False,
    ) -> AIOverviewData:
        """Async single-query search."""
        payload = self._build_payload(query, follow_up, screenshot)
        async with session.post(
            self._ENDPOINT,
            headers=self._headers,
            json=payload,
            timeout=aiohttp.ClientTimeout(total=self._timeout),
        ) as resp:
            data = await resp.json()
            return self._parse(query, data)

    async def batch_search(
        self,
        queries: list[str],
        max_concurrent: int = 5,
    ) -> list[AIOverviewData | Exception]:
        """
        Async batch search with bounded concurrency.

        Args:
            queries: List of search queries
            max_concurrent: Max simultaneous requests (5–10 is safe for most use cases)

        Returns:
            List of AIOverviewData results (or Exceptions for failed queries)
        """
        sem = asyncio.Semaphore(max_concurrent)

        async def bounded(q: str, session: aiohttp.ClientSession):
            async with sem:
                return await self.search_async(q, session)

        connector = aiohttp.TCPConnector(limit=max_concurrent)
        async with aiohttp.ClientSession(connector=connector) as session:
            tasks = [bounded(q, session) for q in queries]
            return await asyncio.gather(*tasks, return_exceptions=True)

    # ------------------------------------------------------------------
    # Private helpers
    # ------------------------------------------------------------------

    def _build_payload(self, query: str, follow_up, screenshot: bool) -> dict:
        payload: dict = {
            "url": self._GOOGLE_AI_MODE.format(q=quote(query)),
            "parserName": "googleAISearch",
            "screenshot": screenshot,
        }
        if follow_up:
            payload["param"] = follow_up[: self._MAX_PARAMS]
        return payload

    def _parse(self, query: str, raw: dict) -> AIOverviewData:
        if raw.get("code") != 0:
            raise RuntimeError(f"API error [{raw.get('code')}]: {raw.get('message')}")
        data = raw["data"]
        result = AIOverviewData(
            query=query,
            has_ai_overview=bool(data.get("ai_overview")),
            screenshot_url=data.get("screenshot") or "",
            task_id=data.get("taskId") or "",
        )
        for item in data.get("json", {}).get("items", []):
            if item.get("type") != "ai_overview":
                continue
            for sub in item.get("items", []):
                if sub.get("type") == "ai_overview_elem":
                    result.ai_content.extend(sub.get("content", []))
            result.references = [
                {"title": r["title"], "url": r["url"], "domain": r["domain"]}
                for r in item.get("references", [])
            ]
        return result

Usage Examples

# Initialize
skill = PangolinAISearchSkill(api_key="YOUR_PANGOLINFO_API_KEY")

# --- Single sync query ---
result = skill.search("best Python web scraping library 2026")
print(result.summary())

# --- With multi-turn follow-up ---
result = skill.search(
    "asyncio tutorial python",
    follow_up=["show me error handling", "compare to threading"]
)
if result.has_ai_overview:
    for point in result.ai_content:
        print(f"• {point}")

# --- Async batch (recommended for bulk keyword research) ---
keywords = [
    "OpenClaw agent configuration",
    "Google AI Mode scraping",
    "Pangolinfo API review",
    "SERP data extraction tools",
]

results = asyncio.run(skill.batch_search(keywords, max_concurrent=3))

for r in results:
    if isinstance(r, Exception):
        print(f"Error: {r}")
    else:
        status = "✅" if r.has_ai_overview else "⚠️  No AI Overview"
        print(f"{status} — {r.query}")

Production Considerations

Concurrency: Keep max_concurrent at 5–10 for most workloads. Higher values can cause timeout clustering under load.

Multi-turn prompts: The param array supports up to 5 items per the API docs — beyond that, response efficiency degrades. Design multi-turn scenarios as sequential calls rather than single large batches.

AI Overview trigger rate: Not every query generates an AI Overview. Informational queries (how/what/why) trigger at higher rates than navigational (brand names) or transactional queries. Check has_ai_overview before processing to avoid false negatives.

Caching: For repeated queries on hot keywords, a simple TTL cache (1–2 hours) significantly reduces credit consumption without meaningfully affecting data freshness.

Credit budgeting: At 2 credits per AI Overview call, budget planning is straightforward. Consider routing queries that don't need AI Overview to standard parsers to conserve credits.

Where to Go From Here

Get your API key: Pangolinfo Console
Full API docs: docs.pangolinfo.com
AI Overview SERP API product page: pangolinfo.com/ai-overview-serp-api

Questions about the integration? Drop them in the comments — happy to help debug edge cases.

Tags: #api #python #aiagent #webdev #googlesearch #webparsing #automation #llm