"""Web search tool using Google Custom Search API."""
import logging
import httpx
from pydantic import BaseModel, Field
from ..bot import BotContext
from .base import ToolResult
from .registry import folder_bot, get_services, get_tool_config
logger = logging.getLogger(__name__)
GOOGLE_SEARCH_URL = "https://www.googleapis.com/customsearch/v1"
[docs]
class WebSearchRequest(BaseModel, frozen=True):
"""Request for searching the web."""
query: str = Field(description="Search query to look up on the web")
max_results: int = Field(
default=5,
description="Maximum number of search results to return (1-10)",
)
[docs]
@folder_bot.tool(
name="web_search",
request_type=WebSearchRequest,
response_type=ToolResult,
)
async def web_search(
request: WebSearchRequest, _context: BotContext | None = None
) -> ToolResult:
"""Search the web using Google.
Returns titles, URLs, and snippets for the top results.
Use this to find information online.
"""
services = get_services(_context)
if services is None or services.config is None:
return ToolResult(
content="Web search not available: no configuration.", is_error=True
)
config = services.config
tool_cfg = get_tool_config(_context, "web_search")
api_key = tool_cfg.get("google_api_key") or getattr(config, "google_api_key", "")
cx = tool_cfg.get("google_cx") or getattr(config, "google_cx", "")
if not api_key or not cx:
return ToolResult(
content="Web search not configured. Set google_api_key and google_cx in config.",
is_error=True,
)
num = min(max(1, request.max_results), 10)
try:
params: dict[str, str | int] = {
"key": api_key,
"cx": cx,
"q": request.query,
"num": num,
}
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.get(GOOGLE_SEARCH_URL, params=params)
resp.raise_for_status()
data = resp.json()
items = data.get("items", [])
if not items:
return ToolResult(content=f"No results found for: {request.query}")
lines = []
for i, item in enumerate(items, 1):
title = item.get("title", "No title")
url = item.get("link", "")
snippet = item.get("snippet", "")
lines.append(f"{i}. {title}\n {url}\n {snippet}\n")
return ToolResult(content="\n".join(lines))
except Exception as e:
logger.exception("Web search error")
return ToolResult(content=f"Search error: {e}", is_error=True)
[docs]
def is_available() -> bool:
"""Check if web search is available (always True — uses httpx which is a core dep)."""
return True