Webhooks

Receive SERP results automatically via HTTP callbacks instead of polling. Webhooks are the recommended approach for production applications processing large volumes of data.

Overview

When you include a postback_url in your crawl request, the SerpWatch API will POST the completed results to that URL when processing finishes. This eliminates the need to poll for results and enables real-time data processing.

Benefits of Webhooks

• No polling required - Results are pushed to you automatically
• Real-time delivery - Process results as soon as they're ready
• Efficient for batch - Perfect for processing thousands of keywords
• Reduced API calls - No need to repeatedly check task status

How It Works

Setting Up Webhooks

Add the postback_url parameter to any crawl request:

curl -X POST "https://engine.v2.serpwatch.io/api/v2/serp/crawl" \
  -H "Authorization: Bearer $SERPWATCH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "keyword": "project management software",
    "location_name": "United States",
    "iso_code": "US",
    "depth": 10,
    "device": "desktop",
    "postback_url": "https://yourapp.com/webhooks/serpwatch"
  }'

import requests
import os

API_KEY = os.environ.get("SERPWATCH_API_KEY")

response = requests.post(
    "https://engine.v2.serpwatch.io/api/v2/serp/crawl",
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    },
    json={
        "keyword": "project management software",
        "location_name": "United States",
        "iso_code": "US",
        "depth": 10,
        "device": "desktop",
        "postback_url": "https://yourapp.com/webhooks/serpwatch"
    }
)

task = response.json()
print(f"Task created: {task['id']}")
print("Results will be sent to your webhook URL when ready")

const response = await fetch(
  "https://engine.v2.serpwatch.io/api/v2/serp/crawl",
  {
    method: "POST",
    headers: {
      "Authorization": `Bearer ${API_KEY}`,
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      keyword: "project management software",
      location_name: "United States",
      iso_code: "US",
      depth: 10,
      device: "desktop",
      postback_url: "https://yourapp.com/webhooks/serpwatch"
    })
  }
);

const task = await response.json();
console.log(`Task created: ${task.id}`);
console.log("Results will be sent to your webhook URL when ready");

Webhook Payload

When a task completes, the API sends a POST request to your postback_url with the following JSON payload:

{
  "id": 1166085028196491264,
  "status": "success",
  "keyword": "project management software",
  "location_name": "United States",
  "iso_code": "US",
  "device": "desktop",
  "depth": 10,
  "language_code": "en",
  "created_at": 1769712336,
  "succeed_at": 1769712381,
  "result": {
    "organic": [
      {
        "position": 1,
        "title": "15 Best Project Management Software of 2026",
        "url": "https://example.com/pm-software",
        "domain": "example.com",
        "description": "Compare the top project management tools..."
      },
      {
        "position": 2,
        "title": "Top 10 Project Management Tools",
        "url": "https://reviews.com/pm-tools",
        "domain": "reviews.com",
        "description": "In-depth reviews of popular PM software..."
      }
    ],
    "ads": [],
    "local_pack": [],
    "featured_snippet": null,
    "total_results": "About 234,000,000 results",
    "search_time": 0.52
  },
  "top_domains": ["example.com", "reviews.com", "..."],
  "html_url": "https://storage.serpwatch.io/results/task_abc123xyz.html"
}

Receiving Webhooks

Your webhook endpoint should accept POST requests with JSON bodies. Here's how to implement a webhook handler:

from flask import Flask, request, jsonify
import json

app = Flask(__name__)

@app.route('/webhooks/serpwatch', methods=['POST'])
def handle_serpwatch_webhook():
    """
    Handle incoming webhook from SerpWatch API.
    """
    # Parse the JSON payload
    data = request.get_json()

    if not data:
        return jsonify({"error": "Invalid payload"}), 400

    # Extract key information
    task_id = data.get("id")
    status = data.get("status")
    keyword = data.get("keyword")

    print(f"Received webhook for task {task_id}")
    print(f"  Keyword: {keyword}")
    print(f"  Status: {status}")

    if status == "completed":
        # Process the results
        result = data.get("result", {})
        organic = result.get("organic", [])
        print(f"  Found {len(organic)} organic results")

        # Store results in your database
        save_results_to_database(data)

    elif status == "error":
        # Handle errors
        error_message = data.get("error_message", "Unknown error")
        print(f"  Error: {error_message}")

        # Log or retry as needed
        log_failed_task(data)

    # Return 200 to acknowledge receipt
    return jsonify({"status": "received"}), 200

