SearXNG MCP Server

"""Search tools for SearXNG MCP."""

import logging
from typing import Dict, Any, List, Optional, Literal
import requests
from mcp.types import TextContent


class SearchTools:
    """Tools for searching via SearXNG."""

    def __init__(self, searxng_url: str, timeout: int = 10):
        """Initialize search tools.

        Args:
            searxng_url: SearXNG instance URL
            timeout: Request timeout in seconds
        """
        self.searxng_url = searxng_url.rstrip("/")
        self.timeout = timeout
        self.logger = logging.getLogger("searxng-mcp.search")

    def _search(
        self,
        query: str,
        category: Optional[str] = None,
        engines: Optional[str] = None,
        language: str = "en",
        page: int = 1
    ) -> Dict[str, Any]:
        """Internal search method.

        Args:
            query: Search query
            category: Search category (general, images, videos, news, etc.)
            engines: Comma-separated list of engines
            language: Search language
            page: Page number

        Returns:
            Search results from SearXNG
        """
        params = {
            "q": query,
            "format": "json",
            "language": language,
            "pageno": page
        }

        if category:
            params["categories"] = category

        if engines:
            params["engines"] = engines

        try:
            self.logger.info(f"Searching: {query} (category: {category})")
            response = requests.get(
                f"{self.searxng_url}/search",
                params=params,
                timeout=self.timeout
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            self.logger.error(f"Search failed: {e}")
            raise RuntimeError(f"Search failed: {e}")

    def _deduplicate_results(self, results: List[Dict]) -> List[Dict]:
        """Remove duplicate results by URL.

        Args:
            results: List of search results

        Returns:
            Deduplicated list
        """
        seen_urls = set()
        unique_results = []

        for result in results:
            url = result.get('url', '')
            if url and url not in seen_urls:
                seen_urls.add(url)
                unique_results.append(result)

        return unique_results

    def search(
        self,
        query: str,
        category: Literal["general", "news"] = "general",
        engines: Optional[str] = None,
        max_results: int = 10
    ) -> List[TextContent]:
        """Quick search for web or news.

        Args:
            query: Search query
            category: "general" for web search, "news" for news
            engines: Comma-separated engine list (e.g., "google,bing,brave")
            max_results: Maximum results to return

        Returns:
            Formatted search results
        """
        results = self._search(query, category=category, engines=engines)

        if category == "news":
            output = f"📰 News Results for: {query}\n\n"
        else:
            output = f"🔍 Search Results for: {query}\n\n"

        for i, result in enumerate(results.get("results", [])[:max_results], 1):
            output += f"{i}. **{result.get('title', 'No title')}**\n"
            output += f"   {result.get('url', '')}\n"
            if result.get('content'):
                content = result['content'][:200] + "..." if len(result['content']) > 200 else result['content']
                output += f"   {content}\n"
            if category == "news" and result.get('publishedDate'):
                output += f"   📅 {result['publishedDate']}\n"
            output += "\n"

        if not results.get("results"):
            output += "No results found.\n"

        return [TextContent(type="text", text=output)]

    def search_media(
        self,
        query: str,
        media_type: Literal["images", "videos"] = "images",
        engines: Optional[str] = None,
        max_results: int = 10
    ) -> List[TextContent]:
        """Search for images or videos.

        Args:
            query: Search query
            media_type: "images" or "videos"
            engines: Comma-separated engine list
            max_results: Maximum results to return

        Returns:
            Formatted media search results
        """
        results = self._search(query, category=media_type, engines=engines)

        if media_type == "images":
            output = f"🖼️ Image Results for: {query}\n\n"
            for i, result in enumerate(results.get("results", [])[:max_results], 1):
                output += f"{i}. **{result.get('title', 'No title')}**\n"
                output += f"   URL: {result.get('img_src', 'N/A')}\n"
                output += f"   Source: {result.get('url', 'N/A')}\n"
                if result.get('thumbnail_src'):
                    output += f"   Thumbnail: {result['thumbnail_src']}\n"
                output += "\n"
        else:  # videos
            output = f"🎥 Video Results for: {query}\n\n"
            for i, result in enumerate(results.get("results", [])[:max_results], 1):
                output += f"{i}. **{result.get('title', 'No title')}**\n"
                output += f"   {result.get('url', '')}\n"
                if result.get('content'):
                    output += f"   {result['content']}\n"
                if result.get('publishedDate'):
                    output += f"   Published: {result['publishedDate']}\n"
                output += "\n"

        if not results.get("results"):
            output += f"No {media_type} found.\n"

        return [TextContent(type="text", text=output)]

    def research_topic(
        self,
        query: str,
        depth: Literal["quick", "standard", "deep"] = "standard"
    ) -> List[TextContent]:
        """Deep research with multiple searches and deduplication.

        Performs multiple searches with different strategies to gather
        comprehensive information from diverse sources. Automatically
        deduplicates results.

        Args:
            query: Research topic
            depth: Research depth
                - quick: 2 searches, ~15 unique results
                - standard: 4 searches, ~30 unique results
                - deep: 6 searches, ~50 unique results

        Returns:
            Deduplicated and aggregated research results
        """
        self.logger.info(f"Starting {depth} research on: {query}")

        all_results = []
        search_strategies = []

        # Define search strategies based on depth
        if depth == "quick":
            search_strategies = [
                {"category": "general", "engines": None},
                {"category": "news", "engines": None},
            ]
            max_per_search = 10
        elif depth == "standard":
            search_strategies = [
                {"category": "general", "engines": "google,bing"},
                {"category": "general", "engines": "duckduckgo,brave"},
                {"category": "news", "engines": None},
                {"category": "general", "engines": "wikipedia"},
            ]
            max_per_search = 10
        else:  # deep
            search_strategies = [
                {"category": "general", "engines": "google,bing"},
                {"category": "general", "engines": "duckduckgo,brave"},
                {"category": "news", "engines": "google,bing"},
                {"category": "news", "engines": "duckduckgo"},
                {"category": "general", "engines": "wikipedia"},
                {"category": "general", "engines": None},  # All engines
            ]
            max_per_search = 15

        # Execute all searches
        for strategy in search_strategies:
            try:
                results = self._search(
                    query,
                    category=strategy["category"],
                    engines=strategy["engines"]
                )
                all_results.extend(results.get("results", [])[:max_per_search])
            except Exception as e:
                self.logger.warning(f"Search strategy failed: {e}")
                continue

        # Deduplicate
        unique_results = self._deduplicate_results(all_results)

        # Format output - present as raw material to analyze, not numbered references
        output = f"🔬 RESEARCH DATA for analysis: {query}\n"
        output += f"📊 {len(unique_results)} unique sources gathered from {len(search_strategies)} search strategies\n\n"
        output += f"{'='*80}\n"
        output += f"RAW SOURCE MATERIAL (analyze and synthesize - do NOT list to user):\n"
        output += f"{'='*80}\n\n"

        for result in unique_results[:25]:
            output += f"• **{result.get('title', 'No title')}**\n"
            output += f"  URL: {result.get('url', '')}\n"
            if result.get('content'):
                content = result['content'][:100] + "..." if len(result['content']) > 100 else result['content']
                output += f"  Content: {content}\n"

            if result.get('publishedDate'):
                output += f"  Date: {result['publishedDate']}\n"

            output += "\n"

        if not unique_results:
            output += "No results found. Try a different query.\n"

        output += f"\n{'='*80}\n"
        output += f"⚠️  YOUR TASK: ANALYZE & SYNTHESIZE (NOT list sources!)\n"
        output += f"{'='*80}\n\n"
        output += f"You have {min(len(unique_results), 25)} sources above as RAW MATERIAL.\n\n"
        output += f"REQUIRED ANALYSIS PROCESS:\n"
        output += f"1. Read all source titles and content snippets above\n"
        output += f"2. Extract key claims and facts from the content\n"
        output += f"3. Cross-reference: What do MULTIPLE sources say? (HIGH confidence)\n"
        output += f"4. What's only in ONE source? (LOW confidence - note as unverified)\n"
        output += f"5. Any contradictions between sources? (flag for user)\n\n"
        output += f"REQUIRED OUTPUT FORMAT:\n"
        output += f"- Executive summary (2-3 sentences)\n"
        output += f"- Key findings with confidence indicators:\n"
        output += f"  ✓ HIGH (5+ sources agree)\n"
        output += f"  ~ MEDIUM (2-4 sources)\n"
        output += f"  ? LOW (single source only)\n"
        output += f"- Contradictions/uncertainties if any\n"
        output += f"- Brief conclusion\n\n"
        output += f"DO NOT output source URLs or numbered lists - synthesize into narrative!\n"
        output += f"{'='*80}\n"

        return [TextContent(type="text", text=output)]