Tool di ricerca Google per LangChain dopo la chiusura di Custom Search

GoogleSearchAPIWrapper e i tool google_search costruiti sopra chiamano la Google Custom Search JSON API, che Google ha chiuso ai nuovi clienti e che termina il 1° gennaio 2027. Nel wrapper non c’è nulla che possa aggirare la cosa, quindi la soluzione è cambiare il backend. Qui sotto trovi un tool LangChain di 79 righe, eseguito il 19 ottobre 2026, che mantiene gli stessi nomi di tool e gli stessi campi title, link e snippet, così le catene e i prompt degli agenti esistenti continuano a funzionare. Una ricerca costa 5 crediti FastAF.

Cosa smette di funzionare esattamente in LangChain il 1° gennaio 2027?

La panoramica della Custom Search JSON API di Google dice che l’API “is closed to new customers” e che i clienti esistenti “have until January 1, 2027 to transition to an alternative solution” (controllato il 19 ottobre 2026). La sua unica alternativa citata, Vertex AI Search, è descritta come “a favorable alternative for searching up to 50 domains”, cioè ricerca dentro un sito, non ricerca web.

In langchain-google-community 5.1.1 (installato oggi in un venv pulito, con langchain 1.4.6 e langchain-core 1.6.12), GoogleSearchAPIWrapper costruisce un client customsearch v1 da GOOGLE_API_KEY e da un GOOGLE_CSE_ID, e ogni chiamata passa da cse().list(). La copia più vecchia in langchain_community è già marcata come deprecata. La pagina Google Search di LangChain documenta ancora il wrapper senza nessuna nota sulla chiusura, e il suo indice dei tool elenca decine di altre integrazioni di ricerca. Non ho potuto testare il comportamento dopo la data di chiusura, quindi quella parte è una dichiarazione di Google, non una mia misura.

Per il quadro più ampio vedi l’API di ricerca ufficiale di Google contro le SERP API e la ricerca web per agenti AI dopo Bing. Questo articolo riguarda solo la sostituzione in LangChain.

Come sostituisco GoogleSearchAPIWrapper con un tool drop-in?

Il vecchio wrapper ha due chiamate da cui dipendono le catene: run(query), che unisce gli snippet in un’unica stringa, e results(query, num_results, search_params), che restituisce dizionari con title, link e snippet, oppure [{"Result": "No good Google Search Result was found"}] quando è vuoto. I tool google_search e google_search_results_json poggiano su queste. Questo modulo riproduce tutto, sync e async, sopra l’endpoint /v1/serp:

import os, time, httpx
from langchain_core.tools import StructuredTool, ToolException

BASE = "https://api.fastaf.co/v1"
NOTHING = [{"Result": "No good Google Search Result was found"}]
COST = 5  # credits per search

