Bright Data MCPを使用してAmazon価格監視システムを構築する

業界調査データによると、60%以上の消費者は購入前に少なくとも3つのプラットフォームの価格を比較し、価格差が5%を超えると、70%のトラフィックが競合相手に流れるとされています。Amazonセラーにとって、競合商品の価格をリアルタイムで監視し、市場の変化に迅速に対応することは競争力を維持する鍵です。しかし、数十件の競合商品の価格を手動で確認するのは時間がかかるうえ、リアルタイム性も確保できないため、自動化された価格監視システムが不可欠になっています。

Amazonは世界で最も強力なアンチクローリングシステムの1つを備えており、従来のクローリング手法（requests + BeautifulSoup）はほぼ通用しません。SeleniumやPuppeteerであっても、数分以内に検知されてブロックされます。本ガイドでは、Bright Data MCPプロトコルを使ってこれらの制限を突破し、本番運用レベルの価格監視システムを構築する方法を紹介します。

1. Amazonのアンチスクレイピング機構

Amazonの技術的防御体系は複数の層で構成されており、これらの仕組みを理解することは有効なデータ収集方案を設計するうえで極めて重要です。

五層防御システム

第1層：IPブロック - Amazonはアクセス頻度を監視しており、短時間に大量のリクエストがあると一時的なブロックが発生します。

第2層：行動分析 - マウスの移動軌跡、スクロール速度、ページ滞在時間などの行動特性は、クローラーの識別に使用されます。

第3層：動的コンテンツの読み込み - 価格や在庫などの中核データはJavaScriptによって非同期に読み込まれるため、従来のHTTPリクエストでは取得できません。

第4層：CAPTCHAシステム - 不審なアクセスは直ちにCAPTCHA認証を引き起こします。

第5層：ブラウザフィンガープリント識別 - 最も複雑な防御層。AmazonはCanvasフィンガープリント、WebGLパラメータ、フォント一覧、Navigatorオブジェクトなど数十の次元を通じて一意のデバイスフィンガープリントを生成しており、IPアドレスを変更しても、同じブラウザフィンガープリントは同一デバイスとして識別されます。

Bright Data MCPの3層突破技術

Bright Data MCPは3層の技術によってAmazonの防御を突破します：

MCP（Model Context Protocol）プロトコルの導入により、統合はさらに簡素化されます。開発者は複雑なプロキシ管理や検知回避ロジックを処理する必要がなく、統一されたAPIインターフェースを呼び出すだけでよく、すべての技術的詳細はBright Dataがクラウド側で処理します。このアーキテクチャにより、データ収集の複雑さは90%以上低減されます。

2. 環境準備とAPI設定

Bright Data APIキーを取得する

Bright Dataは新規ユーザー向けに非常に手厚い無料トライアルプランを提供しています。最初の3か月間は毎月5000リクエストが完全無料で、クレジットカードの登録も不要です。登録手順は非常に簡単で、アクセス先は公式サイトの登録ページ基本情報を入力するだけで完了です。登録成功後、コントロールパネルのSettings → Usersページに入り、Generate API TokenボタンをクリックしてAPIキーを生成してください。

Linux/Mac 環境変数設定

# 在~/.bashrc或~/.zshrc中添加
export BRIGHT_DATA_TOKEN="your_api_token_here"

Windows 環境変数設定

# 在项目的.env文件中配置
BRIGHT_DATA_TOKEN=your_api_token_here

Python環境設定

このガイドでは開発言語としてPython 3.8+を使用し、プロジェクト依存関係を分離するために仮想環境の作成を推奨します：

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境（Linux/Mac）
source venv/bin/activate

# 激活虚拟环境（Windows）
venv\Scripts\activate

# 安装依赖
pip install requests beautifulsoup4 lxml pandas python-dotenv schedule aiohttp

プロジェクト構造設計

amazon-price-monitor/
├── config/
│   ├── __init__.py
│   └── settings.py          # 配置参数
├── src/
│   ├── __init__.py
│   ├── mcp_client.py         # MCP客户端
│   ├── scraper.py            # 亚马逊页面解析
│   ├── monitor.py            # 价格监控逻辑
│   └── storage.py            # 数据存储
├── data/
│   ├── products.json         # 监控产品一覧
│   └── prices.db             # SQLite数据库
├── logs/
│   └── monitor.log           # 日志文件
├── main.py                   # 主程序入口
├── requirements.txt
└── .env                      # 环境变量

