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,18 @@ class GoogleCSEPlugin(IntegrationPlugin):
 
     async def validate(self, fields: dict[str, str], mgr: IntegrationManager) -> tuple[dict, str | None]:
         key, cx = _require(fields, "api_key", "cx")
+        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://.")
         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})
+            r = await http.get(base_url or GOOGLE_CSE_URL, params={"key": key, "cx": cx, "q": "test", "num": 1})
         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
+        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

@@ -11,6 +11,8 @@
 
 log = logging.getLogger(__name__)
 
+GOOGLE_CSE_URL = "https://www.googleapis.com/customsearch/v1"
+
 
 def _ddg_sync(query: str, n: int, region: str) -> list[dict]:
     from ddgs import DDGS
@@ -36,9 +38,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",
+        r = await http.get(base_url or GOOGLE_CSE_URL,
                            params={"key": key, "cx": cx, "q": query, "num": min(n, 10)})
     r.raise_for_status()
     rows = r.json().get("items") or []
@@ -72,7 +74,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,60 @@
+"""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}
+
+
+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)