nothingworry's picture
feat(web-search): use Google Custom Search for live web results
29116ed
import os
import httpx
from dotenv import load_dotenv
load_dotenv()
class WebClient:
"""
Communicates with the Google Custom Search API.
"""
def __init__(self) -> None:
self.search_endpoint = "https://www.googleapis.com/customsearch/v1"
async def search(self, query: str, max_results: int = 5, region: str = "us"):
"""
Sends the query to Google Custom Search and returns search results.
"""
max_results_value = self._sanitize_max_results(max_results)
api_key = os.getenv("GOOGLE_SEARCH_API_KEY")
cx_id = os.getenv("GOOGLE_SEARCH_CX_ID")
if not api_key or not cx_id:
raise RuntimeError("Google Custom Search credentials not configured.")
params = {
"key": api_key,
"cx": cx_id,
"q": query,
"num": max_results_value,
"gl": self._sanitize_region(region),
}
try:
async with httpx.AsyncClient(timeout=10) as client:
response = await client.get(self.search_endpoint, params=params)
response.raise_for_status()
except Exception as exc:
raise RuntimeError(f"Google Custom Search request failed: {exc}") from exc
data = response.json()
items = data.get("items", [])
return [
{
"title": item.get("title"),
"link": item.get("link"),
"snippet": item.get("snippet"),
}
for item in items
]
@staticmethod
def _sanitize_max_results(value: int) -> int:
try:
return max(1, min(int(value), 10))
except (TypeError, ValueError):
raise RuntimeError("max_results must be an integer between 1 and 10.")
@staticmethod
def _sanitize_region(region: str) -> str:
region_value = (region or "us").lower().split("-", 1)[0]
if len(region_value) != 2:
return "us"
return region_value