LangChain y Google Custom Search: alternativa antes del cierre

Reemplaza GoogleSearchAPIWrapper antes de que Google Custom Search termine el 1 de enero de 2027: una tool de LangChain probada que conserva title, link y snippet.

GoogleSearchAPIWrapper y las tools google_search construidas sobre él llaman a la Custom Search JSON API de Google, que Google ha cerrado a clientes nuevos y que termina el 1 de enero de 2027. Nada en el wrapper puede esquivar eso, así que la solución es cambiar el backend. Abajo tienes una tool de LangChain de 79 líneas, ejecutada el 19 de octubre de 2026, que mantiene los mismos nombres de tool y los mismos campos title, link y snippet, de modo que las cadenas y los prompts de agentes que ya tienes siguen funcionando. Una búsqueda cuesta 5 créditos de FastAF.

¿Qué deja de funcionar exactamente en LangChain el 1 de enero de 2027?

La descripción general de la Custom Search JSON API de Google dice que la API “está cerrada a clientes nuevos” y que los clientes existentes “tienen hasta el 1 de enero de 2027 para pasar a una solución alternativa” (consultado el 19 de octubre de 2026). Su única alternativa con nombre, Vertex AI Search, se describe como “una alternativa favorable para buscar en hasta 50 dominios”, que es búsqueda dentro de un sitio, no búsqueda web.

En langchain-google-community 5.1.1 (instalado hoy en un venv limpio, con langchain 1.4.6 y langchain-core 1.6.12), GoogleSearchAPIWrapper construye un cliente customsearch v1 a partir de GOOGLE_API_KEY y un GOOGLE_CSE_ID, y cada llamada pasa por cse().list(). La copia anterior en langchain_community ya está marcada como obsoleta. La página de Google Search de LangChain sigue documentando el wrapper sin ninguna nota sobre el cierre, y su índice de tools lista decenas de otras integraciones de búsqueda. No pude probar el comportamiento después de la fecha de cierre, así que esa parte es lo que dice Google, no una medición mía.

Para una visión más amplia, mira la API oficial de búsqueda de Google vs las SERP API y la búsqueda web para agentes de IA tras Bing. Este post trata solo del cambio en LangChain.

¿Cómo reemplazo GoogleSearchAPIWrapper con una tool que lo sustituya sin cambios?

El wrapper antiguo tiene dos llamadas de las que dependen las cadenas: run(query), que une los snippets en un solo string, y results(query, num_results, search_params), que devuelve dicts con title, link y snippet, o [{"Result": "No good Google Search Result was found"}] cuando no hay nada. Las tools google_search y google_search_results_json se apoyan en ellas. Este módulo reproduce todo eso, síncrono y asíncrono, sobre el 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),
    ]

Úsalo donde usabas el par antiguo:

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)

Hay un detalle que está medido, no supuesto. En 20 consultas variadas (192 resultados, 100 créditos, 19 de octubre de 2026), 38 resultados (el 20 %) traían un inline_snippet. En 36 de ellos, el campo snippet solo contenía etiquetas de sitelinks como “More results from www.reddit.com”, y la descripción real estaba en inline_snippet. El módulo prefiere inline_snippet, que es lo que se parecía al antiguo snippet. Todas las consultas devolvieron 9 o 10 resultados orgánicos.

¿Qué ejecuté realmente y cuánto costó?

No tenía ninguna API key de LLM en este entorno, así que no interviene ningún modelo real. Ejecuté las tools directamente (invoke y ainvoke), y luego dentro de un bucle create_agent movido por un modelo de chat falso con guion que reproduce tres llamadas a tools. Eso demuestra el cableado en el runtime de LangChain, no el comportamiento de un modelo.

Contra https://api.fastaf.co/v1 el 19 de octubre de 2026:

Ejecución Latencia Créditos Resultado
google_search_results_json, síncrona 3,20 s 5 5 resultados, claves link, snippet, title
La misma consulta otra vez 0,0002 s 0 servida desde la caché
google_search, asíncrona 4,74 s 5 1.642 caracteres de snippets unidos
Agente con guion, 3 llamadas a tools, presupuesto 10 5,8 s 10 la tercera llamada fue rechazada con un mensaje