3. MCPクライアントのコア実装

MCPクライアントはBright Dataサービスと通信するための中核コンポーネントです。以下は本番運用レベルの実装例です：

import os
import json
import time
import logging
from typing import Dict, List, Any, Optional
from datetime import datetime
import requests
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

class BrightDataMCPClient:
    """Bright Data MCP客户端实现"""

    def __init__(self, api_token: Optional[str] = None):
        self.api_token = api_token or os.getenv('BRIGHT_DATA_TOKEN')
        if not self.api_token:
            raise ValueError("API Token未设置")

        self.base_url = f"https://mcp.brightdata.com/mcp?token={self.api_token}"
        self.session = requests.Session()
        self.session_id: Optional[str] = None
        self.message_id = 1

        # 配置请求头
        self.session.headers.update({
            'Content-Type': 'application/json',
            'Accept': 'application/json, text/event-stream',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })

    def _send_request(self, payload: Dict[str, Any], max_retries: int = 3) -> Dict[str, Any]:
        """发送JSON-RPC请求（带重试机制）"""
        if self.session_id:
            self.session.headers['mcp-session-id'] = self.session_id

        for attempt in range(max_retries):
            try:
                response = self.session.post(self.base_url, json=payload, timeout=30)

                # 保存会话ID
                if 'mcp-session-id' in response.headers:
                    self.session_id = response.headers['mcp-session-id']

                # 处理速率限制
                if response.status_code == 429:
                    retry_after = int(response.headers.get('Retry-After', 60))
                    time.sleep(retry_after)
                    continue

                response.raise_for_status()
                return response.json()

            except requests.RequestException as e:
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)  # 指数退避

    def initialize(self) -> bool:
        """初始化MCP协议"""
        init_payload = {
            "jsonrpc": "2.0",
            "id": self.message_id,
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "capabilities": {"roots": {"listChanged": True}, "sampling": {}},
                "clientInfo": {"name": "Amazon-Price-Monitor", "version": "1.0.0"}
            }
        }
        self.message_id += 1
        response = self._send_request(init_payload)

        if 'error' in response:
            return False

        # 发送initialized通知
        self._send_request({"jsonrpc": "2.0", "method": "notifications/initialized"})
        return True

    def scrape_amazon_product(self, url: str) -> Optional[str]:
        """抓取亚马逊产品页面（返回Markdown格式）"""
        scrape_payload = {
            "jsonrpc": "2.0",
            "id": self.message_id,
            "method": "tools/call",
            "params": {
                "name": "scrape_as_markdown",
                "arguments": {"url": url, "formats": ["markdown"]}
            }
        }
        self.message_id += 1
        response = self._send_request(scrape_payload)

        if 'error' in response:
            return None

        # 提取Markdown内容
        content_list = response.get('result', {}).get('content', [])
        markdown_text = ''
        for item in content_list:
            if isinstance(item, dict) and 'text' in item:
                markdown_text += item['text']

        return markdown_text

    def close(self):
        """关闭会话"""
        if self.session:
            self.session.close()

4. Amazonページデータの抽出

Amazonの商品ページ構造はかなり複雑で、価格情報は複数の場所に分散しています。中核となる価格は通常id="priceblock_ourprice"またはid="priceblock_dealprice"の要素内。

正規表現に基づく抽出方法

import re
from typing import Dict, Optional
from datetime import datetime