def save_results_to_database(data):
    """Save webhook data to your database."""
    # Your database logic here
    pass

def log_failed_task(data):
    """Log failed task for review."""
    # Your error logging logic here
    pass

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

const express = require('express');
const app = express();

app.use(express.json());

app.post('/webhooks/serpwatch', (req, res) => {
  /**
   * Handle incoming webhook from SerpWatch API.
   */
  const data = req.body;

  if (!data || !data.id) {
    return res.status(400).json({ error: "Invalid payload" });
  }

  // Extract key information
  const { id: taskId, status, keyword, result } = data;

  console.log(`Received webhook for task ${taskId}`);
  console.log(`  Keyword: ${keyword}`);
  console.log(`  Status: ${status}`);

  if (status === "completed") {
    // Process the results
    const organic = result?.organic || [];
    console.log(`  Found ${organic.length} organic results`);

    // Store results in your database
    saveResultsToDatabase(data);

  } else if (status === "error") {
    // Handle errors
    const errorMessage = data.error_message || "Unknown error";
    console.log(`  Error: ${errorMessage}`);

    // Log or retry as needed
    logFailedTask(data);
  }

  // Return 200 to acknowledge receipt
  res.status(200).json({ status: "received" });
});

function saveResultsToDatabase(data) {
  // Your database logic here
}

function logFailedTask(data) {
  // Your error logging logic here
}

app.listen(5000, () => {
  console.log('Webhook server listening on port 5000');
});

Retry Behavior

If your webhook endpoint is unavailable or returns a non-2xx status code, the API will retry delivery with exponential backoff.

Security Considerations

Ensure your webhook endpoint is secure to prevent unauthorized access:

Use HTTPS

Always use HTTPS URLs for your webhook endpoints to encrypt data in transit. The API will not deliver webhooks to HTTP URLs in production.

Verify the Source

Validate that webhook requests are coming from SerpWatch:

# Option 1: Check for expected task IDs
# Store task IDs when you create them, verify incoming webhooks match

# Option 2: Use a secret token in your URL
# https://yourapp.com/webhooks/serpwatch?token=YOUR_SECRET_TOKEN

# Option 3: Verify by fetching the task from the API
def verify_webhook(task_id):
    """Verify webhook by checking task exists in API."""
    response = requests.get(
        f"https://engine.v2.serpwatch.io/api/v2/serp/crawl/{task_id}",
        headers={"Authorization": f"Bearer {API_KEY}"}
    )
    return response.status_code == 200

Testing Webhooks

Use these tools to test your webhook endpoint during development:

Local Development with ngrok

Use ngrok to expose your local server to the internet:

# Start your local server
python app.py  # Running on localhost:5000

# In another terminal, start ngrok
ngrok http 5000

# Use the ngrok URL as your postback_url
# https://abc123.ngrok.io/webhooks/serpwatch

Testing with curl

Simulate a webhook delivery to test your handler:

curl -X POST "http://localhost:5000/webhooks/serpwatch" \
  -H "Content-Type: application/json" \
  -d '{
    "id": "test_task_123",
    "status": "completed",
    "keyword": "test keyword",
    "result": {
      "organic": [
        {"position": 1, "title": "Test Result", "url": "https://example.com", "domain": "example.com"}
      ]
    }
  }'

Best Practices

• Respond quickly - Return 200 as soon as possible, process data asynchronously
• Handle duplicates - Use task ID to prevent processing the same result twice
• Log everything - Log all webhook deliveries for debugging
• Monitor failures - Set up alerts for webhook delivery failures
• Use a queue - For high volume, queue webhook payloads for background processing

# Recommended: Async processing pattern
@app.route('/webhooks/serpwatch', methods=['POST'])
def handle_webhook():
    data = request.get_json()

    # Quick validation
    if not data or not data.get("id"):
        return jsonify({"error": "Invalid"}), 400

    # Queue for async processing (don't block the response)
    task_queue.enqueue(process_serpwatch_result, data)

    # Return immediately
    return jsonify({"status": "queued"}), 200

Related Topics