core/engine/src/tools/web_search.py - main - alexefan136/flowstack

core/engine/src/tools/web_search.py
1 717 строк57 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:1976704c6
"""
Web Search Tool — MCP tool для поиска в интернете.

Реализует MCP Tool specification 2024-11-05:
https://modelcontextprotocol.io/specification/2024-11-05/server/tools

Поддерживаемые операции:
- `search` — поиск веб-страниц
- `news` — поиск новостей
- `images` — поиск изображений
- `instant_answer` — быстрые ответы (DuckDuckGo Instant Answer)

Поддерживаемые провайдеры (adapter pattern):
- `duckduckgo` — бесплатный, без API key (через duckduckgo-search библиотеку)
- `brave` — Brave Search API (бесплатный tier: 2000 запросов/мес)
- `tavily` — AI-оптимизированный поиск
- `searxng` — self-hosted meta-search engine
- `serpapi` — Google через SerpAPI

Отличие от browser tool:
- **browser** — работа с конкретными URL (fetch, screenshot, scrape)
- **web_search** — discovery: найти релевантные URL по запросу

Архитектура безопасности:
1. Rate limiting — запросы в секунду
2. Timeouts — ограничение времени ответа
3. Output truncation — ограничение длины результатов
4. Safe search — фильтрация inappropriate контента
5. API key isolation — credentials в config, не в коде
6. Provider fallback — graceful degradation при недоступности провайдера

Примеры:
    search(query="Python async best practices", max_results=5)
    news(query="AI breakthroughs 2025", days=7)
    images(query="cute cats", max_results=10)
"""
from __future__ import annotations

import asyncio
import logging
import os
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, ClassVar
from urllib.parse import urlparse

from src.tools.base import Content, TextContent, Tool, ToolResult


logger = logging.getLogger(__name__)


# ============================================================================
# Exceptions
# ============================================================================


class WebSearchError(Exception):
    """Базовое исключение для web search."""


class ProviderNotAvailableError(WebSearchError):
    """Провайдер поиска недоступен."""


class ProviderAuthError(WebSearchError):
    """Ошибка аутентификации у провайдера."""


class ProviderRateLimitError(WebSearchError):
    """Превышен rate limit провайдера."""


class ProviderTimeoutError(WebSearchError):
    """Превышен таймаут запроса к провайдеру."""


class InvalidQueryError(WebSearchError):
    """Невалидный поисковый запрос."""


# ============================================================================
# Configuration
# ============================================================================


@dataclass
class WebSearchConfig:
    """Конфигурация для web search tool."""

    default_provider: str = "duckduckgo"

    brave_api_key: str | None = None
    tavily_api_key: str | None = None
    serpapi_key: str | None = None
    google_api_key: str | None = None
    google_cx: str | None = None

    searxng_url: str | None = None

    timeout_seconds: float = 15.0
    max_results: int = 20
    requests_per_second: float = 2.0
    max_snippet_length: int = 500
    max_content_length: int = 10_000

    safe_search: str = "moderate"

    default_region: str = "wt-wt"
    default_language: str = "en"

    enable_fallback: bool = True
    fallback_providers: list[str] = field(
        default_factory=lambda: ["duckduckgo", "brave", "searxng"]
    )

    def validate(self) -> list[str]:
        """Валидировать конфигурацию. Чистая функция."""
        errors: list[str] = []

        valid_providers = {
            "duckduckgo",
            "brave",
            "tavily",
            "searxng",
            "serpapi",
        }
        if self.default_provider not in valid_providers:
            errors.append(
                f"Invalid default_provider: {self.default_provider}. "
                f"Valid: {', '.join(sorted(valid_providers))}"
            )

        if self.safe_search not in ("off", "moderate", "strict"):
            errors.append(
                f"Invalid safe_search: {self.safe_search}. "
                "Valid: off, moderate, strict"
            )

        if self.timeout_seconds <= 0:
            errors.append("timeout_seconds must be positive")

        if self.max_results <= 0:
            errors.append("max_results must be positive")

        return errors

    @classmethod
    def from_env(cls) -> WebSearchConfig:
        """Создать конфигурацию из переменных окружения."""
        return cls(
            default_provider=os.getenv("WEB_SEARCH_PROVIDER", "duckduckgo"),
            brave_api_key=os.getenv("BRAVE_API_KEY"),
            tavily_api_key=os.getenv("TAVILY_API_KEY"),
            serpapi_key=os.getenv("SERPAPI_KEY"),
            google_api_key=os.getenv("GOOGLE_API_KEY"),
            google_cx=os.getenv("GOOGLE_CX"),
            searxng_url=os.getenv("SEARXNG_URL"),
            timeout_seconds=float(os.getenv("WEB_SEARCH_TIMEOUT", "15")),
            max_results=int(os.getenv("WEB_SEARCH_MAX_RESULTS", "20")),
            safe_search=os.getenv("WEB_SEARCH_SAFE", "moderate"),
            default_region=os.getenv("WEB_SEARCH_REGION", "wt-wt"),
            default_language=os.getenv("WEB_SEARCH_LANGUAGE", "en"),
        )


# ============================================================================
# Response Models
# ============================================================================


@dataclass
class SearchResult:
    """Один результат веб-поиска."""

    title: str
    url: str
    snippet: str
    source: str = ""
    published_date: datetime | None = None
    language: str | None = None
    favicon: str | None = None
    score: float = 0.0
    extra: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        result: dict[str, Any] = {
            "title": self.title,
            "url": self.url,
            "snippet": self.snippet,
            "source": self.source,
            "score": self.score,
        }
        if self.published_date:
            result["published_date"] = self.published_date.isoformat()
        if self.language:
            result["language"] = self.language
        if self.favicon:
            result["favicon"] = self.favicon
        if self.extra:
            result["extra"] = self.extra
        return result


