backend/mcp_server/web/search.py

from __future__ import annotations

from typing import Mapping

from dotenv import load_dotenv

from backend.api.mcp_clients.web_client import WebClient
from backend.mcp_server.common.tenant import TenantContext
from backend.mcp_server.common.utils import ToolValidationError, tool_handler


load_dotenv()

_web_client = WebClient()


@tool_handler("web.search")
async def web_search(context: TenantContext, payload: Mapping[str, object]) -> dict[str, object]:
    """
    Perform a Google Custom Search query with basic max-results and region controls.
    """

    query = payload.get("query")
    if not isinstance(query, str) or not query.strip():
        raise ToolValidationError("query must be a non-empty string")

    max_results = payload.get("max_results", 5)
    try:
        max_results_value = max(1, min(int(max_results), 10))
    except (TypeError, ValueError):
        raise ToolValidationError("max_results must be an integer between 1 and 10")

    region = str(payload.get("region", "us"))

    metadata = {
        "max_results": max_results_value,
        "region": region,
        "source": "google",
    }

    try:
        results = await _web_client.search(query, max_results=max_results_value, region=region)
    except RuntimeError as exc:
        metadata["error"] = str(exc)
        return {"query": query, "results": [], "metadata": metadata}

    return {
        "query": query,
        "results": [
            {
                "title": item.get("title"),
                "snippet": item.get("snippet"),
                "url": item.get("link"),
            }
            for item in results
        ],
        "metadata": metadata,
    }