class AmazonProductExtractor:
    """亚马逊产品数据提取器"""

    @staticmethod
    def extract_price(markdown: str) -> Optional[float]:
        """提取价格信息"""
        patterns = [
            r'\$\s?([\d,]+\.?\d*)',           # $19.99 或 $ 19.99
            r'USD\s?([\d,]+\.?\d*)',          # USD 19.99
            r'Price:\s*\$\s*([\d,]+\.?\d*)',  # Price: $19.99
        ]

        for pattern in patterns:
            match = re.search(pattern, markdown, re.IGNORECASE)
            if match:
                price_str = match.group(1).replace(',', '')
                try:
                    return float(price_str)
                except ValueError:
                    continue
        return None

    @staticmethod
    def extract_title(markdown: str) -> Optional[str]:
        """提取产品标题"""
        patterns = [
            r'^#\s+(.+)$',                   # 一级标题
            r'Product Name:\s*(.+)',         # 产品名称
            r'Amazon\.com\s*:\s*(.+)',       # Amazon.com: 产品名
        ]

        for pattern in patterns:
            match = re.search(pattern, markdown, re.MULTILINE)
            if match:
                title = match.group(1).strip()
                if 10 < len(title) < 200:
                    return title
        return None

    @staticmethod
    def extract_availability(markdown: str) -> str:
        """提取库存状态"""
        markdown_lower = markdown.lower()

        if any(p in markdown_lower for p in ['in stock', 'available', 'add to cart']):
            return 'In Stock'
        if any(p in markdown_lower for p in ['out of stock', 'unavailable']):
            return 'Out of Stock'
        return 'Unknown'

    @staticmethod
    def extract_all(markdown: str) -> Dict:
        """提取所有产品信息"""
        return {
            'title': AmazonProductExtractor.extract_title(markdown),
            'price': AmazonProductExtractor.extract_price(markdown),
            'availability': AmazonProductExtractor.extract_availability(markdown),
            'extracted_at': datetime.now().isoformat()
        }

5. 価格監視システムアーキテクチャ

データモデル設計

from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Optional

@dataclass
class ProductPrice:
    """价格记录数据模型"""
    sku: str                    # 产品SKU（ASIN）
    title: str                  # 产品标题
    price: Optional[float]      # 当前价格
    currency: str               # 货币代码
    availability: str           # 库存状态
    timestamp: datetime         # 采集时间
    source_url: str             # 源URL

@dataclass
class PriceAlert:
    """价格告警配置"""
    sku: str
    alert_type: str  # 'above', 'below', 'change_percent'
    threshold: float
    enabled: bool = True

    def should_alert(self, current_price: float, previous_price: Optional[float] = None) -> bool:
        """判断是否应该触发告警"""
        if not self.enabled:
            return False

        if self.alert_type == 'above' and current_price > self.threshold:
            return True
        elif self.alert_type == 'below' and current_price < self.threshold:
            return True
        elif self.alert_type == 'change_percent' and previous_price:
            change_percent = abs((current_price - previous_price) / previous_price * 100)
            if change_percent >= self.threshold:
                return True
        return False

監視のコアロジック

import time
import schedule
from typing import List, Dict, Optional

class PriceMonitor:
    """价格监控主控制器"""

    def __init__(self, mcp_client, storage):
        self.client = mcp_client
        self.storage = storage
        self.extractor = AmazonProductExtractor()
        self.products = {}  # SKU -> URL映射
        self.alerts = {}    # SKU -> Alert配置

    def add_product(self, sku: str, url: str):
        """添加监控产品"""
        self.products[sku] = url

    def set_alert(self, sku: str, alert: PriceAlert):
        """设置价格告警"""
        self.alerts[sku] = alert

    def check_product(self, sku: str) -> Optional[ProductPrice]:
        """检查单个产品的价格"""
        if sku not in self.products:
            return None

        url = self.products[sku]
        markdown = self.client.scrape_amazon_product(url)
        if not markdown:
            return None

        # 提取数据
        extracted = self.extractor.extract_all(markdown)

        # 创建价格记录
        price_record = ProductPrice(
            sku=sku,
            title=extracted.get('title', 'Unknown'),
            price=extracted.get('price'),
            currency='USD',
            availability=extracted.get('availability', 'Unknown'),
            timestamp=datetime.now(),
            source_url=url
        )

        # 保存到数据库
        self.storage.save_price(price_record)

        # 检查告警
        if sku in self.alerts and price_record.price:
            previous = self.storage.get_recent_prices(sku, limit=1)
            prev_price = previous[0].price if previous else None
            if self.alerts[sku].should_alert(price_record.price, prev_price):
                self._trigger_alert(sku, price_record)

        return price_record

    def start(self, interval_minutes: int = 60):
        """启动定时监控"""
        # 立即执行一次
        for sku in self.products:
            self.check_product(sku)
            time.sleep(2)  # 避免请求过快

        # 设置定时任务
        schedule.every(interval_minutes).minutes.do(
            lambda: [self.check_product(sku) for sku in self.products]
        )

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

6. データ保存とトレンド分析

SQLiteデータベース実装

import sqlite3
from typing import List, Dict
from contextlib import contextmanager