@dataclass
class NewsResult:
    """Один результат поиска новостей."""

    title: str
    url: str
    snippet: str
    source: str
    author: str | None = None
    published_date: datetime | None = None
    image_url: str | None = None
    category: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        result: dict[str, Any] = {
            "title": self.title,
            "url": self.url,
            "snippet": self.snippet,
            "source": self.source,
        }
        if self.author:
            result["author"] = self.author
        if self.published_date:
            result["published_date"] = self.published_date.isoformat()
        if self.image_url:
            result["image_url"] = self.image_url
        if self.category:
            result["category"] = self.category
        return result


@dataclass
class ImageResult:
    """Один результат поиска изображений."""

    title: str
    url: str
    thumbnail_url: str
    source_url: str
    source: str = ""
    width: int | None = None
    height: int | None = None
    size_bytes: int | None = None

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        result: dict[str, Any] = {
            "title": self.title,
            "url": self.url,
            "thumbnail_url": self.thumbnail_url,
            "source_url": self.source_url,
            "source": self.source,
        }
        if self.width is not None:
            result["width"] = self.width
        if self.height is not None:
            result["height"] = self.height
        if self.size_bytes is not None:
            result["size_bytes"] = self.size_bytes
        return result


@dataclass
class InstantAnswer:
    """Быстрый ответ (DuckDuckGo Instant Answer / Knowledge Graph)."""

    query: str
    abstract: str = ""
    abstract_source: str = ""
    abstract_url: str = ""
    image_url: str | None = None
    related_topics: list[dict[str, str]] = field(default_factory=list)
    definitions: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        result: dict[str, Any] = {
            "query": self.query,
            "abstract": self.abstract,
            "abstract_source": self.abstract_source,
            "abstract_url": self.abstract_url,
            "related_topics": self.related_topics,
            "definitions": self.definitions,
        }
        if self.image_url:
            result["image_url"] = self.image_url
        return result


# ============================================================================
# Helper Functions
# ============================================================================


def _extract_domain(url: str) -> str:
    """Извлечь домен из URL. Чистая функция."""
    try:
        parsed = urlparse(url)
        return parsed.netloc or ""
    except Exception:
        return ""


def _truncate_text(text: str, max_length: int) -> str:
    """Обрезать текст до max_length. Чистая функция."""
    if not text or len(text) <= max_length:
        return text
    truncated = text[:max_length].rsplit(" ", 1)[0]
    return truncated + "..."


def _clean_html(text: str) -> str:
    """Удалить HTML теги и нормализовать whitespace. Чистая функция."""
    clean = re.sub(r"<[^>]+>", " ", text)
    clean = re.sub(r"\s+", " ", clean)
    return clean.strip()


def _parse_datetime_safe(value: Any) -> datetime | None:
    """Безопасно парсить datetime. Чистая функция."""
    if value is None:
        return None
    if isinstance(value, datetime):
        return value

    value_str = str(value).strip()
    if not value_str:
        return None

    try:
        normalized = value_str.replace("Z", "+00:00")
        return datetime.fromisoformat(normalized)
    except (ValueError, TypeError):
        pass

    try:
        ts = float(value_str)
        if ts > 1e12:
            ts = ts / 1000
        return datetime.fromtimestamp(ts)
    except (ValueError, TypeError, OSError, OverflowError):
        pass

    return None


# ============================================================================
# Search Providers (Abstract Base + Implementations)
# ============================================================================


class SearchProvider(ABC):
    """Абстрактный базовый класс для search providers."""

    name: ClassVar[str] = "base"

    def __init__(self, config: WebSearchConfig):
        self.config = config
        self._session: Any = None
        self._rate_limiter = asyncio.Semaphore(int(config.requests_per_second))

    async def _get_session(self) -> Any:
        """Получить или создать aiohttp сессию (lazy init)."""
        if self._session is None:
            import aiohttp  # type: ignore[import-not-found,import-untyped]

            timeout = aiohttp.ClientTimeout(total=self.config.timeout_seconds)
            self._session = aiohttp.ClientSession(timeout=timeout)
        return self._session

    async def close(self) -> None:
        """Закрыть HTTP сессию."""
        if self._session is not None:
            await self._session.close()
            self._session = None

    @abstractmethod
    async def search(
        self,
        query: str,
        max_results: int,
        region: str | None = None,
        language: str | None = None,
        safe_search: str | None = None,
    ) -> list[SearchResult]:
        """Выполнить веб-поиск."""
        ...

    async def search_news(
        self,
        query: str,
        max_results: int,
        region: str | None = None,
        language: str | None = None,
        days: int | None = None,
    ) -> list[NewsResult]:
        """Поиск новостей (опционально)."""
        raise NotImplementedError(
            f"Provider {self.name} does not support news search"
        )

    async def search_images(
        self,
        query: str,
        max_results: int,
        region: str | None = None,
        safe_search: str | None = None,
    ) -> list[ImageResult]:
        """Поиск изображений (опционально)."""
        raise NotImplementedError(
            f"Provider {self.name} does not support image search"
        )

    async def instant_answer(self, query: str) -> InstantAnswer | None:
        """Быстрый ответ (если поддерживается)."""
        return None

    def is_available(self) -> bool:
        """Проверить доступность провайдера."""
        return True


