backend/mcp_server/web/search.py

from __future__ import annotations

from typing import Mapping

from duckduckgo_search import DDGS

from backend.mcp_server.common.tenant import TenantContext
from backend.mcp_server.common.utils import ToolExecutionError, ToolValidationError, tool_handler


@tool_handler("web.search")
async def web_search(context: TenantContext, payload: Mapping[str, object]) -> dict[str, object]:
    """
    Perform a DuckDuckGo web search with an English-results bias.
    """

    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-en"))

    try:
        ddg = DDGS()
        query_string = query
        if "lang:en" not in query_string.lower():
            query_string = f"{query_string} lang:en"

        try:
            results = ddg.text(query_string, max_results=max_results_value, region=region)
        except TypeError:
            results = ddg.text(query_string, max_results=max_results_value)

        formatted = [
            {
                "title": item.get("title"),
                "snippet": item.get("body"),
                "url": item.get("href"),
            }
            for item in results
        ]

        return {
            "query": query,
            "results": formatted,
            "metadata": {"max_results": max_results_value, "region": region},
        }
    except Exception as exc:
        raise ToolExecutionError(f"web search failed: {exc}") from exc