class FastAFSearch:
    """Same surface as GoogleSearchAPIWrapper: run(), results(), arun(), aresults()."""

    def __init__(self, k=10, max_credits=None, ttl=300, **defaults):
        self.k, self.max_credits, self.ttl, self.defaults = k, max_credits, ttl, defaults
        self.spent, self._cache = 0, {}
        self._headers = {"Authorization": f"Bearer {os.environ['FASTAF_API_KEY']}"}

    def _hit(self, key):
        hit = self._cache.get(key)
        return hit[1] if hit and time.time() - hit[0] < self.ttl else None

    def _params(self, query, extra):
        if self.max_credits is not None and self.spent + COST > self.max_credits:
            raise ToolException(f"Search budget of {self.max_credits} credits is used up.")
        self.spent += COST  # reserve before the request, so parallel async calls cannot both pass
        return {"query": query, **self.defaults, **(extra or {})}

    def _parse(self, key, resp, num):
        # An API error is never charged: give the reservation back. A 200 reports its real cost.
        self.spent += (int(resp.headers["X-FastAF-Credits-Used"]) if resp.status_code == 200 else 0) - COST
        if resp.status_code != 200:
            raise ToolException(f"Search failed: HTTP {resp.status_code}")
        out = [{"title": r["title"], "link": r["link"],
                "snippet": r.get("inline_snippet") or r.get("snippet", "")}
               for r in resp.json()["organic_results"][:min(num, 10)]]
        self._cache[key] = (time.time(), out or NOTHING)
        return out or NOTHING

    def results(self, query, num_results, search_params=None):
        key = (query, num_results, str(search_params))
        if (hit := self._hit(key)) is not None:
            return hit
        resp = httpx.get(f"{BASE}/serp", params=self._params(query, search_params),
                         headers=self._headers, timeout=60)
        return self._parse(key, resp, num_results)

    async def aresults(self, query, num_results, search_params=None):
        key = (query, num_results, str(search_params))
        if (hit := self._hit(key)) is not None:
            return hit
        async with httpx.AsyncClient(timeout=60) as c:
            resp = await c.get(f"{BASE}/serp", params=self._params(query, search_params),
                               headers=self._headers)
        return self._parse(key, resp, num_results)

    @staticmethod
    def _join(rs):
        return " ".join(r["snippet"] for r in rs if "snippet" in r) or NOTHING[0]["Result"]

    def run(self, query: str) -> str:
        return self._join(self.results(query, self.k))

    async def arun(self, query: str) -> str:
        return self._join(await self.aresults(query, self.k))

def fastaf_search_tools(api: FastAFSearch, num_results: int = 4):
    """Two tools with the names and descriptions of GoogleSearchRun / GoogleSearchResults."""
    def as_json(query: str) -> str:
        return str(api.results(query, num_results))
    async def aas_json(query: str) -> str:
        return str(await api.aresults(query, num_results))
    desc = ("A wrapper around Google Search. Useful for when you need to answer questions "
            "about current events. Input should be a search query.")
    return [
        StructuredTool.from_function(func=api.run, coroutine=api.arun, name="google_search",
                                     description=desc, handle_tool_error=True),
        StructuredTool.from_function(func=as_json, coroutine=aas_json,
                                     name="google_search_results_json",
                                     description=desc + " Output is a JSON array of the query results",
                                     handle_tool_error=True),
    ]

Usalo dove usavi la vecchia coppia:

from fastaf_search_tools import FastAFSearch, fastaf_search_tools

api = FastAFSearch(country="us", max_credits=50)
tools = fastaf_search_tools(api, num_results=5)   # drop into create_agent(model, tools=tools)

Un dettaglio è misurato, non supposto. Su 20 query varie (192 risultati, 100 crediti, 19 ottobre 2026), 38 risultati (20%) avevano un inline_snippet. In 36 di questi il campo snippet conteneva solo etichette di sitelink come “More results from www.reddit.com”, e la descrizione vera stava in inline_snippet. Il modulo preferisce inline_snippet, che è come si presentava il vecchio snippet. Ogni query ha restituito 9 o 10 risultati organici.

Cosa ho eseguito davvero e quanto è costato?

In questo ambiente non avevo nessuna chiave API di un LLM, quindi non c’è nessun modello vero. Ho eseguito i tool direttamente (invoke e ainvoke), poi dentro un loop create_agent guidato da un finto chat model scriptato che riproduce tre chiamate ai tool. Questo dimostra il collegamento nel runtime di LangChain, non il comportamento di un modello.

Contro https://api.fastaf.co/v1 il 19 ottobre 2026:

Esecuzione	Latenza	Crediti	Risultato
`google_search_results_json`, sync	3,20 s	5	5 risultati, chiavi `link`, `snippet`, `title`
Stessa query di nuovo	0,0002 s	0	servita dalla cache
`google_search`, async	4,74 s	5	1.642 caratteri di snippet uniti
Agente scriptato, 3 chiamate ai tool, budget 10	5,8 s	10	terza chiamata rifiutata con un messaggio