class DuckDuckGoProvider(SearchProvider):
    """DuckDuckGo search provider."""

    name: ClassVar[str] = "duckduckgo"

    async def search(
        self,
        query: str,
        max_results: int,
        region: str | None = None,
        language: str | None = None,
        safe_search: str | None = None,
    ) -> list[SearchResult]:
        """Поиск через duckduckgo-search библиотеку."""
        effective_region = region or self.config.default_region
        effective_safe = safe_search or self.config.safe_search

        ddg_safe_map = {"off": "off", "moderate": "moderate", "strict": "on"}
        ddg_safe = ddg_safe_map.get(effective_safe, "moderate")

        async with self._rate_limiter:
            try:
                from duckduckgo_search import AsyncDDGS  # type: ignore[import-not-found]

                async with AsyncDDGS() as ddgs:
                    raw_results = [
                        r
                        async for r in ddgs.atext(
                            query,
                            region=effective_region,
                            safesearch=ddg_safe,
                            max_results=max_results,
                        )
                    ]

                results: list[SearchResult] = []
                for r in raw_results:
                    if not isinstance(r, dict):
                        continue
                    title = str(r.get("title", ""))
                    href = str(r.get("href", ""))
                    body = str(r.get("body", ""))

                    if not href:
                        continue

                    results.append(
                        SearchResult(
                            title=title,
                            url=href,
                            snippet=_truncate_text(body, self.config.max_snippet_length),
                            source=_extract_domain(href),
                        )
                    )

                return results

            except ImportError:
                logger.info(
                    "duckduckgo-search not installed, falling back to HTML parsing"
                )
                return await self._search_html(query, max_results, effective_region)

    async def _search_html(
        self, query: str, max_results: int, region: str
    ) -> list[SearchResult]:
        """Fallback: поиск через HTML парсинг DuckDuckGo."""
        session = await self._get_session()

        url = "https://html.duckduckgo.com/html/"
        params = {"q": query, "kl": region}
        headers = {
            "User-Agent": (
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/120.0.0.0 Safari/537.36"
            )
        }

        try:
            async with session.post(url, data=params, headers=headers) as resp:
                if resp.status != 200:
                    raise WebSearchError(
                        f"DuckDuckGo HTML returned status {resp.status}"
                    )
                html = await resp.text()
        except Exception as e:
            raise WebSearchError(f"DuckDuckGo HTML search failed: {e}") from e

        try:
            from bs4 import BeautifulSoup  # type: ignore[import-not-found,import-untyped]
        except ImportError:
            raise ProviderNotAvailableError(
                "DuckDuckGo HTML fallback requires beautifulsoup4. "
                "Install with: pip install beautifulsoup4"
            )

        soup = BeautifulSoup(html, "html.parser")
        results: list[SearchResult] = []

        for result_div in soup.find_all("div", class_="result"):
            if len(results) >= max_results:
                break

            title_tag = result_div.find("a", class_="result__a")
            snippet_tag = result_div.find("a", class_="result__snippet")

            if not title_tag:
                continue

            title = title_tag.get_text(strip=True)
            href = title_tag.get("href", "")
            snippet = snippet_tag.get_text(strip=True) if snippet_tag else ""

            if not href or not title:
                continue

            if "uddg=" in href:
                from urllib.parse import parse_qs, unquote, urlparse as url_parse

                parsed = url_parse(href)
                qs = parse_qs(parsed.query)
                uddg = qs.get("uddg", [None])[0]
                if uddg:
                    href = unquote(uddg)

            results.append(
                SearchResult(
                    title=_clean_html(title),
                    url=href,
                    snippet=_truncate_text(
                        _clean_html(snippet), self.config.max_snippet_length
                    ),
                    source=_extract_domain(href),
                )
            )

        return results

    async def search_news(
        self,
        query: str,
        max_results: int,
        region: str | None = None,
        language: str | None = None,
        days: int | None = None,
    ) -> list[NewsResult]:
        """Поиск новостей через DuckDuckGo."""
        effective_region = region or self.config.default_region

        async with self._rate_limiter:
            try:
                from duckduckgo_search import AsyncDDGS  # type: ignore[import-not-found]

                async with AsyncDDGS() as ddgs:
                    raw_results = [
                        r
                        async for r in ddgs.anews(
                            query,
                            region=effective_region,
                            max_results=max_results,
                        )
                    ]

                results: list[NewsResult] = []
                for r in raw_results:
                    if not isinstance(r, dict):
                        continue

                    title = str(r.get("title", ""))
                    url_val = str(r.get("url", ""))
                    body = str(r.get("body", ""))
                    source = str(r.get("source", ""))
                    date_val = r.get("date")
                    image = r.get("image")

                    if not url_val:
                        continue

                    results.append(
                        NewsResult(
                            title=title,
                            url=url_val,
                            snippet=_truncate_text(body, self.config.max_snippet_length),
                            source=source or _extract_domain(url_val),
                            published_date=_parse_datetime_safe(date_val),
                            image_url=str(image) if image else None,
                        )
                    )

                return results

            except ImportError:
                raise ProviderNotAvailableError(
                    "DuckDuckGo news search requires duckduckgo-search library. "
                    "Install with: pip install duckduckgo-search"
                )

    async def instant_answer(self, query: str) -> InstantAnswer | None:
        """DuckDuckGo Instant Answer API."""
        session = await self._get_session()
        url = "https://api.duckduckgo.com/"
        params = {"q": query, "format": "json", "no_html": 1, "skip_disambig": 1}

        try:
            async with session.get(url, params=params) as resp:
                if resp.status != 200:
                    return None
                data = await resp.json(content_type=None)
        except Exception:
            return None

        if not isinstance(data, dict):
            return None

        abstract = str(data.get("AbstractText", "") or "")
        abstract_source = str(data.get("AbstractSource", "") or "")
        abstract_url = str(data.get("AbstractURL", "") or "")
        image = data.get("Image")

        related: list[dict[str, str]] = []
        for topic in data.get("RelatedTopics", [])[:10]:
            if not isinstance(topic, dict):
                continue
            text = topic.get("Text")
            first_url = topic.get("FirstURL")
            if text and first_url:
                related.append(
                    {"text": str(text)[:200], "url": str(first_url)}
                )

        if not abstract and not related:
            return None

        return InstantAnswer(
            query=query,
            abstract=abstract,
            abstract_source=abstract_source,
            abstract_url=abstract_url,
            image_url=str(image) if image else None,
            related_topics=related,
        )


