DEV Community

dodou
dodou

Posted on

SERP API + Slack Alert Bot: Build an SEO Anomaly Notifier in 30 Minutes

Project Goal

Build a Slack alert bot in 30 minutes:

  • Check 10 core keywords SERP ranking every hour
  • Send Slack alert if ranking changes more than 3 positions
  • New competitors / disappeared rankings trigger immediate alert
  • Daily 9am report (overall changes summary)

Step 1: Slack Setup (5 minutes)

Create Slack App

  1. Go to https://api.slack.com/apps
  2. Click "Create New App" → "From scratch"
  3. Name: "SERP Alert Bot", select workspace
  4. Click "Incoming Webhooks" → toggle on
  5. "Add New Webhook to Workspace" → select channel (#seo-alerts)
  6. Copy webhook URL (looks like https://hooks.slack.com/services/T.../B.../X...)

Slack Utility 

import requests
import os

SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK"]

def send_slack(text, blocks=None):
    payload = {"text": text}
    if blocks:
        payload["blocks"] = blocks
    r = requests.post(SLACK_WEBHOOK, json=payload, timeout=10)
    r.raise_for_status()

# Test
send_slack("SERP alert bot online")
Enter fullscreen mode Exit fullscreen mode

Step 2: Data Storage (3 minutes) 

import sqlite3
from datetime import datetime

def init_db():
    conn = sqlite3.connect("serp_history.db")
    c = conn.cursor()
    c.execute("""
        CREATE TABLE IF NOT EXISTS history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            ts TIMESTAMP NOT NULL,
            keyword TEXT NOT NULL,
            position INTEGER,
            url TEXT,
            title TEXT,
            new_competitor TEXT,
            disappeared BOOLEAN DEFAULT 0
        )
    """)
    c.execute("CREATE INDEX IF NOT EXISTS idx_keyword_ts ON history(keyword, ts)")
    conn.commit()
    conn.close()
Enter fullscreen mode Exit fullscreen mode

Step 3: SERP Query (5 minutes) 

import os
import requests

API_KEY = os.environ["SERPBASE_KEY"]
ENDPOINT = "https://api.serpbase.dev/google/search"
DOMAIN = "yourdomain.com"
KEYWORDS = [
    "your brand keyword 1",
    "your brand keyword 2",
    "your product keyword 1",
    "your product keyword 2",
    "your product keyword 3",
    # up to 10
]

def fetch_serp(keyword, gl="us", hl="en", num=10):
    r = requests.post(
        ENDPOINT,
        headers={"X-API-Key": API_KEY},
        json={"q": keyword, "gl": gl, "hl": hl, "num": num},
        timeout=10,
    )
    r.raise_for_status()
    return r.json()

def get_my_position(data, domain=DOMAIN):
    for i, item in enumerate(data.get("organic", []), 1):
        if domain in item.get("link", ""):
            return i, item.get("link"), item.get("title")
    return None, None, None

def get_competitors(data, exclude_domain=DOMAIN, top_n=10):
    """Extract top 10 competitor domains"""
    competitors = set()
    for item in data.get("organic", [])[:top_n]:
        link = item.get("link", "")
        if exclude_domain not in link:
            domain = "/".join(link.split("/")[:3])
            competitors.add(domain)
    return list(competitors)
Enter fullscreen mode Exit fullscreen mode

Step 4: Anomaly Detection (10 minutes) 

import sqlite3
from datetime import datetime, timedelta

def check_changes(threshold=3):
    """Compare with last data, alert on changes > threshold"""
    conn = sqlite3.connect("serp_history.db")
    c = conn.cursor()

    alerts = []

    for kw in KEYWORDS:
        # Get last 2 data points
        c.execute("""
            SELECT ts, position, url, new_competitor, disappeared
            FROM history
            WHERE keyword = ?
            ORDER BY ts DESC LIMIT 2
        """, (kw,))
        rows = c.fetchall()

        if len(rows) < 2:
            continue

        current_ts, current_pos, current_url, current_competitor, _ = rows[0]
        last_ts, last_pos, last_url, last_competitor, last_disappeared = rows[1]

        # Ranking change
        if current_pos and last_pos:
            delta = last_pos - current_pos
            if abs(delta) >= threshold:
                direction = "up" if delta > 0 else "down"
                alerts.append({
                    "keyword": kw,
                    "type": "rank_change",
                    "old": last_pos,
                    "new": current_pos,
                    "delta": delta,
                    "direction": direction,
                })

        # Disappeared (had rank, now no)
        if last_pos and not current_pos:
            alerts.append({
                "keyword": kw,
                "type": "disappeared",
                "old": last_pos,
                "new": None,
            })

        # Appeared (no rank before, now has)
        if not last_pos and current_pos:
            alerts.append({
                "keyword": kw,
                "type": "appeared",
                "old": None,
                "new": current_pos,
            })

        # New competitor
        if current_competitor and current_competitor != last_competitor:
            alerts.append({
                "keyword": kw,
                "type": "new_competitor",
                "competitor": current_competitor,
            })

    conn.close()
    return alerts

def save_check(keyword, position, url, new_competitor):
    conn = sqlite3.connect("serp_history.db")
    c = conn.cursor()
    c.execute("""
        INSERT INTO history (ts, keyword, position, url, new_competitor)
        VALUES (?, ?, ?, ?, ?)
    """, (datetime.now().isoformat(), keyword, position, url, new_competitor))
    conn.commit()
    conn.close()
Enter fullscreen mode Exit fullscreen mode

Step 5: Slack Notification (5 minutes) 

def send_alerts(alerts):
    if not alerts:
        return

    # Categorize
    rank_changes = [a for a in alerts if a["type"] == "rank_change"]
    disappeared = [a for a in alerts if a["type"] == "disappeared"]
    appeared = [a for a in alerts if a["type"] == "appeared"]
    new_competitors = [a for a in alerts if a["type"] == "new_competitor"]

    blocks = [
        {"type": "header", "text": {"type": "plain_text", "text": f"SERP Alert ({len(alerts)} changes)"}},
    ]

    if rank_changes:
        text = ""
        for a in rank_changes:
            arrow = "UP" if a["delta"] > 0 else "DOWN"
            text += f"• {a['keyword']}: #{a['old']} -> #{a['new']} {arrow}{abs(a['delta'])}\n"
        blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": f"*Rank Changes*\n{text}"}})

    if disappeared:
        text = "\n".join(f"• {a['keyword']} (was #{a['old']})" for a in disappeared)
        blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": f"*Disappeared*\n{text}"}})

    if appeared:
        text = "\n".join(f"• {a['keyword']}: entered #{a['new']}" for a in appeared)
        blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": f"*New Entries*\n{text}"}})

    if new_competitors:
        text = "\n".join(f"• {a['keyword']}: {a['competitor']} entered" for a in new_competitors)
        blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": f"*New Competitors*\n{text}"}})

    send_slack("", blocks=blocks)

def send_daily_report():
    """Send daily 9am summary"""
    conn = sqlite3.connect("serp_history.db")
    c = conn.cursor()

    today = datetime.now().date().isoformat()
    c.execute("""
        SELECT keyword, position, url FROM history
        WHERE ts LIKE ?
        ORDER BY keyword
    """, (f"{today}%",))
    rows = c.fetchall()
    conn.close()

    if not rows:
        send_slack("No SERP data today")
        return

    text = f"*SERP Daily Report ({today})*\n\n"
    for kw, pos, url in rows:
        text += f"• {kw}: #{pos or 'unranked'} {url or ''}\n"

    send_slack(text)
Enter fullscreen mode Exit fullscreen mode

Step 6: Main Loop (2 minutes) 

import schedule
import time

def hourly_check():
    """Check every hour, alert on changes"""
    for kw in KEYWORDS:
        data = fetch_serp(kw)
        pos, url, _ = get_my_position(data)
        new_competitors = get_competitors(data)
        new_comp = new_competitors[0] if new_competitors else None
        save_check(kw, pos, url, new_comp)

    alerts = check_changes(threshold=3)
    send_alerts(alerts)

if __name__ == "__main__":
    init_db()

    # Run once at startup
    hourly_check()

    # Daily 9am report
    schedule.every().day.at("09:00").do(send_daily_report)

    # Every hour check
    schedule.every().hour.do(hourly_check)

    while True:
        schedule.run_pending()
        time.sleep(60)
Enter fullscreen mode Exit fullscreen mode

Deployment Options

Option Best for
VPS ($5/month) 24/7 run
Cloud Function (scheduled) Serverless, cheap
GitHub Actions Free, hourly limit

Slack Message Example 

[Header] SERP Alert (3 changes)

[Section] Rank Changes
• your brand keyword 1: #3 -> #6 DOWN3
• your product keyword 1: #5 -> #2 UP3

[Section] New Competitors
• your product keyword 2: competitor-x.com entered
Enter fullscreen mode Exit fullscreen mode

5 Advanced Directions

  1. Add PAA change alerts: People Also Ask new questions = content opportunity
  2. Add AI Overview monitoring: brand citation in AI answers
  3. Add multi-region monitoring: us / uk / cn separately
  4. Add dashboard: weekly summary chart to Slack channel
  5. Add anomaly detection: ML detect unusual volatility (Google update mass alerts)

100 free searches: serpbase.dev signup, small-scale 10 keywords / 1 week test.

Top comments (0)