main.py

import asyncio
import json
import random
import uuid

import aiohttp
from bs4 import BeautifulSoup
from readability import Document

from astrbot.api import AstrBotConfig, llm_tool, logger, sp, star
from astrbot.api.event import AstrMessageEvent, filter
from astrbot.api.provider import ProviderRequest
from astrbot.core.provider.func_tool_manager import FunctionToolManager

from .engines import HEADERS, USER_AGENTS, SearchResult
from .engines.bing import Bing
from .engines.sogo import Sogo


class Main(star.Star):
    TOOLS = [
        "web_search",
        "fetch_url",
        "web_search_tavily",
        "tavily_extract_web_page",
        "web_search_bocha",
        "web_search_exa",
        "exa_extract_web_page",
        "exa_find_similar",
    ]

    def __init__(self, context: star.Context) -> None:
        self.context = context
        self.tavily_key_index = 0
        self.tavily_key_lock = asyncio.Lock()

        self.bocha_key_index = 0
        self.bocha_key_lock = asyncio.Lock()

        self.exa_key_index = 0
        self.exa_key_lock = asyncio.Lock()

        # 将 str 类型的 key 迁移至 list[str]，并保存
        cfg = self.context.get_config()
        provider_settings = cfg.get("provider_settings")
        if provider_settings:
            tavily_key = provider_settings.get("websearch_tavily_key")
            if isinstance(tavily_key, str):
                logger.info(
                    "检测到旧版 websearch_tavily_key (字符串格式)，自动迁移为列表格式并保存。",
                )
                if tavily_key:
                    provider_settings["websearch_tavily_key"] = [tavily_key]
                else:
                    provider_settings["websearch_tavily_key"] = []
                cfg.save_config()

            bocha_key = provider_settings.get("websearch_bocha_key")
            if isinstance(bocha_key, str):
                if bocha_key:
                    provider_settings["websearch_bocha_key"] = [bocha_key]
                else:
                    provider_settings["websearch_bocha_key"] = []
                cfg.save_config()

            exa_key = provider_settings.get("websearch_exa_key")
            if isinstance(exa_key, str):
                if exa_key:
                    provider_settings["websearch_exa_key"] = [exa_key]
                else:
                    provider_settings["websearch_exa_key"] = []
                cfg.save_config()

        self.bing_search = Bing()
        self.sogo_search = Sogo()
        self.baidu_initialized = False

    async def _tidy_text(self, text: str) -> str:
        """清理文本，去除空格、换行符等"""
        return text.strip().replace("\n", " ").replace("\r", " ").replace("  ", " ")

    async def _get_from_url(self, url: str, timeout: int = 30) -> str:
        """获取网页内容"""
        if timeout < 30:
            timeout = 30
        header = HEADERS
        header.update({"User-Agent": random.choice(USER_AGENTS)})
        async with aiohttp.ClientSession(trust_env=True) as session:
            async with session.get(
                url, headers=header, timeout=aiohttp.ClientTimeout(total=timeout)
            ) as response:
                html = await response.text(encoding="utf-8")
                doc = Document(html)
                ret = doc.summary(html_partial=True)
                soup = BeautifulSoup(ret, "html.parser")
                ret = await self._tidy_text(soup.get_text())
                return ret

    async def _process_search_result(
        self,
        result: SearchResult,
        idx: int,
        websearch_link: bool,
    ) -> str:
        """处理单个搜索结果"""
        logger.info(f"web_searcher - scraping web: {result.title} - {result.url}")
        try:
            site_result = await self._get_from_url(result.url)
        except BaseException:
            site_result = ""
        site_result = (
            f"{site_result[:700]}..." if len(site_result) > 700 else site_result
        )

        header = f"{idx}. {result.title} "

        if websearch_link and result.url:
            header += result.url

        return f"{header}\n{result.snippet}\n{site_result}\n\n"

    async def _web_search_default(
        self,
        query,
        num_results: int = 5,
    ) -> list[SearchResult]:
        results = []
        try:
            results = await self.bing_search.search(query, num_results)
        except Exception as e:
            logger.error(f"bing search error: {e}, try the next one...")
        if len(results) == 0:
            logger.debug("search bing failed")
            try:
                results = await self.sogo_search.search(query, num_results)
            except Exception as e:
                logger.error(f"sogo search error: {e}")
        if len(results) == 0:
            logger.debug("search sogo failed")
            return []

        return results

    async def _get_tavily_key(self, cfg: AstrBotConfig) -> str:
        """并发安全的从列表中获取并轮换Tavily API密钥。"""
        tavily_keys = cfg.get("provider_settings", {}).get("websearch_tavily_key", [])
        if not tavily_keys:
            raise ValueError("错误：Tavily API密钥未在AstrBot中配置。")

        async with self.tavily_key_lock:
            key = tavily_keys[self.tavily_key_index]
            self.tavily_key_index = (self.tavily_key_index + 1) % len(tavily_keys)
            return key

    async def _web_search_tavily(
        self,
        cfg: AstrBotConfig,
        payload: dict,
        timeout: int = 30,
    ) -> list[SearchResult]:
        """使用 Tavily 搜索引擎进行搜索"""
        tavily_key = await self._get_tavily_key(cfg)
        base_url = (
            cfg.get("provider_settings", {})
            .get("websearch_tavily_base_url", "https://api.tavily.com")
            .rstrip("/")
        )
        if timeout < 30:
            timeout = 30
        url = f"{base_url}/search"
        header = {
            "Authorization": f"Bearer {tavily_key}",
            "Content-Type": "application/json",
        }
        async with aiohttp.ClientSession(trust_env=True) as session:
            async with session.post(
                url,
                json=payload,
                headers=header,
                timeout=aiohttp.ClientTimeout(total=timeout),
            ) as response:
                if response.status != 200:
                    reason = await response.text()
                    raise Exception(
                        f"Tavily web search failed: {reason}, status: {response.status}",
                    )
                data = await response.json()
                results = []
                for item in data.get("results", []):
                    result = SearchResult(
                        title=item.get("title"),
                        url=item.get("url"),
                        snippet=item.get("content"),
                        favicon=item.get("favicon"),
                    )
                    results.append(result)
                return results

    async def _extract_tavily(
        self, cfg: AstrBotConfig, payload: dict, timeout: int = 30
    ) -> list[dict]:
        """使用 Tavily 提取网页内容"""
        tavily_key = await self._get_tavily_key(cfg)
        base_url = (
            cfg.get("provider_settings", {})
            .get("websearch_tavily_base_url", "https://api.tavily.com")
            .rstrip("/")
        )
        if timeout < 30:
            timeout = 30
        url = f"{base_url}/extract"
        header = {
            "Authorization": f"Bearer {tavily_key}",
            "Content-Type": "application/json",
        }
        async with aiohttp.ClientSession(trust_env=True) as session:
            async with session.post(
                url,
                json=payload,
                headers=header,
                timeout=aiohttp.ClientTimeout(total=timeout),
            ) as response:
                if response.status != 200:
                    reason = await response.text()
                    raise Exception(
                        f"Tavily web search failed: {reason}, status: {response.status}",
                    )
                data = await response.json()
                results: list[dict] = data.get("results", [])
                if not results:
                    raise ValueError(
                        "Error: Tavily web searcher does not return any results.",
                    )
                return results

    @llm_tool(name="web_search")
    async def search_from_search_engine(
        self,
        event: AstrMessageEvent,
        query: str,
        max_results: int = 5,
    ) -> str:
        """搜索网络以回答用户的问题。当用户需要搜索网络以获取即时性的信息时调用此工具。

        Args:
            query(string): 和用户的问题最相关的搜索关键词，用于在 Google 上搜索。
            max_results(number): 返回的最大搜索结果数量，默认为 5。

        """
        logger.info(f"web_searcher - search_from_search_engine: {query}")
        cfg = self.context.get_config(umo=event.unified_msg_origin)
        websearch_link = cfg["provider_settings"].get("web_search_link", False)

        results = await self._web_search_default(query, max_results)
        if not results:
            return "Error: web searcher does not return any results."

        tasks = []
        for idx, result in enumerate(results, 1):
            task = self._process_search_result(result, idx, websearch_link)
            tasks.append(task)
        processed_results = await asyncio.gather(*tasks, return_exceptions=True)
        ret = ""
        for processed_result in processed_results:
            if isinstance(processed_result, BaseException):
                logger.error(f"Error processing search result: {processed_result}")
                continue
            ret += processed_result

        if websearch_link:
            ret += "\n\n针对问题，请根据上面的结果分点总结，并且在结尾处附上对应内容的参考链接（如有）。"

        return ret

    async def ensure_baidu_ai_search_mcp(self, umo: str | None = None) -> None:
        if self.baidu_initialized:
            return
        cfg = self.context.get_config(umo=umo)
        key = cfg.get("provider_settings", {}).get(
            "websearch_baidu_app_builder_key",
            "",
        )
        if not key:
            raise ValueError(
                "Error: Baidu AI Search API key is not configured in AstrBot.",
            )
        func_tool_mgr = self.context.get_llm_tool_manager()
        await func_tool_mgr.enable_mcp_server(
            "baidu_ai_search",
            config={
                "transport": "sse",
                "url": f"http://appbuilder.baidu.com/v2/ai_search/mcp/sse?api_key={key}",
                "headers": {},
                "timeout": 600,
            },
        )
        self.baidu_initialized = True
        logger.info("Successfully initialized Baidu AI Search MCP server.")

    @llm_tool(name="fetch_url")
    async def fetch_website_content(
        self, event: AstrMessageEvent, url: str, timeout: int = 30
    ) -> str:
        """Fetch the content of a website with the given web url

        Args:
            url(string): The url of the website to fetch content from
            timeout(number): Optional. Request timeout in seconds. Minimum is 30. Default is 30.

        """
        if timeout < 30:
            timeout = 30
        resp = await self._get_from_url(url, timeout=timeout)
        return resp

    @llm_tool("web_search_tavily")
    async def search_from_tavily(
        self,
        event: AstrMessageEvent,
        query: str,
        max_results: int = 7,
        search_depth: str = "basic",
        topic: str = "general",
        days: int = 3,
        time_range: str = "",
        start_date: str = "",
        end_date: str = "",
        timeout: int = 30,
    ) -> str:
        """A web search tool that uses Tavily to search the web for relevant content.
        Ideal for gathering current information, news, and detailed web content analysis.

        Args:
            query(string): Required. Search query.
            max_results(number): Optional. The maximum number of results to return. Default is 7. Range is 5-20.
            search_depth(string): Optional. The depth of the search, must be one of 'basic', 'advanced'. Default is "basic".
            topic(string): Optional. The topic of the search, must be one of 'general', 'news'. Default is "general".
            days(number): Optional. The number of days back from the current date to include in the search results. Please note that this feature is only available when using the 'news' search topic.
            time_range(string): Optional. The time range back from the current date to include in the search results. This feature is available for both 'general' and 'news' search topics. Must be one of 'day', 'week', 'month', 'year'.
            start_date(string): Optional. The start date for the search results in the format 'YYYY-MM-DD'.
            end_date(string): Optional. The end date for the search results in the format 'YYYY-MM-DD'.
            timeout(number): Optional. Request timeout in seconds. Minimum is 30. Default is 30.

        """
        if timeout < 30:
            timeout = 30
        logger.info(f"web_searcher - search_from_tavily: {query}")
        cfg = self.context.get_config(umo=event.unified_msg_origin)
        # websearch_link = cfg["provider_settings"].get("web_search_link", False)
        if not cfg.get("provider_settings", {}).get("websearch_tavily_key", []):
            raise ValueError("Error: Tavily API key is not configured in AstrBot.")

        # build payload
        payload = {"query": query, "max_results": max_results, "include_favicon": True}
        if search_depth not in ["basic", "advanced"]:
            search_depth = "basic"
        payload["search_depth"] = search_depth

        if topic not in ["general", "news"]:
            topic = "general"
        payload["topic"] = topic

        if topic == "news":
            payload["days"] = days

        if time_range in ["day", "week", "month", "year"]:
            payload["time_range"] = time_range
        if start_date:
            payload["start_date"] = start_date
        if end_date:
            payload["end_date"] = end_date

        results = await self._web_search_tavily(cfg, payload, timeout=timeout)
        if not results:
            return "Error: Tavily web searcher does not return any results."

        ret_ls = []
        ref_uuid = str(uuid.uuid4())[:4]
        for idx, result in enumerate(results, 1):
            index = f"{ref_uuid}.{idx}"
            ret_ls.append(
                {
                    "title": f"{result.title}",
                    "url": f"{result.url}",
                    "snippet": f"{result.snippet}",
                    # TODO: do not need ref for non-webchat platform adapter
                    "index": index,
                }
            )
            if result.favicon:
                sp.temporary_cache["_ws_favicon"][result.url] = result.favicon
        # ret = "\n".join(ret_ls)
        ret = json.dumps({"results": ret_ls}, ensure_ascii=False)
        return ret

    @llm_tool("tavily_extract_web_page")
    async def tavily_extract_web_page(
        self,
        event: AstrMessageEvent,
        url: str = "",
        extract_depth: str = "basic",
        timeout: int = 30,
    ) -> str:
        """Extract the content of a web page using Tavily.

        Args:
            url(string): Required. An URl to extract content from.
            extract_depth(string): Optional. The depth of the extraction, must be one of 'basic', 'advanced'. Default is "basic".
            timeout(number): Optional. Request timeout in seconds. Minimum is 30. Default is 30.

        """
        if timeout < 30:
            timeout = 30
        cfg = self.context.get_config(umo=event.unified_msg_origin)
        if not cfg.get("provider_settings", {}).get("websearch_tavily_key", []):
            raise ValueError("Error: Tavily API key is not configured in AstrBot.")

        if not url:
            raise ValueError("Error: url must be a non-empty string.")
        if extract_depth not in ["basic", "advanced"]:
            extract_depth = "basic"
        payload = {
            "urls": [url],
            "extract_depth": extract_depth,
        }
        results = await self._extract_tavily(cfg, payload, timeout=timeout)
        ret_ls = []
        for result in results:
            ret_ls.append(f"URL: {result.get('url', 'No URL')}")
            ret_ls.append(f"Content: {result.get('raw_content', 'No content')}")
        ret = "\n".join(ret_ls)
        if not ret:
            return "Error: Tavily web searcher does not return any results."
        return ret

    async def _get_bocha_key(self, cfg: AstrBotConfig) -> str:
        """并发安全的从列表中获取并轮换BoCha API密钥。"""
        bocha_keys = cfg.get("provider_settings", {}).get("websearch_bocha_key", [])
        if not bocha_keys:
            raise ValueError("错误：BoCha API密钥未在AstrBot中配置。")

        async with self.bocha_key_lock:
            key = bocha_keys[self.bocha_key_index]
            self.bocha_key_index = (self.bocha_key_index + 1) % len(bocha_keys)
            return key

    async def _web_search_bocha(
        self,
        cfg: AstrBotConfig,
        payload: dict,
        timeout: int = 30,
    ) -> list[SearchResult]:
        """使用 BoCha 搜索引擎进行搜索"""
        bocha_key = await self._get_bocha_key(cfg)
        if timeout < 30:
            timeout = 30
        url = "https://api.bochaai.com/v1/web-search"
        header = {
            "Authorization": f"Bearer {bocha_key}",
            "Content-Type": "application/json",
        }
        async with aiohttp.ClientSession(trust_env=True) as session:
            async with session.post(
                url,
                json=payload,
                headers=header,
                timeout=aiohttp.ClientTimeout(total=timeout),
            ) as response:
                if response.status != 200:
                    reason = await response.text()
                    raise Exception(
                        f"BoCha web search failed: {reason}, status: {response.status}",
                    )
                data = await response.json()
                data = data["data"]["webPages"]["value"]
                results = []
                for item in data:
                    result = SearchResult(
                        title=item.get("name"),
                        url=item.get("url"),
                        snippet=item.get("snippet"),
                        favicon=item.get("siteIcon"),
                    )
                    results.append(result)
                return results

    @llm_tool("web_search_bocha")
    async def search_from_bocha(
        self,
        event: AstrMessageEvent,
        query: str,
        freshness: str = "noLimit",
        summary: bool = False,
        include: str = "",
        exclude: str = "",
        count: int = 10,
        timeout: int = 30,
    ) -> str:
        """
        A web search tool based on Bocha Search API, used to retrieve web pages
        related to the user's query.

        Args:
            query (string): Required. User's search query.

            freshness (string): Optional. Specifies the time range of the search.
                Supported values:
                - "noLimit": No time limit (default, recommended).
                - "oneDay": Within one day.
                - "oneWeek": Within one week.
                - "oneMonth": Within one month.
                - "oneYear": Within one year.
                - "YYYY-MM-DD..YYYY-MM-DD": Search within a specific date range.
                  Example: "2025-01-01..2025-04-06".
                - "YYYY-MM-DD": Search on a specific date.
                  Example: "2025-04-06".
                It is recommended to use "noLimit", as the search algorithm will
                automatically optimize time relevance. Manually restricting the
                time range may result in no search results.

            summary (boolean): Optional. Whether to include a text summary
                for each search result.
                - True: Include summary.
                - False: Do not include summary (default).

            include (string): Optional. Specifies the domains to include in
                the search. Multiple domains can be separated by "|" or ",".
                A maximum of 100 domains is allowed.
                Examples:
                - "qq.com"
                - "qq.com|m.163.com"

            exclude (string): Optional. Specifies the domains to exclude from
                the search. Multiple domains can be separated by "|" or ",".
                A maximum of 100 domains is allowed.
                Examples:
                - "qq.com"
                - "qq.com|m.163.com"

            count (number): Optional. Number of search results to return.
                - Range: 1–50
                - Default: 10
                The actual number of returned results may be less than the
                specified count.

            timeout(number): Optional. Request timeout in seconds. Minimum is 30. Default is 30.
        """
        if timeout < 30:
            timeout = 30
        logger.info(f"web_searcher - search_from_bocha: {query}")
        cfg = self.context.get_config(umo=event.unified_msg_origin)
        # websearch_link = cfg["provider_settings"].get("web_search_link", False)
        if not cfg.get("provider_settings", {}).get("websearch_bocha_key", []):
            raise ValueError("Error: BoCha API key is not configured in AstrBot.")

        # build payload
        payload = {
            "query": query,
            "count": count,
        }

        # freshness：时间范围
        if freshness:
            payload["freshness"] = freshness

        # 是否返回摘要
        payload["summary"] = summary

        # include：限制搜索域
        if include:
            payload["include"] = include

        # exclude：排除搜索域
        if exclude:
            payload["exclude"] = exclude

        results = await self._web_search_bocha(cfg, payload, timeout=timeout)
        if not results:
            return "Error: BoCha web searcher does not return any results."

        ret_ls = []
        ref_uuid = str(uuid.uuid4())[:4]
        for idx, result in enumerate(results, 1):
            index = f"{ref_uuid}.{idx}"
            ret_ls.append(
                {
                    "title": f"{result.title}",
                    "url": f"{result.url}",
                    "snippet": f"{result.snippet}",
                    "index": index,
                }
            )
            if result.favicon:
                sp.temporary_cache["_ws_favicon"][result.url] = result.favicon
        # ret = "\n".join(ret_ls)
        ret = json.dumps({"results": ret_ls}, ensure_ascii=False)
        return ret

    async def _get_exa_key(self, cfg: AstrBotConfig) -> str:
        """并发安全的从列表中获取并轮换 Exa API 密钥。"""
        exa_keys = cfg.get("provider_settings", {}).get("websearch_exa_key", [])
        if not exa_keys:
            raise ValueError("错误：Exa API 密钥未在 AstrBot 中配置。")

        async with self.exa_key_lock:
            key = exa_keys[self.exa_key_index]
            self.exa_key_index = (self.exa_key_index + 1) % len(exa_keys)
            return key

    async def _web_search_exa(
        self,
        cfg: AstrBotConfig,
        payload: dict,
        timeout: int = 30,
    ) -> list[SearchResult]:
        """使用 Exa 搜索引擎进行搜索"""
        exa_key = await self._get_exa_key(cfg)
        base_url = (
            cfg.get("provider_settings", {})
            .get("websearch_exa_base_url", "https://api.exa.ai")
            .rstrip("/")
        )
        url = f"{base_url}/search"
        header = {
            "x-api-key": exa_key,
            "Content-Type": "application/json",
        }
        async with aiohttp.ClientSession(trust_env=True) as session:
            async with session.post(
                url,
                json=payload,
                headers=header,
                timeout=aiohttp.ClientTimeout(total=timeout),
            ) as response:
                if response.status != 200:
                    reason = await response.text()
                    raise Exception(
                        f"Exa web search failed: {reason}, status: {response.status}",
                    )
                data = await response.json()
                results = []
                for item in data.get("results", []):
                    result = SearchResult(
                        title=item.get("title", ""),
                        url=item.get("url", ""),
                        snippet=(item.get("text") or "")[:500],
                    )
                    results.append(result)
                return results

    @llm_tool("web_search_exa")
    async def search_from_exa(
        self,
        event: AstrMessageEvent,
        query: str,
        max_results: int = 10,
        search_type: str = "auto",
        category: str = "",
        timeout: int = 30,
    ) -> str:
        """A web search tool that uses Exa to search the web for relevant content.
        Ideal for gathering current information with semantic search capabilities.
        Supports vertical search categories: company, people, research paper, news, personal site, financial report.

        Args:
            query(string): Required. Search query.
            max_results(number): Optional. The maximum number of results to return. Default is 10. Range is 1-100.
            search_type(string): Optional. The type of search, must be one of 'auto', 'neural', 'fast', 'instant', 'deep'. Default is "auto".
            category(string): Optional. The category of search. Supported values: 'company'(50M+ company pages), 'people'(1B+ people profiles), 'research paper'(100M+ papers), 'news', 'personal site', 'financial report'. Default is empty (general web search).
            timeout(number): Optional. Request timeout in seconds. Minimum is 30. Default is 30.

        """
        if timeout < 30:
            timeout = 30
        logger.info(f"web_searcher - search_from_exa: {query}")
        cfg = self.context.get_config(umo=event.unified_msg_origin)
        if not cfg.get("provider_settings", {}).get("websearch_exa_key", []):
            raise ValueError("Error: Exa API key is not configured in AstrBot.")

        valid_types = ("auto", "neural", "fast", "instant", "deep")
        if search_type not in valid_types:
            search_type = "auto"

        max_results = max(1, min(max_results, 100))

        payload = {
            "query": query,
            "numResults": max_results,
            "type": search_type,
            "contents": {"text": {"maxCharacters": 500}},
        }

        valid_categories = (
            "company",
            "people",
            "research paper",
            "news",
            "personal site",
            "financial report",
        )
        if category in valid_categories:
            payload["category"] = category

        results = await self._web_search_exa(cfg, payload, timeout=timeout)
        if not results:
            return "Error: Exa web searcher does not return any results."

        ret_ls = []
        ref_uuid = str(uuid.uuid4())[:4]
        for idx, result in enumerate(results, 1):
            index = f"{ref_uuid}.{idx}"
            ret_ls.append(
                {
                    "title": result.title,
                    "url": result.url,
                    "snippet": result.snippet,
                    "index": index,
                }
            )
        ret = json.dumps({"results": ret_ls}, ensure_ascii=False)
        return ret

    async def _extract_exa(
        self, cfg: AstrBotConfig, payload: dict, timeout: int = 30
    ) -> list[dict]:
        """使用 Exa 提取网页内容"""
        exa_key = await self._get_exa_key(cfg)
        base_url = (
            cfg.get("provider_settings", {})
            .get("websearch_exa_base_url", "https://api.exa.ai")
            .rstrip("/")
        )
        if timeout < 30:
            timeout = 30
        url = f"{base_url}/contents"
        header = {
            "x-api-key": exa_key,
            "Content-Type": "application/json",
        }
        async with aiohttp.ClientSession(trust_env=True) as session:
            async with session.post(
                url,
                json=payload,
                headers=header,
                timeout=aiohttp.ClientTimeout(total=timeout),
            ) as response:
                if response.status != 200:
                    reason = await response.text()
                    raise Exception(
                        f"Exa content extraction failed: {reason}, status: {response.status}",
                    )
                data = await response.json()
                results: list[dict] = data.get("results", [])
                if not results:
                    raise ValueError(
                        "Error: Exa content extraction does not return any results.",
                    )
                return results

    @llm_tool("exa_extract_web_page")
    async def exa_extract_web_page(
        self,
        event: AstrMessageEvent,
        url: str = "",
        timeout: int = 30,
    ) -> str:
        """Extract the content of a web page using Exa.
        Use this tool when the user wants to extract or summarize content from a specific URL.

        Args:
            url(string): Required. A URL to extract content from.
            timeout(number): Optional. Request timeout in seconds. Minimum is 30. Default is 30.

        """
        if timeout < 30:
            timeout = 30
        cfg = self.context.get_config(umo=event.unified_msg_origin)
        if not cfg.get("provider_settings", {}).get("websearch_exa_key", []):
            raise ValueError("Error: Exa API key is not configured in AstrBot.")

        if not url:
            raise ValueError("Error: url must be a non-empty string.")

        payload = {
            "urls": [url],
            "text": True,
        }

        results = await self._extract_exa(cfg, payload, timeout=timeout)
        ret_ls = []
        for result in results:
            ret_ls.append(f"URL: {result.get('url', 'No URL')}")
            text = await self._tidy_text(result.get("text", "No content"))
            ret_ls.append(f"Content: {text}")
        ret = "\n".join(ret_ls)
        if not ret:
            return "Error: Exa content extraction does not return any results."
        return ret

    async def _find_similar_exa(
        self, cfg: AstrBotConfig, payload: dict, timeout: int = 30
    ) -> list[SearchResult]:
        """使用 Exa 查找相似链接"""
        exa_key = await self._get_exa_key(cfg)
        base_url = (
            cfg.get("provider_settings", {})
            .get("websearch_exa_base_url", "https://api.exa.ai")
            .rstrip("/")
        )
        if timeout < 30:
            timeout = 30
        url = f"{base_url}/findSimilar"
        header = {
            "x-api-key": exa_key,
            "Content-Type": "application/json",
        }
        async with aiohttp.ClientSession(trust_env=True) as session:
            async with session.post(
                url,
                json=payload,
                headers=header,
                timeout=aiohttp.ClientTimeout(total=timeout),
            ) as response:
                if response.status != 200:
                    reason = await response.text()
                    raise Exception(
                        f"Exa find similar failed: {reason}, status: {response.status}",
                    )
                data = await response.json()
                results = []
                for item in data.get("results", []):
                    result = SearchResult(
                        title=item.get("title", ""),
                        url=item.get("url", ""),
                        snippet=(item.get("text") or "")[:500],
                    )
                    results.append(result)
                return results

    @llm_tool("exa_find_similar")
    async def find_similar_links(
        self,
        event: AstrMessageEvent,
        url: str,
        max_results: int = 10,
        timeout: int = 30,
    ) -> str:
        """Find web pages that are semantically similar to a given URL.
        Use this tool when the user wants to discover content related to a specific webpage they have found interesting.

        Args:
            url(string): Required. The URL of the webpage to find similar content for.
            max_results(number): Optional. The maximum number of similar results to return. Default is 10. Range is 1-100.
            timeout(number): Optional. Request timeout in seconds. Minimum is 30. Default is 30.

        """
        if timeout < 30:
            timeout = 30
        logger.info(f"web_searcher - find_similar_links: {url}")
        cfg = self.context.get_config(umo=event.unified_msg_origin)
        if not cfg.get("provider_settings", {}).get("websearch_exa_key", []):
            raise ValueError("Error: Exa API key is not configured in AstrBot.")

        if not url:
            raise ValueError("Error: url must be a non-empty string.")

        num = max(1, min(max_results, 100))

        payload = {
            "url": url,
            "numResults": num,
            "contents": {"text": {"maxCharacters": 500}},
        }

        results = await self._find_similar_exa(cfg, payload, timeout=timeout)
        if not results:
            return "Error: Exa find similar does not return any results."

        ret_ls = []
        ref_uuid = str(uuid.uuid4())[:4]
        for idx, result in enumerate(results, 1):
            index = f"{ref_uuid}.{idx}"
            ret_ls.append(
                {
                    "title": result.title,
                    "url": result.url,
                    "snippet": result.snippet,
                    "index": index,
                }
            )
        ret = json.dumps({"results": ret_ls}, ensure_ascii=False)
        return ret

    @filter.on_llm_request(priority=-10000)
    async def edit_web_search_tools(
        self,
        event: AstrMessageEvent,
        req: ProviderRequest,
    ) -> None:
        """Get the session conversation for the given event."""
        cfg = self.context.get_config(umo=event.unified_msg_origin)
        prov_settings = cfg.get("provider_settings", {})
        websearch_enable = prov_settings.get("web_search", False)
        provider = prov_settings.get("websearch_provider", "default")

        tool_set = req.func_tool
        if isinstance(tool_set, FunctionToolManager):
            req.func_tool = tool_set.get_full_tool_set()
            tool_set = req.func_tool

        if not tool_set:
            return

        if not websearch_enable:
            # pop tools
            for tool_name in self.TOOLS:
                tool_set.remove_tool(tool_name)
            return

        func_tool_mgr = self.context.get_llm_tool_manager()
        if provider == "default":
            web_search_t = func_tool_mgr.get_func("web_search")
            fetch_url_t = func_tool_mgr.get_func("fetch_url")
            if web_search_t and web_search_t.active:
                tool_set.add_tool(web_search_t)
            if fetch_url_t and fetch_url_t.active:
                tool_set.add_tool(fetch_url_t)
            tool_set.remove_tool("web_search_tavily")
            tool_set.remove_tool("tavily_extract_web_page")
            tool_set.remove_tool("AIsearch")
            tool_set.remove_tool("web_search_bocha")
            tool_set.remove_tool("web_search_exa")
            tool_set.remove_tool("exa_extract_web_page")
            tool_set.remove_tool("exa_find_similar")
        elif provider == "tavily":
            web_search_tavily = func_tool_mgr.get_func("web_search_tavily")
            tavily_extract_web_page = func_tool_mgr.get_func("tavily_extract_web_page")
            if web_search_tavily and web_search_tavily.active:
                tool_set.add_tool(web_search_tavily)
            if tavily_extract_web_page and tavily_extract_web_page.active:
                tool_set.add_tool(tavily_extract_web_page)
            tool_set.remove_tool("web_search")
            tool_set.remove_tool("fetch_url")
            tool_set.remove_tool("AIsearch")
            tool_set.remove_tool("web_search_bocha")
            tool_set.remove_tool("web_search_exa")
            tool_set.remove_tool("exa_extract_web_page")
            tool_set.remove_tool("exa_find_similar")
        elif provider == "baidu_ai_search":
            try:
                await self.ensure_baidu_ai_search_mcp(event.unified_msg_origin)
                aisearch_tool = func_tool_mgr.get_func("AIsearch")
                if aisearch_tool and aisearch_tool.active:
                    tool_set.add_tool(aisearch_tool)
                tool_set.remove_tool("web_search")
                tool_set.remove_tool("fetch_url")
                tool_set.remove_tool("web_search_tavily")
                tool_set.remove_tool("tavily_extract_web_page")
                tool_set.remove_tool("web_search_bocha")
                tool_set.remove_tool("web_search_exa")
                tool_set.remove_tool("exa_extract_web_page")
                tool_set.remove_tool("exa_find_similar")
            except Exception as e:
                logger.error(f"Cannot Initialize Baidu AI Search MCP Server: {e}")
        elif provider == "bocha":
            web_search_bocha = func_tool_mgr.get_func("web_search_bocha")
            if web_search_bocha and web_search_bocha.active:
                tool_set.add_tool(web_search_bocha)
            tool_set.remove_tool("web_search")
            tool_set.remove_tool("fetch_url")
            tool_set.remove_tool("AIsearch")
            tool_set.remove_tool("web_search_tavily")
            tool_set.remove_tool("tavily_extract_web_page")
            tool_set.remove_tool("web_search_exa")
            tool_set.remove_tool("exa_extract_web_page")
            tool_set.remove_tool("exa_find_similar")
        elif provider == "exa":
            web_search_exa = func_tool_mgr.get_func("web_search_exa")
            exa_extract_web_page = func_tool_mgr.get_func("exa_extract_web_page")
            exa_find_similar = func_tool_mgr.get_func("exa_find_similar")
            if web_search_exa and web_search_exa.active:
                tool_set.add_tool(web_search_exa)
            if exa_extract_web_page and exa_extract_web_page.active:
                tool_set.add_tool(exa_extract_web_page)
            if exa_find_similar and exa_find_similar.active:
                tool_set.add_tool(exa_find_similar)
            tool_set.remove_tool("web_search")
            tool_set.remove_tool("fetch_url")
            tool_set.remove_tool("AIsearch")
            tool_set.remove_tool("web_search_tavily")
            tool_set.remove_tool("tavily_extract_web_page")
            tool_set.remove_tool("web_search_bocha")