Ejecuciones anteriores del mismo código en producción tardaron 6,9 y 8,1 s por búsqueda, así que cuenta con entre 3 y 8 segundos, no con milisegundos. En la ejecución de 20 consultas con concurrencia 4, la mediana fue de 4,0 s y el máximo de 12,4 s. Las solicitudes fallidas no se cobran. La comprobación del presupuesto se afinó después de estas ejecuciones (ahora reserva el coste antes de la solicitud) y se repitió solo con un stub: con max_credits=6, dos llamadas paralelas a aresults dejan pasar una y rechazan la otra.

¿Cómo se corresponden los parámetros antiguos con la tool nueva?

GoogleSearchAPIWrapper Esta tool
GOOGLE_API_KEY + GOOGLE_CSE_ID solo FASTAF_API_KEY, sin ID de motor
k, num_results mismos nombres; máximo 10 por llamada, como el num de Google
search_params={"gl": "de"} {"country": "de"}
{"hl": "de"} {"lang": "de"}
{"lr": ...}, {"cr": ...}, {"safe": "active"} mismos nombres
{"start": 11} (base 1) {"page": 1} (base cero; la página 1 devolvió resultados distintos a la 0)
siterestrict=True, siteSearch pon site:example.com en la consulta (probado)
exactTerms, excludeTerms comillas y -término en la consulta
dateRestrict="w1" {"tbs": "qdr:w"}, un string de filtro de Google (aceptado; no verifiqué el filtrado por fecha)
campos image, thumbnail no se devuelven
searchType="image" no lo cubre este endpoint

La diferencia de comportamiento es la más importante. Custom Search consultaba un motor que tú configurabas. Esto devuelve lo que google.com devolvió para la consulta, así que el ranking será distinto de tus resultados anteriores. Revisa cualquier conjunto de evaluación que dependiera de URLs concretas. El parámetro results de la API no redujo el tamaño de la página en mis pruebas (results=3 y results=20 devolvieron 9 las dos), así que el módulo recorta en el cliente.

¿Cómo cacheo y limito el gasto por ejecución?

Las dos cosas viven en el módulo. ttl cachea por consulta, cantidad y parámetros, de modo que un agente que repite una búsqueda paga una sola vez. max_credits reserva los 5 créditos de una búsqueda antes de la solicitud, los cambia por la cabecera X-FastAF-Credits-Used de la respuesta (un error de la API los devuelve) y lanza ToolException cuando la siguiente búsqueda ya no cabría; con handle_tool_error=True, el agente recibe “Search budget of 10 credits is used up.” como un resultado normal de la tool y puede razonar sobre ello, como en la tabla de arriba. El wrapper no guarda estado entre procesos, y la caché es en memoria.

El saldo de la cuenta es el límite exterior: un saldo agotado devuelve 402 y lo detiene todo. Dale a un agente de IA búsqueda web en vivo con un límite de gasto estricto cubre las capas que rodean esto, y SERP API para RAG explica qué hacer con los resultados.

¿Puedo usar MCP en lugar de una tool propia?

Sí. FastAF tiene un servidor MCP alojado con una tool search_web. Lo cargué con langchain-mcp-adapters 0.3.2 (MultiServerMCPClient, transporte streamable_http, cabecera Bearer), obtuve las 15 tools, y search_web devolvió 3 resultados por 5 créditos en 4,9 s. Dos pegas: devuelve url en lugar de link, y pasa snippet sin modificar, así que te llegan las etiquetas de sitelinks que describí antes. Para las cadenas que ya leen link, quédate con la tool propia. Para un agente nuevo que además deba abrir páginas, usa MCP.

En resumen

Google Custom Search termina el 1 de enero de 2027 y GoogleSearchAPIWrapper se va con él. Copia el módulo, conserva los nombres de las tools, revisa tus tests sensibles al ranking y fija max_credits para que un agente en bucle no pueda disparar la factura. Una búsqueda son 5 créditos, de 1,33 $ a 2,50 $ por 1.000 según el pack, y el nivel gratuito son 500 créditos, unas 100 búsquedas. Consigue una clave gratis y pruébalo con tus propias consultas.