Author : Leo, Technical Lead at Pangolinfo Tags : amazon python api mcp web-scraping data-analysis Reading time : ~12 minutes TL;DR This tutorial walks through building a complete Amazon competitor research system using Python and the Pangolinfo API. You'll learn the 5-step IBADM framework (Identify → Baseline → Analyze → Differentiate → Monitor) and get production-ready code you can run today. We'll also cover how to use the Amazon Data MCP for no-code competitor analysis. Amazon Scraper API: https://www.pangolinfo.com/amazon-scraper-api/?referrer=devto_amz Amazon Data MCP: https://www.pangolinfo.com/amazon-data-mcp/?referrer=devto_mcp Why This Matters If you've ever done Amazon competitor research manually, you know the pain: 3-4 hours per competitor to do a thorough analysis Estimated data with 20-50% error from third-party tools No continuous monitoring — you get a snapshot, not a stream Poor SP ad coverage — manual browsing catches maybe 50-70% of sponsored placements I've spent five years at Pangolinfo building data infrastructure for Amazon sellers. This article is the system I wish I had when I started. Everything here is battle-tested in production. The IBADM Framework Before we write code, let's define the framework. Good competitor research follows five steps: Identify → Baseline → Analyze → Differentiate → Monitor Identify : Find all ASINs competing for your keywords (organic + sponsored) Baseline : Capture current state of all competitors simultaneously Analyze : Break down listing structure, keyword strategy, review patterns Differentiate : Find gaps — competitor weaknesses are your opportunities Monitor : Track changes continuously, get alerted on significant shifts Each step maps to specific API calls. Let's build it. Setup pip install requests pandas schedule import requests import pandas as pd import json import time from datetime import datetime from typing import List , Dict from concurrent.futures import ThreadPoolExecutor Step 0: API Client First, let's build a clean API client. Get your API key from Pangolinfo . class AmazonAPIClient : """ Client for Pangolinfo Amazon Scraper API.
Key specs:
- Data latency: ~3 seconds
- Daily capacity: 30M+ requests
- Success rate: 99%
- SP ad placement coverage: 98% (industry #1) """ def __init__ ( self , api_key : str , domain : str = " com " ): self . api_key = api_key self . domain = domain self . base_url = " https://api.pangolinfo.com/v1 " self . session = requests . Session () self . session . headers . update ({ " Authorization " : f " Bearer { api_key } " , " Content-Type " : " application/json " }) def _post ( self , endpoint : str , payload : dict ) -> dict : url = f " { self . base_url }{ endpoint } " payload [ " domain " ] = self . domain resp = self . session . post ( url , json = payload , timeout = 30 ) resp . raise_for_status () return resp . json () def search_keyword ( self , keyword : str , page : int = 1 ) -> dict : """ Search Amazon by keyword. Returns organic + sponsored results. """ return self . _post ( " /amazon/search " , { " keyword " : keyword , " page " : page }) def get_product ( self , asin : str ) -> dict : """ Get full product details for an ASIN. """ return self . _post ( " /amazon/product " , { " asin " : asin }) def get_keyword_ranking ( self , asin : str , keywords : List [ str ]) -> dict : """ Get ASIN ' s ranking position for specified keywords. """ return self . _post ( " /amazon/keyword-ranking " , { " asin " : asin , " keywords " : keywords }) def get_reviews ( self , asin : str , review_type : str = " critical " ) -> dict : """ Get reviews. review_type: ' critical ' or ' positive ' . """ return self . _post ( " /amazon/reviews " , { " asin " : asin , " filter_type " : review_type , " page " : 1 }) def get_bsr_history ( self , asin : str ) -> dict : """ Get BSR ranking history for an ASIN. """ return self . _post ( " /amazon/bsr-history " , { " asin " : asin }) Step 1: Identify Competitors def identify_competitors ( client : AmazonAPIClient , keywords : List [ str ]) -> List [ str ]: """ Find all competitor ASINs from keyword search results.
Pulls both organic results and SP ad placements.
Returns a deduplicated list of ASINs. """ competitors = set () for kw in keywords : result = client . search_keyword ( kw ) # Organic results for item in result . get ( " organic_results " , []): competitors . add ( item [ " asin " ]) # Sponsored placements (SP ads) — 98% coverage for item in result . get ( " sponsored_results " , []): competitors . add ( item [ " asin " ]) # Polite delay between keywords time . sleep ( 0.3 ) return list ( competitors ) # Usage client = AmazonAPIClient ( api_key = " your_key " ) keywords = [ " silicone spatula set " , " heat resistant spatula " , " kitchen spatula " ] competitors = identify_competitors ( client , keywords ) print ( f " Found { len ( competitors ) } competitor ASINs " ) Step 2: Build Baseline def build_baseline ( client : AmazonAPIClient , asins : List [ str ]) -> pd . DataFrame : """ Capture current state of all competitors simultaneously.
Uses ThreadPoolExecutor for parallel data fetching.
Total time for 20 ASINs: ~3-5 seconds. """ def fetch_one ( asin : str ) -> dict : data = client . get_product ( asin ) return { " asin " : asin , " title " : data . get ( " title " , "" ), " price " : data . get ( " price " , 0 ), " bsr " : data . get ( " bsr " , 0 ), " rating " : data . get ( " rating " , 0 ), " review_count " : data . get ( " review_count " , 0 ), " variant_count " : len ( data . get ( " variants " , [])), " image_count " : len ( data . get ( " images " , [])), " has_a_plus " : data . get ( " has_a_plus " , False ), " qa_count " : data . get ( " qa_count " , 0 ), " timestamp " : datetime . now (). isoformat () } # Parallel fetch — all competitors at once with ThreadPoolExecutor ( max_workers = 10 ) as executor : results = list ( executor . map ( fetch_one , asins )) return pd . DataFrame ( results ) # Usage baseline_df = build_baseline ( client , competitors ) baseline_df . to_csv ( " competitor_baseline.csv " , index = False ) print ( baseline_df [[ " asin " , " price " , " bsr " , " rating " , " review_count " ]]. head ( 10 )) Step 3: Analyze 3a: Keyword Gap Analysis def keyword_gap_analysis ( client : AmazonAPIClient , target_asin : str , competitor_asins : List [ str ], keywords : List [ str ] ) -> pd . DataFrame : """ Compare keyword rankings between your ASIN and competitors.
Identifies keywords where competitors rank but you don ' t,
and vice versa. """ rows = [] # Your rankings my_ranks = client . get_keyword_ranking ( target_asin , keywords ) for kw , rank in my_ranks . get ( " rankings " , {}). items (): rows . append ({ " keyword " : kw , " your_rank " : rank , " your_search_volume " : my_ranks . get ( " volumes " , {}). get ( kw , 0 ) }) df = pd . DataFrame ( rows ) # Competitor rankings for comp_asin in competitor_asins : comp_ranks = client . get_keyword_ranking ( comp_asin , keywords ) rank_map = comp_ranks . get ( " rankings " , {}) df [ f " comp_ { comp_asin } _rank " ] = df [ " keyword " ]. map ( rank_map ) return df # Usage gap_df = keyword_gap_analysis ( client , " B0YOURASIN " , competitors [: 5 ], keywords ) print ( gap_df ) 3b: Review Pain Point Analysis def analyze_review_pain_points ( client : AmazonAPIClient , asin : str , top_n : int = 50 ) -> Dict [ str , int ]: """ Extract frequent keywords from negative reviews.
These pain points are your differentiation opportunities.
If competitors ' customers complain about ' breaks easily ' ,
that ' s your chance to emphasize durability. """ from collections import Counter import re reviews = client . get_reviews ( asin , review_type = " critical " ) review_texts = [ r [ " content " ] for r in reviews . get ( " reviews " , [])[: top_n ]] # Simple word frequency (use NLP libraries for production) words = [] for text in review_texts : words . extend ( re . findall ( r ' [a-z]{4,} ' , text . lower ())) # Filter common stop words stop_words = { " this " , " that " , " with " , " have " , " from " , " they " , " were " , " been " , " would " , " could " , " should " , " really " , " very " } words = [ w for w in words if w not in stop_words ] return dict ( Counter ( words ). most_common ( 15 )) # Usage for asin in competitors [: 3 ]: pain_points = analyze_review_pain_points ( client , asin ) print ( f " \n ASIN { asin } — Top pain points: " ) for word , count in pain_points . items (): print ( f " { word } : { count } " ) Step 4: Differentiate def find_differentiation_opportunities ( client : AmazonAPIClient , my_asin : str , competitors : List [ str ], keywords : List [ str ] ) -> dict : """ Combine keyword gaps and review pain points into actionable insights. """ # Keyword gaps: where competitors rank but you don't gap_df = keyword_gap_analysis ( client , my_asin , competitors , keywords ) missing_keywords = gap_df [ gap_df [ " your_rank " ]. isna ()][ " keyword " ]. tolist () # Pain points from all competitors all_pain_points = {} for asin in competitors : all_pain_points [ asin ] = analyze_review_pain_points ( client , asin ) # Cross-reference: find pain points mentioned across multiple competitors from collections import Counter pain_word_counts = Counter () for pp in all_pain_points . values (): for word in pp : pain_word_counts [ word ] += 1 shared_pain_points = { word : count for word , count in pain_word_counts . items () if count >= 2 # Mentioned by 2+ competitors' customers } return { " keyword_opportunities " : missing_keywords , " shared_pain_points " : shared_pain_points , " recommendation " : ( " Target keywords where competitors rank but you don ' t. " " Address shared pain points in your listing to differentiate. " ) } # Usage opportunities = find_differentiation_opportunities ( client , " B0YOURASIN " , competitors [: 5 ], keywords ) print ( json . dumps ( opportunities , indent = 2 )) Step 5: Monitor def start_monitoring ( client : AmazonAPIClient , asins : List [ str ], output_file : str = " monitor_log.csv " , alert_threshold_pct : float = 10.0 ): """ Set up daily monitoring with price change alerts.
Runs daily, logs all data, and prints alerts when
price changes exceed threshold. """ import schedule def job (): # Fetch current state current = build_baseline ( client , asins ) current [ " check_time " ] = datetime . now (). isoformat () # Append to log try : existing = pd . read_csv ( output_file ) pd . concat ([ existing , current ]). to_csv ( output_file , index = False ) except FileNotFoundError : current . to_csv ( output_file , index = False ) # Check for price changes (compare with last entry) try : history = pd . read_csv ( output_file ) last_entry = history [ history [ " check_time " ] != current [ " check_time " ]. iloc [ 0 ]] if len ( last_entry ) > 0 : last_entry = last_entry . tail ( len ( asins )) for _ , curr_row in current . iterrows (): prev = last_entry [ last_entry [ " asin " ] == curr_row [ " asin " ]] if len ( prev ) > 0 : prev_price = prev . iloc [ 0 ][ " price " ] if prev_price > 0 : change_pct = (( curr_row [ " price " ] - prev_price ) / prev_price ) * 100 if abs ( change_pct ) > alert_threshold_pct : direction = " dropped " if change_pct < 0 else " increased " print ( f " 🚨 ALERT: ASIN { curr_row [ ' asin ' ] } price { direction } " f " { abs ( change_pct ) : . 1 f } % " f " ($ { prev_price } → $ { curr_row [ ' price ' ] } ) " ) except Exception as e : print ( f " Alert check error: { e } " ) print ( f " [ { datetime . now () } ] Monitor cycle complete — { len ( asins ) } ASINs checked " ) # Run daily at 9 AM schedule . every (). day . at ( " 09:00 " ). do ( job ) print ( f " Monitoring started. Daily at 09:00. Output: { output_file } " ) print ( f " Alert threshold: ± { alert_threshold_pct } % price change " ) print ( " Press Ctrl+C to stop. \n " ) # Initial run job () while True : schedule . run_pending () time . sleep ( 60 ) # Usage start_monitoring ( client , competitors , output_file = " daily_monitor.csv " ) No-Code Alternative: Amazon Data MCP If you don't want to write code, or want your operations team to run analysis independently, use the Amazon Data MCP . What is MCP? MCP (Model Context Protocol) is a protocol that lets AI models call external tools. Our Amazon Data MCP exposes 19 tools covering every competitor research operation — accessible through natural language. Setup Add this to your AI assistant's config (e.g., claude_desktop_config.json ): { "mcpServers" : { "amazon-data" : { "url" : "https://mcp.pangolinfo.com/amazon-data-mcp" , "transport" : "http" } } } Remote HTTP. Zero installation. No Python, no dependencies. Usage Just type natural language: "Pull the price and BSR for these 5 ASINs: B0xxx, B0yyy, B0zzz,
B0aaa, B0bbb. Compare them in a table and highlight which one has the best price-to-rating ratio." The AI calls the right MCP tools, pulls the data, and returns a formatted analysis. The 19 tools cover: Category Tools Identify search_products, get_sponsored_ads, get_category_bestsellers Baseline get_product_detail, get_variants, get_bsr_history, get_listing_content Analyze get_keyword_ranking, get_reviews, get_qa, analyze_review_sentiment, get_price_history Differentiate compare_products, find_keyword_gap, analyze_competitor_weakness Monitor create_monitor_task, get_monitor_alerts, list_monitor_tasks, get_change_history Performance Comparison Here's the real-world difference between approaches: Metric Manual Traditional Tools API + MCP Time per competitor 3-4 hours 20-30 min ~3 seconds Data accuracy 50-80% 70-90% 99% SP ad coverage Low 50-70% 98% Continuous monitoring None Limited Full control Batch capacity 1 at a time Tool-limited 30M+/day Non-technical usage N/A Limited (UI only) Full (via MCP) Production Tips Tip 1: Store historical data The monitor_log.csv file is your most valuable asset. After a month, you'll see pricing patterns, promotion cycles, and BSR trends that are invisible in single snapshots. Don't just log — analyze the time series. Tip 2: Focus on negative reviews Most sellers only look at positive reviews for competitor insights. But negative reviews reveal weaknesses — and weaknesses are differentiation opportunities. Always pull review_type="critical" . Tip 3: Set meaningful alert thresholds A 5% price change might be noise. A 15% change is a strategy shift. Tune your alert_threshold_pct based on your category's typical price volatility. Tip 4: Use MCP for exploration, API for automation MCP is great for ad-hoc analysis and exploration. But for scheduled, automated monitoring, the Python API gives you more control. Use both. Complete Script Here's the full runnable script combining everything: #!/usr/bin/env python3 """ Amazon Competitor Research System — IBADM Framework Author: Leo, Pangolinfo Technical Lead """ import requests import pandas as pd import time from datetime import datetime from concurrent.futures import ThreadPoolExecutor # ... (all the code from above, combined into one runnable file)