BuyWhere
← Back to documentation

Building a Deal Alert Bot with BuyWhere Webhooks

The Problem with Manual Deal Hunting

Checking prices across multiple e-commerce platforms is tedious. You find a product you want, bookmark it, check back a week later, and hope the price dropped. Meanwhile, flash sales come and go while you're not watching.

Automated deal alerts solve this. Instead of checking prices manually, your agent watches for you and notifies you when conditions are met.

BuyWhere Webhooks Overview

BuyWhere supports webhook callbacks for price and availability changes. Instead of polling the API repeatedly, you register a webhook URL, and BuyWhere pushes updates when relevant changes occur.

Webhook Registration

curl -X POST "https://api.buywhere.ai/v2/webhooks" \
  -H "Authorization: Bearer $BUYWHERE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-server.com/webhooks/buywhere",
    "events": ["price_drop", "back_in_stock"],
    "product_ids": ["bw_prod_123", "bw_prod_456"],
    "conditions": {
      "min_discount_pct": 15
    }
  }'

Webhook Payload

When a matching event occurs, BuyWhere sends:

{
  "event": "price_drop",
  "product_id": "bw_prod_123",
  "timestamp": "2026-04-16T10:30:00Z",
  "data": {
    "previous_price_sgd": 449.00,
    "current_price_sgd": 379.00,
    "discount_pct": 15.6,
    "source": "shopee_sg",
    "buy_url": "https://shopee.sg/product/123",
    "affiliate_url": "https://affiliate.buywhere.ai/track/123"
  }
}

Building the Alert Bot

Prerequisites

# Flask for the webhook receiver
pip install flask requests

# Or for async with FastAPI
pip install fastapi uvicorn requests

Flask Implementation

from flask import Flask, request, jsonify
import requests
import hmac
import hashlib

app = Flask(__name__)

WEBHOOK_SECRET = "your_webhook_secret"
TELEGRAM_BOT_TOKEN = "your_telegram_token"
TELEGRAM_CHAT_ID = "your_chat_id"

def verify_webhook_signature(payload, signature):
    """Verify BuyWhere webhook authenticity."""
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, expected)

def send_telegram_message(message):
    """Send alert via Telegram."""
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    requests.post(url, json={
        "chat_id": TELEGRAM_CHAT_ID,
        "text": message,
        "parse_mode": "HTML"
    })

@app.route("/webhooks/buywhere", methods=["POST"])
def handle_webhook():
    # Verify signature
    signature = request.headers.get("X-BuyWhere-Signature", "")
    if not verify_webhook_signature(request.data.decode(), signature):
        return jsonify({"error": "Invalid signature"}), 401
    
    event = request.json
    
    # Process price drop alerts
    if event["event"] == "price_drop":
        data = event["data"]
        discount = data["discount_pct"]
        previous = data["previous_price_sgd"]
        current = data["current_price_sgd"]
        product_id = event["product_id"]
        
        message = (
            f"🚨 <b>Price Drop Alert!</b>\n\n"
            f"Product: {product_id}\n"
            f"Previous: S${previous:.2f}\n"
            f"Current: S${current:.2f}\n"
            f"💰 Savings: {discount:.1f}%\n\n"
            f"<a href='{data['affiliate_url']}'>Buy Now</a>"
        )
        
        send_telegram_message(message)
    
    # Process back-in-stock alerts
    elif event["event"] == "back_in_stock":
        data = event["data"]
        message = (
            f"📦 <b>Back in Stock!</b>\n\n"
            f"Product: {event['product_id']}\n"
            f"Price: S${data['current_price_sgd']:.2f}\n\n"
            f"<a href='{data['affiliate_url']}'>Buy Now</a>"
        )
        send_telegram_message(message)
    
    return jsonify({"status": "processed"})

if __name__ == "__main__":
    app.run(port=5000)

Advanced: Multi-Product Watchlist

Track deals across your entire shopping list:

import requests
import time

BUYWHERE_API = "https://api.buywhere.ai/v2"
API_KEY = "your_api_key"

class DealWatcher:
    def __init__(self, api_key):
        self.api_key = api_key
        self.wishlist = []
    
    def add_product(self, product_query):
        """Search and add product to watchlist."""
        response = requests.get(
            f"{BUYWHERE_API}/products",
            headers={"Authorization": f"Bearer {self.api_key}"},
            params={"q": product_query, "region": "sg", "limit": 1}
        )
        data = response.json()
        if data["items"]:
            self.wishlist.append({
                "product_id": data["items"][0]["id"],
                "name": data["items"][0]["name"],
                "target_price": None,
                "last_price": data["items"][0]["price_sgd"]
            })
            print(f"Added: {data['items'][0]['name']}")
    
    def set_target_price(self, index, target_price):
        """Set desired price for a wishlist item."""
        if index < len(self.wishlist):
            self.wishlist[index]["target_price"] = target_price
    
    def check_prices(self):
        """Poll current prices against targets."""
        for item in self.wishlist:
            response = requests.get(
                f"{BUYWHERE_API}/products/{item['product_id']}",
                headers={"Authorization": f"Bearer {self.api_key}"}
            )
            current = response.json()
            current_price = current["price_sgd"]
            item["last_price"] = current_price
            
            if item["target_price"] and current_price <= item["target_price"]:
                print(f"✅ DEAL: {item['name']} is now S${current_price} "
                      f"(target: S${item['target_price']})")
        
        return self.wishlist

# Usage
watcher = DealWatcher(API_KEY)
watcher.add_product("Sony WH-1000XM5")
watcher.add_product("MacBook Air M3")
watcher.set_target_price(0, 350)  # Want Sony under S$350
watcher.set_target_price(1, 1100)  # Want MacBook under S$1100

# Check prices
while True:
    watcher.check_prices()
    time.sleep(3600)  # Check hourly

Production Considerations

Webhook Security

Always verify webhook signatures to prevent spoofing:

def verify_webhook(request, secret):
    signature = request.headers.get("X-BuyWhere-Signature")
    if not signature:
        return False
    
    expected = hmac.new(
        secret.encode(),
        request.get_data(),
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(signature, expected)

Reliability

  • Use a persistent server (not serverless) for webhook receivers
  • Respond with 200 quickly, process asynchronously
  • Implement retry logic for failed notifications
  • Log all events for debugging

Rate Limits

BuyWhere webhooks have configurable rate limits. For production, consider:

  • Batch notifications during peak periods
  • Implement backpressure handling
  • Use queuing (Redis, SQS) for notification delivery

Comparison: Webhooks vs Polling

ApproachProsCons
WebhooksReal-time, efficientRequires public URL, more complex setup
PollingSimple, no server requiredAPI overhead, potential rate limits, delayed detection

For a few products, polling works fine. For monitoring dozens of products across categories, webhooks scale better.

Conclusion

Deal alerts are a practical first project for any BuyWhere integration. The combination of webhooks for real-time updates and a simple notification system (Telegram, Discord, email) creates a powerful price monitoring agent without building complex scraping infrastructure.

The same pattern extends to:

  • Back-in-stock alerts
  • New product releases
  • Price predictions (is this a good time to buy?)
  • Cross-platform arbitrage detection

Focus on the notification logic. Let BuyWhere handle the data.

Get started at api.buywhere.ai