class BraveProvider(SearchProvider):
    """Brave Search API provider."""

    name: ClassVar[str] = "brave"

    def is_available(self) -> bool:
        return bool(
            self.config.brave_api_key or os.getenv("BRAVE_API_KEY")
        )

    def _get_api_key(self) -> str:
        key = self.config.brave_api_key or os.getenv("BRAVE_API_KEY")
        if not key:
            raise ProviderAuthError("BRAVE_API_KEY not configured")
        return key

    async def search(
        self,
        query: str,
        max_results: int,
        region: str | None = None,
        language: str | None = None,
        safe_search: str | None = None,
    ) -> list[SearchResult]:
        """Поиск через Brave Search API."""
        session = await self._get_session()
        api_key = self._get_api_key()

        effective_region = region or "ALL"
        effective_language = language or self.config.default_language
        effective_safe = safe_search or self.config.safe_search

        url = "https://api.search.brave.com/res/v1/web/search"
        params = {
            "q": query,
            "count": min(max_results, 20),
            "country": effective_region,
            "search_lang": effective_language,
            "safesearch": effective_safe,
        }
        headers = {
            "Accept": "application/json",
            "Accept-Encoding": "gzip",
            "X-Subscription-Token": api_key,
        }

        async with self._rate_limiter:
            try:
                async with session.get(url, params=params, headers=headers) as resp:
                    if resp.status == 401:
                        raise ProviderAuthError("Brave API authentication failed")
                    if resp.status == 429:
                        raise ProviderRateLimitError("Brave API rate limit exceeded")
                    if resp.status != 200:
                        text = await resp.text()
                        raise WebSearchError(
                            f"Brave API error {resp.status}: {text[:200]}"
                        )

                    data = await resp.json(content_type=None)
            except (ProviderAuthError, ProviderRateLimitError):
                raise
            except Exception as e:
                raise WebSearchError(f"Brave search failed: {e}") from e

        if not isinstance(data, dict):
            raise WebSearchError("Invalid Brave API response")

        web_data = data.get("web") or {}
        raw_results = web_data.get("results") or []

        results: list[SearchResult] = []
        for r in raw_results:
            if not isinstance(r, dict):
                continue

            title = str(r.get("title", ""))
            url_val = str(r.get("url", ""))
            description = str(r.get("description", "") or "")
            age = r.get("age")
            language_val = r.get("language")

            if not url_val:
                continue

            results.append(
                SearchResult(
                    title=_clean_html(title),
                    url=url_val,
                    snippet=_truncate_text(
                        _clean_html(description), self.config.max_snippet_length
                    ),
                    source=_extract_domain(url_val),
                    published_date=_parse_datetime_safe(age),
                    language=str(language_val) if language_val else None,
                )
            )

        return results


class TavilyProvider(SearchProvider):
    """Tavily Search API provider."""

    name: ClassVar[str] = "tavily"

    def is_available(self) -> bool:
        return bool(
            self.config.tavily_api_key or os.getenv("TAVILY_API_KEY")
        )

    def _get_api_key(self) -> str:
        key = self.config.tavily_api_key or os.getenv("TAVILY_API_KEY")
        if not key:
            raise ProviderAuthError("TAVILY_API_KEY not configured")
        return key

    async def search(
        self,
        query: str,
        max_results: int,
        region: str | None = None,
        language: str | None = None,
        safe_search: str | None = None,
    ) -> list[SearchResult]:
        """Поиск через Tavily API."""
        session = await self._get_session()
        api_key = self._get_api_key()

        url = "https://api.tavily.com/search"
        payload = {
            "api_key": api_key,
            "query": query,
            "max_results": min(max_results, 20),
            "search_depth": "basic",
            "include_answer": False,
        }
        headers = {"Content-Type": "application/json"}

        async with self._rate_limiter:
            try:
                async with session.post(
                    url, json=payload, headers=headers
                ) as resp:
                    if resp.status == 401:
                        raise ProviderAuthError("Tavily API authentication failed")
                    if resp.status == 429:
                        raise ProviderRateLimitError("Tavily API rate limit exceeded")
                    if resp.status != 200:
                        text = await resp.text()
                        raise WebSearchError(
                            f"Tavily API error {resp.status}: {text[:200]}"
                        )

                    data = await resp.json(content_type=None)
            except (ProviderAuthError, ProviderRateLimitError):
                raise
            except Exception as e:
                raise WebSearchError(f"Tavily search failed: {e}") from e

        if not isinstance(data, dict):
            raise WebSearchError("Invalid Tavily API response")

        raw_results = data.get("results") or []
        results: list[SearchResult] = []

        for r in raw_results:
            if not isinstance(r, dict):
                continue

            title = str(r.get("title", ""))
            url_val = str(r.get("url", ""))
            content = str(r.get("content", "") or "")
            score = r.get("score")

            if not url_val:
                continue

            score_val = 0.0
            if isinstance(score, (int, float)):
                score_val = float(score)

            results.append(
                SearchResult(
                    title=_clean_html(title),
                    url=url_val,
                    snippet=_truncate_text(
                        _clean_html(content), self.config.max_snippet_length
                    ),
                    source=_extract_domain(url_val),
                    score=score_val,
                )
            )

        return results