class SQLiteStorage:
    """基于SQLite的数据存储"""

    def __init__(self, db_path: str):
        self.db_path = db_path
        self._init_db()

    @contextmanager
    def _get_connection(self):
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
        finally:
            conn.close()

    def _init_db(self):
        """初始化数据库表"""
        with self._get_connection() as conn:
            conn.execute('''
                CREATE TABLE IF NOT EXISTS price_history (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    sku TEXT NOT NULL,
                    title TEXT,
                    price REAL,
                    currency TEXT DEFAULT 'USD',
                    availability TEXT,
                    timestamp DATETIME NOT NULL,
                    source_url TEXT
                )
            ''')
            conn.execute('''
                CREATE INDEX IF NOT EXISTS idx_sku_timestamp
                ON price_history(sku, timestamp)
            ''')
            conn.commit()

    def save_price(self, price_record) -> bool:
        """保存价格记录"""
        try:
            with self._get_connection() as conn:
                conn.execute('''
                    INSERT INTO price_history
                    (sku, title, price, currency, availability, timestamp, source_url)
                    VALUES (?, ?, ?, ?, ?, ?, ?)
                ''', (
                    price_record.sku, price_record.title, price_record.price,
                    price_record.currency, price_record.availability,
                    price_record.timestamp, price_record.source_url
                ))
                conn.commit()
                return True
        except Exception:
            return False

    def get_price_statistics(self, sku: str, days: int = 30) -> Dict:
        """获取价格统计信息"""
        with self._get_connection() as conn:
            cursor = conn.execute(f'''
                SELECT COUNT(*) as count, AVG(price) as avg_price,
                       MIN(price) as min_price, MAX(price) as max_price
                FROM price_history
                WHERE sku = ? AND price IS NOT NULL
                AND timestamp >= datetime('now', '-{days} days')
            ''', (sku,))
            row = cursor.fetchone()
            return dict(row) if row else {}

7. パフォーマンス最適化と本番デプロイ

非同期並列最適化

監視対象の商品数が50件を超えると、直列取得では総所要時間が長くなりすぎます。非同期並行処理を使うことで、性能を大幅に向上できます：

import asyncio
import aiohttp

class AsyncPriceMonitor:
    """异步价格监控器"""

    def __init__(self, api_token: str, max_concurrent: int = 10):
        self.api_token = api_token
        self.base_url = f"https://mcp.brightdata.com/mcp?token={api_token}"
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def scrape_async(self, url: str, session: aiohttp.ClientSession):
        """异步抓取页面"""
        async with self.semaphore:
            payload = {
                "jsonrpc": "2.0", "id": 1,
                "method": "tools/call",
                "params": {"name": "scrape_as_markdown", "arguments": {"url": url}}
            }
            try:
                async with session.post(self.base_url, json=payload, timeout=30) as response:
                    data = await response.json()
                    content_list = data.get('result', {}).get('content', [])
                    return ''.join([item.get('text', '') for item in content_list if isinstance(item, dict)])
            except Exception:
                return None

    async def check_products_async(self, products: list):
        """并发检查多个产品"""
        async with aiohttp.ClientSession() as session:
            tasks = [self.scrape_async(p['url'], session) for p in products]
            return await asyncio.gather(*tasks)

Dockerコンテナ化デプロイ

# Dockerfile
FROM python:3.10-slim

WORKDIR /app

RUN apt-get update && apt-get install -y gcc && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
RUN mkdir -p logs data

ENV PYTHONUNBUFFERED=1

CMD ["python", "main.py"]

# docker-compose.yml
version: '3.8'

services:
  price-monitor:
    build: .
    container_name: amazon-price-monitor
    restart: unless-stopped
    environment:
      - BRIGHT_DATA_TOKEN=${BRIGHT_DATA_TOKEN}
      - TZ=Asia/Shanghai
    volumes:
      - ./data:/app/data
      - ./logs:/app/logs

# 部署命令
docker-compose build
docker-compose up -d
docker-compose logs -f

まとめ

本ガイドでは、Amazon価格監視システムの完全な実装方法を提供します。環境構築、MCPクライアント、データ抽出、監視ロジックからデータ分析および本番デプロイまで、すべての重要な工程を網羅しています。中核的な利点は、Bright Data MCPを使用してAmazonの複雑なアンチクローリング機構を回避し、開発者がクローリング技術ではなくビジネスロジックに集中できる点にあります。

関連記事

• Bright DataプロキシIPレビュー
• おすすめEC向けプロキシIP
• おすすめ固定住宅プロキシIP