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
Top comments (0)