SerpApi SDK Patterns

Overview

Production patterns for SerpApi: typed result interfaces, response caching (critical since each search costs credits), multi-engine abstraction, and async search with the Searches Archive API.

Instructions

Step 1: Typed Result Interfaces

interface SerpApiOrganicResult {
  position: number;
  title: string;
  link: string;
  snippet: string;
  displayed_link: string;
  source?: string;
}

interface SerpApiSearchResult {
  search_metadata: { id: string; status: string; created_at: string };
  search_parameters: Record<string, string>;
  organic_results: SerpApiOrganicResult[];
  answer_box?: { answer?: string; snippet?: string; title?: string };
  knowledge_graph?: { title: string; description?: string; type?: string };
  related_questions?: Array<{ question: string; snippet: string }>;
  pagination?: { next: string };
}

Step 2: Cached Search Client

import { getJson } from 'serpapi';
import { LRUCache } from 'lru-cache';

const cache = new LRUCache<string, SerpApiSearchResult>({
  max: 500,
  ttl: 3600_000, // 1 hour -- search results are relatively stable
});

async function cachedSearch(params: Record<string, any>): Promise<SerpApiSearchResult> {
  const key = JSON.stringify(params);
  const cached = cache.get(key);
  if (cached) return cached;

  const result = await getJson({
    ...params,
    api_key: process.env.SERPAPI_API_KEY,
  }) as SerpApiSearchResult;

  cache.set(key, result);
  return result;
}

Step 3: Multi-Engine Search Abstraction

import serpapi, os

class SearchService:
    ENGINES = {
        "web": {"engine": "google", "query_param": "q"},
        "news": {"engine": "google_news", "query_param": "q"},
        "images": {"engine": "google_images", "query_param": "q"},
        "youtube": {"engine": "youtube", "query_param": "search_query"},
        "bing": {"engine": "bing", "query_param": "q"},
        "shopping": {"engine": "google_shopping", "query_param": "q"},
    }

    def __init__(self):
        self.client = serpapi.Client(api_key=os.environ["SERPAPI_API_KEY"])

    def search(self, query: str, engine: str = "web", **kwargs) -> dict:
        config = self.ENGINES[engine]
        params = {
            "engine": config["engine"],
            config["query_param"]: query,
            **kwargs,
        }
        return self.client.search(**params)

# Usage
svc = SearchService()
web = svc.search("Claude AI")

SerpApi Upgrade & Migration

Overview

The main migration path: google-search-results (legacy) to serpapi (current official package). The API itself is stable -- changes are in client library interfaces, not the REST API.

Instructions

Python: google-search-results to serpapi

# BEFORE: Legacy package
from serpapi import GoogleSearch
search = GoogleSearch({"q": "test", "api_key": key})
result = search.get_dict()

# AFTER: New official package
import serpapi
client = serpapi.Client(api_key=key)
result = client.search(engine="google", q="test")
# Result is already a dict -- no get_dict() needed

# Migration steps
pip uninstall google-search-results
pip install serpapi

# Update imports across codebase
# OLD: from serpapi import GoogleSearch
# NEW: import serpapi

Node.js: google-search-results-nodejs to serpapi

// BEFORE: Legacy
import { GoogleSearch } from 'google-search-results-nodejs';
const search = new GoogleSearch('api_key');
search.json({ q: 'test', engine: 'google' }, (result) => { ... });

// AFTER: Current (Promise-based)
import { getJson } from 'serpapi';
const result = await getJson({ engine: 'google', q: 'test', api_key: key });
// No callbacks -- uses Promises natively

Key Changes

Migration Checklist

• [ ] Replace package: pip install serpapi / npm install serpapi
• [ ] Update all imports
• [ ] Replace class-per-engine with engine parameter
• [ ] Replace callbacks with async/await (Node.js)
• [ ] Remove .get_dict() calls (Python -- result is already dict)
• [ ] Test all search queries return expected structure
• [ ] Update CI dependencies

Resources

• serpapi Python
• serpapi-webhooks-events SKILL.md View full skill →

  'Implement SerpApi async search callbacks and scheduled search monitoring.

  ReadWriteEditBash(curl:*)

  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):
          self.client = client
          self.db = db
  
      def track_keyword(self, keyword: str, domain: str):
          """Track a domain's ranking position for a keyword."""
          result = self.client.search(engine="google", q=keyword, num=100)
          organic = result.get("organic_results", [])
  
          position = None
          for r in organic:
              if domain in r.get("link", ""):
                  position = r["position"]
                  break
  
          self.db.insert({
              "keyword": keyword,
              "domain": domain,
              "position": position,  # None if not in top 100
              "total_results": result.get("search_information", {}).get("total_results"),
              "checked_at": datetime.utcnow().isoformat(),
              "search_id": result["search_metadata"]["id"],
          })
  
          return position
  
      def detect_changes(self, keyword: str, domain: str):
          """Compare current vs previous ranking."""
          current = self.track_keyword(keyword, domain)
          previous = self.db.get_previous_position(keyword, domain)
  
          if previous and current:
              change = previous - current  # Positive = improved
              if abs(change) >= 3:
                  self.notify(f"Ranking change for '{keyword}': {previous} -> {current} ({'+' if change > 0 else ''}{change})")
  

  Step 3: Schedul