class SearXNGProvider(SearchProvider):
    """SearXNG self-hosted meta-search engine."""

    name: ClassVar[str] = "searxng"

    def is_available(self) -> bool:
        return bool(self.config.searxng_url or os.getenv("SEARXNG_URL"))

    def _get_url(self) -> str:
        url = self.config.searxng_url or os.getenv("SEARXNG_URL")
        if not url:
            raise ProviderNotAvailableError("SEARXNG_URL not configured")
        return url.rstrip("/")

    async def search(
        self,
        query: str,
        max_results: int,
        region: str | None = None,
        language: str | None = None,
        safe_search: str | None = None,
    ) -> list[SearchResult]:
        """Поиск через SearXNG JSON API."""
        session = await self._get_session()
        base_url = self._get_url()

        effective_language = language or self.config.default_language
        effective_safe = safe_search or self.config.safe_search

        safe_map = {"off": 0, "moderate": 1, "strict": 2}
        safe_val = safe_map.get(effective_safe, 1)

        url = f"{base_url}/search"
        params = {
            "q": query,
            "format": "json",
            "language": effective_language,
            "safesearch": safe_val,
            "pageno": 1,
        }

        async with self._rate_limiter:
            try:
                async with session.get(url, params=params) as resp:
                    if resp.status != 200:
                        text = await resp.text()
                        raise WebSearchError(
                            f"SearXNG error {resp.status}: {text[:200]}"
                        )
                    data = await resp.json(content_type=None)
            except Exception as e:
                raise WebSearchError(f"SearXNG search failed: {e}") from e

        if not isinstance(data, dict):
            raise WebSearchError("Invalid SearXNG response")

        raw_results = data.get("results") or []
        results: list[SearchResult] = []

        for r in raw_results[:max_results]:
            if not isinstance(r, dict):
                continue

            title = str(r.get("title", ""))
            url_val = str(r.get("url", ""))
            content = str(r.get("content", "") or "")
            engine = r.get("engine")
            published = r.get("publishedDate")

            if not url_val:
                continue

            results.append(
                SearchResult(
                    title=_clean_html(title),
                    url=url_val,
                    snippet=_truncate_text(
                        _clean_html(content), self.config.max_snippet_length
                    ),
                    source=_extract_domain(url_val),
                    published_date=_parse_datetime_safe(published),
                    extra={"engine": engine},
                )
            )

        return results


class SerpAPIProvider(SearchProvider):
    """SerpAPI provider (Google search via SerpAPI)."""

    name: ClassVar[str] = "serpapi"

    def is_available(self) -> bool:
        return bool(self.config.serpapi_key or os.getenv("SERPAPI_KEY"))

    def _get_api_key(self) -> str:
        key = self.config.serpapi_key or os.getenv("SERPAPI_KEY")
        if not key:
            raise ProviderAuthError("SERPAPI_KEY not configured")
        return key

    async def search(
        self,
        query: str,
        max_results: int,
        region: str | None = None,
        language: str | None = None,
        safe_search: str | None = None,
    ) -> list[SearchResult]:
        """Поиск через SerpAPI."""
        session = await self._get_session()
        api_key = self._get_api_key()

        effective_language = language or self.config.default_language

        url = "https://serpapi.com/search.json"
        params = {
            "q": query,
            "engine": "google",
            "api_key": api_key,
            "num": min(max_results, 100),
            "hl": effective_language,
        }

        async with self._rate_limiter:
            try:
                async with session.get(url, params=params) as resp:
                    if resp.status == 401:
                        raise ProviderAuthError("SerpAPI authentication failed")
                    if resp.status == 429:
                        raise ProviderRateLimitError("SerpAPI rate limit exceeded")
                    if resp.status != 200:
                        text = await resp.text()
                        raise WebSearchError(
                            f"SerpAPI error {resp.status}: {text[:200]}"
                        )
                    data = await resp.json(content_type=None)
            except (ProviderAuthError, ProviderRateLimitError):
                raise
            except Exception as e:
                raise WebSearchError(f"SerpAPI search failed: {e}") from e

        if not isinstance(data, dict):
            raise WebSearchError("Invalid SerpAPI response")

        organic_results = data.get("organic_results") or []
        results: list[SearchResult] = []

        for r in organic_results[:max_results]:
            if not isinstance(r, dict):
                continue

            title = str(r.get("title", ""))
            link = str(r.get("link", ""))
            snippet = str(r.get("snippet", "") or "")
            date = r.get("date")
            position = r.get("position")

            if not link:
                continue

            score_val = 0.0
            if isinstance(position, int) and position > 0:
                score_val = max(0.0, 1.0 - (position - 1) * 0.05)

            results.append(
                SearchResult(
                    title=_clean_html(title),
                    url=link,
                    snippet=_truncate_text(
                        _clean_html(snippet), self.config.max_snippet_length
                    ),
                    source=_extract_domain(link),
                    published_date=_parse_datetime_safe(date),
                    score=score_val,
                    extra={"position": position},
                )
            )

        return results


# ============================================================================
# Provider Factory
# ============================================================================


def create_provider(
    name: str, config: WebSearchConfig
) -> SearchProvider:
    """Создать search provider по имени."""
    providers: dict[str, type[SearchProvider]] = {
        "duckduckgo": DuckDuckGoProvider,
        "brave": BraveProvider,
        "tavily": TavilyProvider,
        "searxng": SearXNGProvider,
        "serpapi": SerpAPIProvider,
    }

    provider_cls = providers.get(name)
    if provider_cls is None:
        raise ProviderNotAvailableError(
            f"Unknown provider: {name}. "
            f"Available: {', '.join(sorted(providers.keys()))}"
        )

    return provider_cls(config)


# ============================================================================
# Web Search Tool (MCP)
# ============================================================================


