Replace the Custom Search JSON API in 10 Lines of Python

The Custom Search JSON API stops working on January 1, 2027. If your Python code calls customsearch/v1 to get web results from the world’s most-used search engine, this post gets you to a working replacement in about 10 lines, then makes it safe to ship.

The steps:

1. Make the new call.
2. Map the fields your code reads and sends.
3. Handle errors by error.code.
4. Keep old callers working with a small adapter.
5. Run both APIs side by side before the cutover.

This is the short, runnable path. The Custom Search migration guide has the complete parameter and response mapping, and the shutdown checklist covers finding every call site and planning the cutover.

Before you start

You need a PacketStream account with funds and an API key:

• Create an account and add funds. The balance is prepaid, the minimum purchase is $50, and there is no subscription.
• Create a key on the Search API page of the dashboard. It is shown once, so copy it when it appears.

Load the key into your shell without writing it into the command history:

read -r -s -p 'PacketStream API key: ' PACKETSTREAM_API_KEY
printf '\n'
export PACKETSTREAM_API_KEY

The examples use Requests: pip install requests.

The 10-line replacement

import os
import requests

response = requests.post(
    "https://api.packetstream.io/v1/search",
    headers={"Authorization": f"Bearer {os.environ['PACKETSTREAM_API_KEY']}"},
    json={"query": "coffee grinders", "country": "us", "language": "en"},
    timeout=35,
)
response.raise_for_status()
for result in response.json()["results"]:
    print(result["position"], result["title"], result["url"])

That is a live web search from the United States in English, with up to 10 results. Compared with a Custom Search call, four things changed:

• Method and endpoint. POST https://api.packetstream.io/v1/search with a JSON body. GET with the same fields as query parameters also works, but POST keeps the query out of URLs and logs.
• The key moved to a header. Send it as Authorization: Bearer, never in the URL.
• There is no cx. The API searches the whole web, so there is no engine ID to send or configure.
• The timeout is 35 seconds. The server finishes or gives up within 25 seconds, so use a client timeout of at least 35. If your client gives up first, the search can still finish and be billed.

Map the fields you use

Most integrations touch only a handful of fields. These are the common ones; the migration guide lists every parameter, including the ones with no equivalent.

Two differences break code that is ported field by field:

• Unknown fields are rejected. A leftover cx or num returns 400 invalid_request instead of being ignored. Send only the fields in the table.
• htmlTitle, htmlSnippet, and pagemap do not exist. Results are plain text. If you used pagemap for page metadata, read the page with the Fetch API, which uses the same key.

Handle errors by code

Every error is JSON with a stable error.code and a readable error.message. Branch on the code, not on the message or only the HTTP status:

Failed searches are free, so a retry costs nothing until one succeeds. The one rare exception is a 500 internal_error, which can follow a charge that went through just before the failure; that is why it gets one retry, not several. The full list is in errors and limits.

Save this as packetstream_search.py. The search function sends one request body and returns the parsed response:

"""Search through the PacketStream Search API, retrying only what is safe to retry."""
import os
import random
import time

import requests

URL = "https://api.packetstream.io/v1/search"
WAIT_RETRY_AFTER = {"rate_limited", "concurrency_limited"}
BACK_OFF = {"search_unavailable", "capacity_exceeded", "search_disabled"}


class SearchError(Exception):
    def __init__(self, status, code, message):
        super().__init__(f"{status} {code}: {message}")
        self.status = status
        self.code = code


def search(body, attempts=4):
    headers = {"Authorization": f"Bearer {os.environ['PACKETSTREAM_API_KEY']}"}
    for attempt in range(attempts):
        response = requests.post(URL, headers=headers, json=body, timeout=35)
        if response.ok:
            return response.json()
        try:
            error = response.json()["error"]
        except (ValueError, KeyError):
            response.raise_for_status()  # not an API error body
        code = error["code"]
        retries_left = attempt < attempts - 1
        if code in WAIT_RETRY_AFTER and retries_left:
            time.sleep(float(response.headers.get("Retry-After", 1)))
        elif code in BACK_OFF and retries_left:
            time.sleep(2 ** attempt + random.random())
        elif code == "internal_error" and attempt == 0 and retries_left:
            time.sleep(1)
        else:
            raise SearchError(response.status_code, code, error["message"])
    raise AssertionError("unreachable")

A client timeout raises requests.Timeout and is deliberately not retried here: the search may still have run and been billed. Decide per job whether a duplicate is acceptable before you retry it.

Keep old callers working

If many places in your code read items and link, you do not have to change them all at once. Add this adapter to packetstream_search.py. It takes the Custom Search parameters you already build and returns the parts of a Custom Search response that most code reads:

from urllib.parse import urlsplit

TIME_RANGES = {"d1": "day", "w1": "week", "m1": "month", "y1": "year"}
IGNORED = {"key", "cx", "num"}
SUPPORTED = {"q", "siteSearch", "siteSearchFilter", "exactTerms", "excludeTerms",
             "fileType", "gl", "hl", "start", "dateRestrict", "safe"}


