SearXNG MCP Server

"""MCP server implementation for SearXNG integration.""" import asyncio import json import logging import os from typing import Any, Optional from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import ( Resource, TextContent, Tool, ) from .client import SearXNGClient # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class SearXNGMCPServer: """MCP Server for SearXNG search engine integration.""" def __init__(self, base_url: str, verify_ssl: bool = True): """Initialize the SearXNG MCP server. Args: base_url: Base URL of the SearXNG instance verify_ssl: Whether to verify SSL certificates """ self.base_url = base_url self.verify_ssl = verify_ssl self.server = Server("searxng-mcp-server") self._client: Optional[SearXNGClient] = None # Register handlers self._register_handlers() def _register_handlers(self) -> None: """Register all MCP handlers.""" @self.server.list_tools() async def list_tools() -> list[Tool]: """List available tools.""" return [ Tool( name="search", description="""Search the web using SearXNG search engine. This tool performs web searches across multiple search engines aggregated by SearXNG. You can filter by categories (general, images, videos, news, etc.), specific engines, language, and time range. Returns comprehensive results including titles, URLs, content snippets, and metadata. Use this when you need to: - Search for information on the web - Find recent news or articles - Search for images or videos - Get diverse results from multiple search engines - Research a topic across different sources""", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The search query string", }, "categories": { "type": "array", "items": {"type": "string"}, "description": "List of categories to search (e.g., ['general', 'images', 'news']). Available: general, images, videos, news, music, files, social media, science, it, map", }, "engines": { "type": "array", "items": {"type": "string"}, "description": "List of specific search engines to use (e.g., ['google', 'bing', 'duckduckgo'])", }, "language": { "type": "string", "description": "Language code for results (default: 'en')", "default": "en", }, "page": { "type": "integer", "description": "Page number for pagination (default: 1)", "default": 1, "minimum": 1, }, "time_range": { "type": "string", "enum": ["day", "month", "year"], "description": "Time range filter for results", }, "safesearch": { "type": "integer", "enum": [0, 1, 2], "description": "Safe search level: 0=off, 1=moderate, 2=strict (default: 0)", "default": 0, }, }, "required": ["query"], }, ), Tool( name="get_suggestions", description="""Get search suggestions/autocomplete for a query prefix. This tool provides search suggestions based on a partial query, similar to autocomplete functionality in search engines. Useful for discovering related searches or expanding on a topic. Use this when you need to: - Get autocomplete suggestions for a search - Discover related search terms - Help users formulate better search queries - Explore variations of a search topic""", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The query prefix to get suggestions for", }, "language": { "type": "string", "description": "Language code for suggestions (default: 'en')", "default": "en", }, }, "required": ["query"], }, ), Tool( name="health_check", description="""Check the health status of the SearXNG instance. This tool verifies that the SearXNG instance is running and accessible. Useful for diagnostics and ensuring the search service is operational before performing searches. Use this when you need to: - Verify the SearXNG instance is accessible - Diagnose connection issues - Check service availability before searching""", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="get_config", description="""Get the configuration of the SearXNG instance. This tool retrieves the SearXNG instance configuration including available search engines, enabled categories, supported locales, plugins, and instance settings. Useful for understanding what capabilities are available. Use this when you need to: - Discover available search engines - See what categories are enabled - Check supported languages/locales - Understand instance capabilities and settings""", inputSchema={ "type": "object", "properties": {}, }, ), ] @self.server.call_tool() async def call_tool(name: str, arguments: Any) -> list[TextContent]: """Handle tool calls.""" try: async with SearXNGClient( self.base_url, verify_ssl=self.verify_ssl ) as client: if name == "search": response = await client.search( query=arguments["query"], categories=arguments.get("categories"), engines=arguments.get("engines"), language=arguments.get("language", "en"), page=arguments.get("page", 1), time_range=arguments.get("time_range"), safesearch=arguments.get("safesearch", 0), ) # Format results for better readability formatted_results = { "query": response.query, "number_of_results": response.number_of_results, "results": [ { "title": r.title, "url": r.url, "content": r.content, "engine": r.engine, "score": r.score, } for r in response.results ], "suggestions": response.suggestions, "answers": response.answers, } return [ TextContent( type="text", text=json.dumps(formatted_results, indent=2), ) ] elif name == "get_suggestions": suggestions = await client.get_suggestions( query=arguments["query"], language=arguments.get("language", "en"), ) return [ TextContent( type="text", text=json.dumps( {"query": arguments["query"], "suggestions": suggestions}, indent=2, ), ) ] elif name == "health_check": health = await client.health_check() return [ TextContent( type="text", text=json.dumps(health, indent=2), ) ] elif name == "get_config": config = await client.get_config() return [ TextContent( type="text", text=json.dumps(config, indent=2), ) ] else: raise ValueError(f"Unknown tool: {name}") except Exception as e: logger.error(f"Error executing tool {name}: {e}") return [ TextContent( type="text", text=json.dumps( {"error": str(e), "tool": name}, indent=2, ), ) ] @self.server.list_resources() async def list_resources() -> list[Resource]: """List available resources.""" return [ Resource( uri="searxng://config", name="SearXNG Instance Configuration", mimeType="application/json", description="Configuration and capabilities of the connected SearXNG instance", ), Resource( uri="searxng://health", name="SearXNG Health Status", mimeType="application/json", description="Health status of the connected SearXNG instance", ), ] @self.server.read_resource() async def read_resource(uri: str) -> str: """Read a resource.""" try: async with SearXNGClient( self.base_url, verify_ssl=self.verify_ssl ) as client: if uri == "searxng://config": config = await client.get_config() return json.dumps(config, indent=2) elif uri == "searxng://health": health = await client.health_check() return json.dumps(health, indent=2) else: raise ValueError(f"Unknown resource: {uri}") except Exception as e: logger.error(f"Error reading resource {uri}: {e}") return json.dumps({"error": str(e), "uri": uri}, indent=2) async def run(self) -> None: """Run the MCP server.""" async with stdio_server() as (read_stream, write_stream): await self.server.run( read_stream, write_stream, self.server.create_initialization_options(), ) def main() -> None: """Main entry point for the server.""" # Get SearXNG base URL from environment variable base_url = os.environ.get("SEARXNG_BASE_URL") if not base_url: logger.error("SEARXNG_BASE_URL environment variable is required") raise ValueError("SEARXNG_BASE_URL environment variable must be set") # Optional: allow disabling SSL verification for self-signed certificates verify_ssl = os.environ.get("SEARXNG_VERIFY_SSL", "true").lower() != "false" logger.info(f"Starting SearXNG MCP Server with base URL: {base_url}") logger.info(f"SSL verification: {'enabled' if verify_ssl else 'disabled'}") # Create and run server server = SearXNGMCPServer(base_url=base_url, verify_ssl=verify_ssl) asyncio.run(server.run()) if __name__ == "__main__": main()