Esecuzioni precedenti dello stesso codice in produzione hanno impiegato 6,9 e 8,1 s per ricerca, quindi calcola da 3 a 8 secondi circa, non millisecondi. Nell’esecuzione da 20 query con concorrenza 4, la mediana è stata 4,0 s e il massimo 12,4 s. Le richieste fallite non vengono addebitate. Il controllo del budget è stato rafforzato dopo queste esecuzioni (ora prenota il costo prima della richiesta) e riprovato solo con uno stub: con max_credits=6, due chiamate parallele a aresults ne lasciano passare una e rifiutano l’altra.

Come si mappano i vecchi parametri sul nuovo tool?

GoogleSearchAPIWrapper	Questo tool
`GOOGLE_API_KEY` + `GOOGLE_CSE_ID`	solo `FASTAF_API_KEY`, nessun ID del motore
`k`, `num_results`	stessi nomi; massimo 10 per chiamata, come il `num` di Google
`search_params={"gl": "de"}`	`{"country": "de"}`
`{"hl": "de"}`	`{"lang": "de"}`
`{"lr": ...}`, `{"cr": ...}`, `{"safe": "active"}`	stessi nomi
`{"start": 11}` (base 1)	`{"page": 1}` (base zero; la pagina 1 ha restituito risultati diversi dalla pagina 0)
`siterestrict=True`, `siteSearch`	metti `site:example.com` nella query (testato)
`exactTerms`, `excludeTerms`	virgolette e `-termine` nella query
`dateRestrict="w1"`	`{"tbs": "qdr:w"}`, una stringa di filtro di Google (accettata; non ho verificato il filtro per data)
campi `image`, `thumbnail`	non restituiti
`searchType="image"`	non coperto da questo endpoint

La differenza di comportamento è quella più grossa. Custom Search interrogava un motore che avevi configurato tu. Questo restituisce quello che google.com ha restituito per la query, quindi le posizioni saranno diverse dai tuoi vecchi risultati. Ricontrolla qualsiasi set di valutazione agganciato a URL specifici. Il parametro results dell’API non ha ridotto la dimensione della pagina nei miei test (results=3 e results=20 hanno restituito entrambi 9), quindi il modulo taglia lato client.

Come faccio cache e tetto di spesa per esecuzione?

Entrambi stanno nel modulo. ttl mette in cache per query, numero e parametri, così un agente che ripete una ricerca la paga una volta sola. max_credits prenota i 5 crediti di una ricerca prima della richiesta, li sostituisce con l’header X-FastAF-Credits-Used della risposta (un errore dell’API li restituisce) e solleva ToolException quando la ricerca successiva non ci starebbe; con handle_tool_error=True l’agente riceve “Search budget of 10 credits is used up.” come normale risultato del tool e può ragionarci, come nella tabella sopra. Il wrapper non conserva stato fra processi, e la cache è in memoria.

Il saldo sull’account è il limite esterno: un saldo esaurito restituisce 402 e ferma tutto. Dare a un agente AI la ricerca web con un tetto di spesa tratta i livelli attorno a questo, e SERP API per RAG tratta cosa fare con i risultati.

Posso usare MCP invece di un tool custom?

Sì. FastAF espone un server MCP ospitato con un tool search_web. L’ho caricato con langchain-mcp-adapters 0.3.2 (MultiServerMCPClient, trasporto streamable_http, header Bearer), ho ottenuto tutti i 15 tool, e search_web ha restituito 3 risultati per 5 crediti in 4,9 s. Due cose da sapere: restituisce url invece di link, e passa snippet senza modifiche, quindi ottieni le etichette di sitelink descritte sopra. Per le catene esistenti che leggono link, tieni il tool custom. Per un agente nuovo che deve anche aprire pagine, usa MCP.

La versione breve

Google Custom Search termina il 1° gennaio 2027 e GoogleSearchAPIWrapper si ferma con lui. Copia il modulo, mantieni i nomi dei tool, controlla i test sensibili alle posizioni, e imposta max_credits perché un agente in loop non possa gonfiare il conto. Una ricerca costa 5 crediti, da $1,33 a $2,50 ogni 1.000 a seconda del pacchetto, e il livello gratuito è di 500 crediti, circa 100 ricerche. Prendi una chiave gratuita e provala sulle tue query.