serpapi-webhooks-events

name	serpapi-webhooks-events
description	Implement SerpApi async search callbacks and scheduled search monitoring. Use when setting up search monitoring, SERP tracking pipelines, or async search result retrieval. Trigger: "serpapi webhooks", "serpapi monitoring", "serpapi scheduled search", "serpapi async".
allowed-tools	Read, Write, Edit, Bash(curl:*)
version	1.4.0
license	MIT
author	Jeremy Longshore <[e-mail]>
tags	["saas","search","seo","serpapi"]
compatibility	Designed for Claude Code

SerpApi Webhooks & Events

Overview

SerpApi does not have traditional webhooks, but supports async searches and the Searches Archive API. Build SERP monitoring by combining scheduled searches with change detection. Common use case: track keyword rankings over time.

Instructions

Step 1: Async Search with Polling

import serpapi, os, time

client = serpapi.Client(api_key=os.environ["SERPAPI_API_KEY"])

# Submit async search (returns immediately)
result = client.search(engine="google", q="your keyword", async_search=True)
search_id = result["search_metadata"]["id"]
print(f"Submitted: {search_id}")

# Poll for completion
while True:
    archived = client.search(engine="google", search_id=search_id)
    status = archived["search_metadata"]["status"]
    if status == "Success":
        break
    elif status == "Error":
        raise Exception(f"Search failed: {archived.get('error')}")
    time.sleep(2)

print(f"Results: {len(archived.get('organic_results', []))}")

Step 2: SERP Monitoring Pipeline

import json, hashlib
from datetime import datetime

class SerpMonitor:
    def __init__(self, client, db):
        .client = client
        .db = db

     ():
        
        result = .client.search(engine=, q=keyword, num=)
        organic = result.get(, [])

        position = 
         r  organic:
             domain  r.get(, ):
                position = r[]
                

        .db.insert({
            : keyword,
            : domain,
            : position,  
            : result.get(, {}).get(),
            : datetime.utcnow().isoformat(),
            : result[][],
        })

         position

     ():
        
        current = .track_keyword(keyword, domain)
        previous = .db.get_previous_position(keyword, domain)

         previous  current:
            change = previous - current  
             (change) >= :
                .notify()