feat(integrations): let Google Custom Search use another API address

changes/+google-cse-address.added.md

@@ -0,0 +1,2 @@
+**Google Custom Search can use another API address.** An optional field when you connect it sends searches to a
+service that answers in the same format; left empty, searches go to Google as before.

sentient/integrations/plugins/alternatives.py

@@ -8,6 +8,7 @@
 
 from sentient.integrations.base import IntegrationError, IntegrationPlugin, SetupField
 from sentient.integrations.common import http_client
+from sentient.integrations.plugins.search import google_cse_url
 
 if TYPE_CHECKING:  # pragma: no cover
     from sentient.integrations.service import IntegrationManager
@@ -126,6 +127,9 @@ class GoogleCSEPlugin(IntegrationPlugin):
     setup_fields = [
         SetupField("api_key", "API key", secret=True, help="Google Cloud Console → APIs & Services → Credentials."),
         SetupField("cx", "Search engine ID", secret=False, help="From programmablesearchengine.google.com (the 'cx' value)."),
+        SetupField("base_url", "API address", required=False, placeholder="https://www.googleapis.com/customsearch/v1",
+                   help="Optional. Leave empty to use Google. Set it only for a service that answers in the "
+                        "Custom Search JSON format."),
     ]
     docs_url = "https://developers.google.com/custom-search/v1/overview"
     instructions_md = (
@@ -140,12 +144,31 @@ class GoogleCSEPlugin(IntegrationPlugin):
 
     async def validate(self, fields: dict[str, str], mgr: IntegrationManager) -> tuple[dict, str | None]:
         key, cx = _require(fields, "api_key", "cx")
-        async with http_client() as http:
-            r = await http.get("https://www.googleapis.com/customsearch/v1", params={"key": key, "cx": cx, "q": "test", "num": 1})
+        base_url = str(fields.get("base_url", "")).strip()
+        if base_url and not base_url.lower().startswith(("http://", "https://")):
+            raise IntegrationError("The API address has to start with http:// or https://.")
+        try:
+            async with http_client() as http:
+                r = await http.get(google_cse_url(base_url, {"key": key, "cx": cx, "q": "test", "num": 1}))
+        except (httpx.HTTPError, httpx.InvalidURL) as exc:
+            if not base_url:
+                raise
+            raise IntegrationError("Couldn't reach the API address. Check it and try again.") from exc
         if r.status_code == 400:
             raise IntegrationError("Google didn't accept the search engine ID. Check the 'cx' value.")
         _rejected(r, "Google Custom Search")
-        return {"api_key": key, "cx": cx}, None
+        if base_url:
+            try:
+                body = r.json()
+            except ValueError:
+                body = None
+            if not isinstance(body, dict):
+                raise IntegrationError("The API address didn't answer in the Custom Search JSON format. "
+                                       "Check the address.")
+        creds = {"api_key": key, "cx": cx}
+        if base_url:
+            creds["base_url"] = base_url
+        return creds, None
 
 
 class GoogleMapsPlugin(IntegrationPlugin):

sentient/integrations/plugins/search.py

@@ -5,12 +5,22 @@
 import asyncio
 import logging
 
+import httpx
+
 from sentient.integrations.base import IntegrationError, IntegrationPlugin, itool, manager_from
 from sentient.integrations.common import http_client
 from sentient.tools.base import ToolContext
 
 log = logging.getLogger(__name__)
 
+GOOGLE_CSE_URL = "https://www.googleapis.com/customsearch/v1"
+
+
+def google_cse_url(base_url: str, params: dict) -> httpx.URL:
+    """The address to call with the search parameters added to any it already has (httpx's ``params=`` would
+    replace a query such as ``?tenant=abc`` in a custom address)."""
+    return httpx.URL(base_url or GOOGLE_CSE_URL).copy_merge_params(params)
+
 
 def _ddg_sync(query: str, n: int, region: str) -> list[dict]:
     from ddgs import DDGS
@@ -36,10 +46,9 @@ async def brave_search(key: str, query: str, n: int) -> list[dict]:
     return [{"title": x.get("title"), "url": x.get("url"), "snippet": x.get("description")} for x in rows]
 
 
-async def google_cse_search(key: str, cx: str, query: str, n: int) -> list[dict]:
+async def google_cse_search(key: str, cx: str, query: str, n: int, base_url: str = "") -> list[dict]:
     async with http_client() as http:
-        r = await http.get("https://www.googleapis.com/customsearch/v1",
-                           params={"key": key, "cx": cx, "q": query, "num": min(n, 10)})
+        r = await http.get(google_cse_url(base_url, {"key": key, "cx": cx, "q": query, "num": min(n, 10)}))
     r.raise_for_status()
     rows = r.json().get("items") or []
     return [{"title": x.get("title"), "url": x.get("link"), "snippet": x.get("snippet")} for x in rows]
@@ -72,7 +81,7 @@ async def web_search(ctx: ToolContext, query: str, max_results: int = 8) -> dict
     elif provider == "google_cse":
         c = await mgr.get_credentials("google_cse") if await mgr.is_connected("google_cse") else None
         if c and c.get("api_key") and c.get("cx"):
-            results = await google_cse_search(c["api_key"], c["cx"], query, n)
+            results = await google_cse_search(c["api_key"], c["cx"], query, n, c.get("base_url", ""))
         else:
             note = "Google Custom Search isn't connected; used DuckDuckGo instead."
     elif provider == "searxng":

tests/integrations/test_google_cse_address.py

@@ -0,0 +1,101 @@
+"""Which address Google Custom Search calls: Google by default, the connected base_url when one is set."""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+import respx
+
+from sentient.integrations.base import IntegrationError
+from sentient.integrations.plugins.alternatives import GoogleCSEPlugin
+from sentient.integrations.plugins.search import GOOGLE_CSE_URL, google_cse_search
+
+OTHER = "http://search.example.test/customsearch/v1"
+ITEMS = {"items": [{"title": "T", "link": "https://example.com/a", "snippet": "S"}]}
+
+
+def test_default_address_is_google():
+    assert GOOGLE_CSE_URL == "https://www.googleapis.com/customsearch/v1"
+
+
+@respx.mock
+async def test_search_calls_google_without_base_url():
+    route = respx.get(GOOGLE_CSE_URL).mock(return_value=httpx.Response(200, json=ITEMS))
+    rows = await google_cse_search("k", "cx1", "sentient", 3)
+    assert route.called
+    assert route.calls.last.request.url.params["cx"] == "cx1"
+    assert rows == [{"title": "T", "url": "https://example.com/a", "snippet": "S"}]
+
+
+@respx.mock
+async def test_search_calls_base_url_with_the_same_params():
+    google = respx.get(GOOGLE_CSE_URL)
+    route = respx.get(OTHER).mock(return_value=httpx.Response(200, json=ITEMS))
+    rows = await google_cse_search("k", "cx1", "sentient", 3, OTHER)
+    params = route.calls.last.request.url.params
+    assert (params["key"], params["cx"], params["q"], params["num"]) == ("k", "cx1", "sentient", "3")
+    assert not google.called
+    assert rows[0]["url"] == "https://example.com/a"
+
+
+@respx.mock
+async def test_connect_without_base_url_checks_google_and_stores_no_address():
+    route = respx.get(GOOGLE_CSE_URL).mock(return_value=httpx.Response(200, json=ITEMS))
+    creds, _ = await GoogleCSEPlugin().validate({"api_key": "k", "cx": "cx1", "base_url": " "}, None)
+    assert route.called
+    assert creds == {"api_key": "k", "cx": "cx1"}
+
+
+@respx.mock
+async def test_connect_with_base_url_checks_it_and_stores_it():
+    google = respx.get(GOOGLE_CSE_URL)
+    route = respx.get(OTHER).mock(return_value=httpx.Response(200, json=ITEMS))
+    creds, _ = await GoogleCSEPlugin().validate({"api_key": "k", "cx": "cx1", "base_url": OTHER}, None)
+    assert route.called and not google.called
+    assert creds == {"api_key": "k", "cx": "cx1", "base_url": OTHER}
+
+
+@respx.mock
+async def test_search_keeps_the_query_already_in_base_url():
+    route = respx.get(OTHER).mock(return_value=httpx.Response(200, json=ITEMS))
+    await google_cse_search("k", "cx1", "sentient", 3, OTHER + "?tenant=abc")
+    params = route.calls.last.request.url.params
+    assert (params["tenant"], params["cx"], params["q"]) == ("abc", "cx1", "sentient")
+
+
+@respx.mock
+async def test_connect_keeps_the_query_already_in_base_url():
+    route = respx.get(OTHER).mock(return_value=httpx.Response(200, json=ITEMS))
+    creds, _ = await GoogleCSEPlugin().validate({"api_key": "k", "cx": "cx1", "base_url": OTHER + "?tenant=abc"}, None)
+    params = route.calls.last.request.url.params
+    assert (params["tenant"], params["cx"]) == ("abc", "cx1")
+    assert creds["base_url"] == OTHER + "?tenant=abc"
+
+
+@respx.mock
+async def test_connect_rejects_a_malformed_address():
+    with pytest.raises(IntegrationError):
+        await GoogleCSEPlugin().validate({"api_key": "k", "cx": "cx1", "base_url": "https://example.test:notaport"}, None)
+
+
+@respx.mock
+async def test_connect_reports_an_unreachable_address_as_an_integration_error():
+    respx.get(OTHER).mock(side_effect=httpx.ConnectError("refused"))
+    with pytest.raises(IntegrationError):
+        await GoogleCSEPlugin().validate({"api_key": "k", "cx": "cx1", "base_url": OTHER}, None)
+
+
+@pytest.mark.parametrize("answer", [
+    httpx.Response(200, text="<html>Sign in</html>", headers={"Content-Type": "text/html"}),
+    httpx.Response(200, json=[1, 2]),
+])
+@respx.mock
+async def test_connect_rejects_an_address_that_does_not_answer_with_a_json_object(answer):
+    respx.get(OTHER).mock(return_value=answer)
+    with pytest.raises(IntegrationError):
+        await GoogleCSEPlugin().validate({"api_key": "k", "cx": "cx1", "base_url": OTHER}, None)
+
+
+async def test_connect_rejects_a_non_http_address():
+    with pytest.raises(IntegrationError):
+        await GoogleCSEPlugin().validate({"api_key": "k", "cx": "cx1", "base_url": "file:///etc/passwd"}, None)