def custom_search(params):
    """Answer a Custom Search style call. Only the mapped parameters are supported."""
    unsupported = set(params) - IGNORED - SUPPORTED
    if unsupported:
        raise ValueError(f"no Search API equivalent for: {', '.join(sorted(unsupported))}")
    body = {"query": params["q"]}
    if "exactTerms" in params:
        body["query"] += ' "' + params["exactTerms"] + '"'
    if "excludeTerms" in params:
        body["query"] += "".join(f" -{word}" for word in params["excludeTerms"].split())
    if "fileType" in params:
        body["query"] += f" filetype:{params['fileType']}"
    if "siteSearch" in params:
        operator = "-site:" if params.get("siteSearchFilter") == "e" else "site:"
        body["query"] += f" {operator}{params['siteSearch']}"
    if "gl" in params:
        country = params["gl"].lower()
        body["country"] = "gb" if country == "uk" else country  # Custom Search used uk
    if "hl" in params:
        body["language"] = params["hl"]
    start = int(params.get("start", 1))
    if (start - 1) % 10:
        raise ValueError(f"start={start} has no page equivalent")
    body["page"] = (start - 1) // 10 + 1
    if "dateRestrict" in params:
        if params["dateRestrict"] not in TIME_RANGES:
            raise ValueError(f"dateRestrict={params['dateRestrict']} has no time_range equivalent")
        body["time_range"] = TIME_RANGES[params["dateRestrict"]]
    safe = params.get("safe", "off").lower()
    if safe in {"active", "high", "medium"}:  # high and medium are old names for active
        body["safe_search"] = True
    elif safe != "off":
        raise ValueError(f"safe={safe} has no safe_search equivalent")

    payload = search(body)
    return {
        "items": [
            {
                "title": result["title"],
                "link": result["url"],
                "snippet": result.get("snippet", ""),
                "displayLink": urlsplit(result["url"]).netloc,
            }
            for result in payload["results"]
        ]
    }

It ignores key, cx, and num, turns exactTerms, excludeTerms, and fileType into query operators, and sends gb when old code asks for uk. Any other parameter, such as searchType=image, raises ValueError, and so does a start or dateRestrict value with no equivalent, instead of quietly returning different results. Treat it as a bridge: once the cutover is done, move callers to results and url and delete it.

Test side by side before the cutover

While Custom Search still answers, run real queries through both and look at the differences. Put a few dozen queries from your logs in queries.txt, one per line, and set CSE_URL, CSE_API_KEY, and CSE_ID to the endpoint, key, and engine ID your code uses today:

"""Print result counts and shared URLs for each query on both APIs."""
import os
import sys

import requests

from packetstream_search import custom_search


def old_links(query):
    response = requests.get(
        os.environ["CSE_URL"],
        params={"key": os.environ["CSE_API_KEY"], "cx": os.environ["CSE_ID"],
                "q": query, "gl": "us", "hl": "en"},
        timeout=30,
    )
    response.raise_for_status()
    return [item["link"] for item in response.json().get("items", [])]


for query in filter(None, map(str.strip, open(sys.argv[1]))):
    old = old_links(query)
    new = [item["link"] for item in custom_search({"q": query, "gl": "us", "hl": "en"})["items"]]
    print(f"{len(old):>2} old {len(new):>2} new {len(set(old) & set(new)):>2} shared  {query}")

Run it as python3 side_by_side.py queries.txt. 200 queries cost $0.50 on the PacketStream side.

Do not expect identical lists. Your old engine applied its own configuration, and the Search API returns what a searcher in the country you choose sees. Look for:

• Empty results on one side. Those are the queries that break features.
• Low overlap on a site-limited engine. If your engine searched only a few sites, add siteSearch (or site: operators) on the new side, or the comparison measures the wrong thing.
• Errors. Any invalid_request points at a parameter you still send that has no mapping.

For a CSV report with URL normalization and a top-three check, use the comparison script in step 7 of the shutdown checklist.

Cost and limits

• Price. $0.0025 per successful search, which is $2.50 per 1,000, for every request type and option. Failed searches are free. A search that finds no results is still billed, because the search ran.
• Billing. A prepaid balance with a $50 minimum purchase and no subscription. GET /v1/account returns your balance and usage for free.
• Rate limit. 20 requests per second per key, with bursts of 200, and up to 32 searches running at once.

Ship it

1. Put custom_search behind the flag your code uses to choose a search provider, so you can switch back without a deploy.
2. Run the side-by-side test and fix every query that errors or comes back empty.
3. Move traffic over gradually and watch error codes and empty results for a few days.
4. Finish well before January 1, 2027, then move callers to the native results fields and remove the Custom Search keys.

The Search API docs cover the other request types, Markdown output, and batch searches. The Search API page has pricing and what the API returns.

FAQ

What replaces items and link in the response?

Read results instead of items, and url instead of link. title and snippet keep their names, and each result also has a 1-based position.

Is there an equivalent of cx?

No. The PacketStream Search API searches the whole web, so there is no engine to configure. If your engine was limited to a few sites, add site: operators to the query.

What timeout should I use?

At least 35 seconds. The server finishes or gives up on every request within 25 seconds, and if your client gives up first the search can still finish and be billed.