class WebSearchTool(Tool):
    """MCP Tool для поиска в интернете."""

    name: ClassVar[str] = "web_search"
    description: ClassVar[str] = (
        "Поиск в интернете. "
        "Поддерживает операции: search (веб-страницы), news (новости), "
        "images (изображения), instant_answer (быстрые ответы). "
        "Провайдеры: duckduckgo (default), brave, tavily, searxng, serpapi. "
        "Пример: {\"operation\": \"search\", \"query\": \"Python async best practices\"}"
    )

    input_schema: ClassVar[dict[str, Any] | None] = {
        "type": "object",
        "properties": {
            "operation": {
                "type": "string",
                "enum": ["search", "news", "images", "instant_answer"],
                "default": "search",
                "description": "Тип поиска",
            },
            "query": {
                "type": "string",
                "description": "Поисковый запрос",
            },
            "max_results": {
                "type": "integer",
                "default": 5,
                "minimum": 1,
                "maximum": 50,
                "description": "Максимальное количество результатов",
            },
            "provider": {
                "type": "string",
                "enum": ["duckduckgo", "brave", "tavily", "searxng", "serpapi"],
                "description": "Провайдер поиска (по умолчанию из конфига)",
            },
            "region": {
                "type": "string",
                "description": "Регион (например, 'US', 'wt-wt' для worldwide)",
            },
            "language": {
                "type": "string",
                "description": "Язык (например, 'en', 'ru', 'de')",
            },
            "safe_search": {
                "type": "string",
                "enum": ["off", "moderate", "strict"],
                "description": "Фильтр безопасности",
            },
            "days": {
                "type": "integer",
                "description": "Для news: только новости за последние N дней",
            },
        },
        "required": ["query"],
        "additionalProperties": False,
    }

    parameters_schema: ClassVar[dict[str, Any] | None] = input_schema

    examples: ClassVar[list[dict[str, Any]]] = [
        {
            "operation": "search",
            "query": "Python async best practices",
            "max_results": 5,
        },
        {
            "operation": "news",
            "query": "AI breakthroughs",
            "days": 7,
        },
        {
            "operation": "instant_answer",
            "query": "Python programming language",
        },
    ]

    tags: ClassVar[list[str]] = ["web", "search", "utility"]
    is_read_only: ClassVar[bool] = True
    requires_confirmation: ClassVar[bool] = False

    def __init__(self, config: WebSearchConfig | None = None):
        self.config = config or WebSearchConfig.from_env()
        self._providers: dict[str, SearchProvider] = {}

    def _get_provider(self, name: str | None = None) -> SearchProvider:
        """Получить или создать provider (lazy init)."""
        provider_name = name or self.config.default_provider
        if provider_name not in self._providers:
            self._providers[provider_name] = create_provider(
                provider_name, self.config
            )
        return self._providers[provider_name]

    async def _try_with_fallback(
        self,
        operation: str,
        primary_provider: str,
        execute_fn: Any,
        **kwargs: Any,
    ) -> tuple[Any, str]:
        """Выполнить операцию с fallback на другие providers."""
        try:
            provider = self._get_provider(primary_provider)
            if not provider.is_available():
                logger.info(
                    f"Provider {primary_provider} not available, trying fallback"
                )
                raise ProviderNotAvailableError(
                    f"Provider {primary_provider} not available"
                )

            if operation == "search":
                result = await provider.search(**kwargs)
            elif operation == "news":
                result = await provider.search_news(**kwargs)
            elif operation == "images":
                result = await provider.search_images(**kwargs)
            elif operation == "instant_answer":
                result = await provider.instant_answer(**kwargs)
            else:
                raise WebSearchError(f"Unknown operation: {operation}")

            return result, primary_provider

        except (ProviderNotAvailableError, ProviderAuthError) as e:
            if not self.config.enable_fallback:
                raise
            logger.info(
                f"Primary provider {primary_provider} failed: {e}. "
                "Trying fallback providers."
            )

        last_error: Exception | None = None
        for fallback_name in self.config.fallback_providers:
            if fallback_name == primary_provider:
                continue

            try:
                provider = self._get_provider(fallback_name)
                if not provider.is_available():
                    continue

                if operation == "search":
                    result = await provider.search(**kwargs)
                elif operation == "news":
                    result = await provider.search_news(**kwargs)
                elif operation == "images":
                    result = await provider.search_images(**kwargs)
                elif operation == "instant_answer":
                    result = await provider.instant_answer(**kwargs)
                else:
                    continue

                logger.info(f"Fallback to {fallback_name} succeeded")
                return result, fallback_name

            except Exception as e:
                last_error = e
                logger.debug(f"Fallback provider {fallback_name} failed: {e}")
                continue

        raise WebSearchError(
            f"All providers failed for operation '{operation}': {last_error}"
        )

    async def execute(self, **kwargs: Any) -> ToolResult:
        """Выполнить поисковую операцию."""
        query_raw = kwargs.get("query", "")
        if not isinstance(query_raw, str) or not query_raw.strip():
            return ToolResult.failure(
                "Parameter 'query' is required and must be non-empty string"
            )
        query = query_raw.strip()

        operation_raw = kwargs.get("operation", "search")
        operation = str(operation_raw).strip().lower()

        valid_ops = {"search", "news", "images", "instant_answer"}
        if operation not in valid_ops:
            return ToolResult.failure(
                f"Unknown operation: {operation}. Valid: {', '.join(sorted(valid_ops))}"
            )

        max_results_raw = kwargs.get("max_results", 5)
        try:
            max_results = int(max_results_raw)
        except (TypeError, ValueError):
            max_results = 5
        max_results = max(1, min(max_results, self.config.max_results))

        provider_name = kwargs.get("provider")
        if provider_name is not None and not isinstance(provider_name, str):
            return ToolResult.failure("provider must be a string")

        region = kwargs.get("region")
        language = kwargs.get("language")
        safe_search = kwargs.get("safe_search")
        days = kwargs.get("days")

        if safe_search is not None and safe_search not in ("off", "moderate", "strict"):
            return ToolResult.failure(
                "safe_search must be 'off', 'moderate', or 'strict'"
            )

        start_time = asyncio.get_event_loop().time()

        try:
            if operation == "instant_answer":
                return await self._handle_instant_answer(
                    query, provider_name, start_time
                )
            elif operation == "search":
                return await self._handle_search(
                    query, max_results, provider_name, region, language, safe_search, start_time
                )
            elif operation == "news":
                return await self._handle_news(
                    query, max_results, provider_name, region, language, days, start_time
                )
            elif operation == "images":
                return await self._handle_images(
                    query, max_results, provider_name, region, safe_search, start_time
                )
            else:
                return ToolResult.failure(f"Unhandled operation: {operation}")

        except ProviderNotAvailableError as e:
            return ToolResult.failure(
                f"Search provider not available: {e}\n\n"
                "Configure API keys or install required libraries:\n"
                "- duckduckgo: `pip install duckduckgo-search` (free, no API key)\n"
                "- brave: Set BRAVE_API_KEY environment variable\n"
                "- tavily: Set TAVILY_API_KEY environment variable\n"
                "- searxng: Set SEARXNG_URL to your instance\n"
                "- serpapi: Set SERPAPI_KEY environment variable",
                metadata={
                    "operation": operation,
                    "query": query,
                    "error_type": "provider_not_available",
                },
            )
        except ProviderAuthError as e:
            return ToolResult.failure(
                f"Search provider authentication failed: {e}",
                metadata={
                    "operation": operation,
                    "query": query,
                    "error_type": "auth_error",
                },
            )
        except ProviderRateLimitError as e:
            return ToolResult.failure(
                f"Search provider rate limit exceeded: {e}. Try again later.",
                metadata={
                    "operation": operation,
                    "query": query,
                    "error_type": "rate_limit",
                },
            )
        except ProviderTimeoutError as e:
            return ToolResult.failure(
                f"Search provider timeout: {e}",
                metadata={
                    "operation": operation,
                    "query": query,
                    "error_type": "timeout",
                },
            )
        except InvalidQueryError as e:
            return ToolResult.failure(
                f"Invalid search query: {e}",
                metadata={
                    "operation": operation,
                    "query": query,
                    "error_type": "invalid_query",
                },
            )
        except WebSearchError as e:
            return ToolResult.failure(
                f"Web search error: {e}",
                metadata={
                    "operation": operation,
                    "query": query,
                },
            )
        except Exception as e:
            logger.exception(f"Unexpected error in web_search tool: {e}")
            return ToolResult.failure(
                f"Unexpected error: {type(e).__name__}: {e}",
                metadata={"operation": operation, "query": query},
            )

    async def _handle_instant_answer(
        self,
        query: str,
        provider_name: str | None,
        start_time: float,
    ) -> ToolResult:
        """Обработать операцию instant_answer."""
        result, used_provider = await self._try_with_fallback(
            "instant_answer",
            provider_name or self.config.default_provider,
            None,
            query=query,
        )
        search_time_ms = (asyncio.get_event_loop().time() - start_time) * 1000

        if result is None:
            return ToolResult.success_result(
                [TextContent(
                    text=f"# Instant Answer\n\n"
                    f"**Query:** {query}\n\n"
                    f"*(No instant answer available — try regular search)*"
                )],
                metadata={
                    "operation": "instant_answer",
                    "query": query,
                    "provider": used_provider,
                    "has_answer": False,
                    "search_time_ms": search_time_ms,
                },
            )

        lines: list[str] = [
            f"# Instant Answer: {query}\n",
            f"**Provider:** {used_provider}\n",
        ]
        if result.abstract:
            lines.append(f"\n## Summary\n{result.abstract}\n")
            if result.abstract_source:
                lines.append(f"*Source: {result.abstract_source}*")
            if result.abstract_url:
                lines.append(f"URL: {result.abstract_url}")

        if result.definitions:
            lines.append("\n## Definitions")
            for d in result.definitions[:5]:
                lines.append(f"- {d}")

        if result.related_topics:
            lines.append("\n## Related Topics")
            for topic in result.related_topics[:10]:
                text = topic.get("text", "")
                url = topic.get("url", "")
                lines.append(f"- [{text}]({url})")

        metadata = {
            "operation": "instant_answer",
            "query": query,
            "provider": used_provider,
            "has_answer": True,
            "search_time_ms": search_time_ms,
            **result.to_dict(),
        }

        return ToolResult.success_result(
            [TextContent(text="\n".join(lines))],
            metadata=metadata,
        )

    async def _handle_search(
        self,
        query: str,
        max_results: int,
        provider_name: str | None,
        region: str | None,
        language: str | None,
        safe_search: str | None,
        start_time: float,
    ) -> ToolResult:
        """Обработать операцию search."""
        search_kwargs: dict[str, Any] = {
            "query": query,
            "max_results": max_results,
        }
        if region:
            search_kwargs["region"] = region
        if language:
            search_kwargs["language"] = language
        if safe_search:
            search_kwargs["safe_search"] = safe_search

        results, used_provider = await self._try_with_fallback(
            "search",
            provider_name or self.config.default_provider,
            None,
            **search_kwargs,
        )

        search_time_ms = (asyncio.get_event_loop().time() - start_time) * 1000

        if not results:
            text = (
                f"# Search Results: {query}\n\n"
                f"**Provider:** {used_provider}\n"
                f"**Time:** {search_time_ms:.0f}ms\n\n"
                f"*(No results found)*"
            )
            return ToolResult.success_result(
                [TextContent(text=text)],
                metadata={
                    "operation": "search",
                    "query": query,
                    "provider": used_provider,
                    "results_count": 0,
                    "search_time_ms": search_time_ms,
                },
            )

        lines = [
            f"# Search Results: {query}\n",
            f"**Provider:** {used_provider} | "
            f"**Results:** {len(results)} | "
            f"**Time:** {search_time_ms:.0f}ms\n",
        ]

        for i, r in enumerate(results, 1):
            lines.append(f"\n## [{i}] {r.title}")
            lines.append(f"**URL:** {r.url}")
            lines.append(f"**Source:** {r.source}")
            if r.published_date:
                lines.append(f"**Date:** {r.published_date.isoformat()}")
            if r.snippet:
                lines.append(f"\n{r.snippet}")
            lines.append("")

        metadata = {
            "operation": "search",
            "query": query,
            "provider": used_provider,
            "results_count": len(results),
            "search_time_ms": search_time_ms,
            "region": region or self.config.default_region,
            "language": language or self.config.default_language,
            "safe_search": safe_search or self.config.safe_search,
            "results": [r.to_dict() for r in results],
        }

        return ToolResult.success_result(
            [TextContent(text="\n".join(lines))],
            metadata=metadata,
        )

    async def _handle_news(
        self,
        query: str,
        max_results: int,
        provider_name: str | None,
        region: str | None,
        language: str | None,
        days: int | None,
        start_time: float,
    ) -> ToolResult:
        """Обработать операцию news."""
        news_kwargs: dict[str, Any] = {
            "query": query,
            "max_results": max_results,
        }
        if region:
            news_kwargs["region"] = region
        if language:
            news_kwargs["language"] = language
        if days is not None:
            news_kwargs["days"] = int(days)

        results, used_provider = await self._try_with_fallback(
            "news",
            provider_name or self.config.default_provider,
            None,
            **news_kwargs,
        )

        search_time_ms = (asyncio.get_event_loop().time() - start_time) * 1000

        if not results:
            text = (
                f"# News Results: {query}\n\n"
                f"**Provider:** {used_provider}\n"
                f"**Time:** {search_time_ms:.0f}ms\n\n"
                f"*(No news found)*"
            )
            return ToolResult.success_result(
                [TextContent(text=text)],
                metadata={
                    "operation": "news",
                    "query": query,
                    "provider": used_provider,
                    "results_count": 0,
                    "search_time_ms": search_time_ms,
                },
            )

        lines = [
            f"# News Results: {query}\n",
            f"**Provider:** {used_provider} | "
            f"**Results:** {len(results)} | "
            f"**Time:** {search_time_ms:.0f}ms\n",
        ]

        for i, r in enumerate(results, 1):
            lines.append(f"\n## [{i}] {r.title}")
            lines.append(f"**Source:** {r.source}")
            lines.append(f"**URL:** {r.url}")
            if r.published_date:
                lines.append(f"**Published:** {r.published_date.isoformat()}")
            if r.author:
                lines.append(f"**Author:** {r.author}")
            if r.snippet:
                lines.append(f"\n{r.snippet}")
            lines.append("")

        metadata = {
            "operation": "news",
            "query": query,
            "provider": used_provider,
            "results_count": len(results),
            "search_time_ms": search_time_ms,
            "results": [r.to_dict() for r in results],
        }

        return ToolResult.success_result(
            [TextContent(text="\n".join(lines))],
            metadata=metadata,
        )

    async def _handle_images(
        self,
        query: str,
        max_results: int,
        provider_name: str | None,
        region: str | None,
        safe_search: str | None,
        start_time: float,
    ) -> ToolResult:
        """Обработать операцию images."""
        image_kwargs: dict[str, Any] = {
            "query": query,
            "max_results": max_results,
        }
        if region:
            image_kwargs["region"] = region
        if safe_search:
            image_kwargs["safe_search"] = safe_search

        results, used_provider = await self._try_with_fallback(
            "images",
            provider_name or self.config.default_provider,
            None,
            **image_kwargs,
        )

        search_time_ms = (asyncio.get_event_loop().time() - start_time) * 1000

        if not results:
            text = (
                f"# Image Results: {query}\n\n"
                f"*(No images found)*"
            )
            return ToolResult.success_result(
                [TextContent(text=text)],
                metadata={
                    "operation": "images",
                    "query": query,
                    "provider": used_provider,
                    "results_count": 0,
                    "search_time_ms": search_time_ms,
                },
            )

        content_items: list[Content] = []

        text_lines = [
            f"# Image Results: {query}\n",
            f"**Provider:** {used_provider} | "
            f"**Results:** {len(results)}\n",
        ]

        for i, r in enumerate(results, 1):
            text_lines.append(f"\n## [{i}] {r.title}")
            text_lines.append(f"**Source:** {r.source}")
            text_lines.append(f"**URL:** {r.url}")
            text_lines.append(f"**Page:** {r.source_url}")
            if r.width and r.height:
                text_lines.append(f"**Size:** {r.width}x{r.height}")
            text_lines.append("")

        content_items.append(TextContent(text="\n".join(text_lines)))

        metadata = {
            "operation": "images",
            "query": query,
            "provider": used_provider,
            "results_count": len(results),
            "search_time_ms": search_time_ms,
            "results": [r.to_dict() for r in results],
        }

        return ToolResult.success_result(
            content_items,
            metadata=metadata,
        )


# ============================================================================
# Exports
# ============================================================================


__all__ = [
    "WebSearchConfig",
    "WebSearchTool",
    "WebSearchError",
    "ProviderNotAvailableError",
    "ProviderAuthError",
    "ProviderRateLimitError",
    "ProviderTimeoutError",
    "InvalidQueryError",
    "SearchResult",
    "NewsResult",
    "ImageResult",
    "InstantAnswer",
    "SearchProvider",
    "DuckDuckGoProvider",
    "BraveProvider",
    "TavilyProvider",
    "SearXNGProvider",
    "SerpAPIProvider",
    "create_provider",
]