SearXNG MCP

mcp_searxng.py

# /// script
# requires-python = ">=3.13"
# dependencies = [
#     "fastmcp",
#     "httpx",
#     "pydantic",
# ]
# ///
## Usage:
#   SEARXNG_INSTANCE_URL=https://path.to.your.searxng uv run https://gist.github.com/nonsleepr/4e504cddca5e158c4682217346ad460f/raw/64674a60383087dc36ad19f658ccb8ff486727bd/mcp_searxng.py
import json
from os import getenv
import httpx # FastMCP uses httpx internally, good to use it here for consistency
from fastmcp import FastMCP, Context
from typing import Annotated
from pydantic import Field

# Initialize FastMCP server
# You can give it a name that will be exposed to clients
mcp = FastMCP(name="SearXNG Search Server")

# Note: Public instances may block automated requests or have rate limits.
# A self-hosted instance is recommended for reliable use.
SEARXNG_INSTANCE_URL = getenv("SEARXNG_INSTANCE_URL")

@mcp.tool(
    name="searxng_search",
    description="Performs a search using a configured SearXNG instance and returns the results."
)
async def searxng_search(
    query: Annotated[str, Field(description="The search query string.")]
) -> str:
    """
    Searches SearXNG for the given query.
    """
    search_url = f"{SEARXNG_INSTANCE_URL.rstrip('/')}/search"
    params = {
        "q": query,
        "format": "json"
    }

    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(search_url, params=params, timeout=30)
            response.raise_for_status() # Raise an exception for bad status codes
            searxng_data = response.json()

        # Assume SearXNG JSON structure has a "results" key with a list of items
        # Each item is assumed to have "title", "url", and "content" keys.
        results = searxng_data.get("results", [])

        # Format the search results into a single string for the LLM
        formatted_results = []
        for item in results:
            title = item.get("title", "No Title")
            url = item.get("url", "No URL")
            content = item.get("content", "No Content")

            formatted_results.append(f"Title: {title}\nURL: {url}\nSnippet: {content}\n---")

        return "\n".join(formatted_results)

    except httpx.TimeoutException:
        return "Error: SearXNG API request timed out."
    except httpx.RequestException as e:
        return f"Error fetching data from SearXNG: {e}"
    except json.JSONDecodeError:
        return "Error: Could not decode JSON response from SearXNG."
    except Exception as e:
        # Catch any other unexpected errors
        return f"An unexpected error occurred during search: {e}"

# This part is for running the FastMCP server
if __name__ == "__main__":
    # To run this server, you would typically use `uvicorn` or similar ASGI server.
    # FastMCP provides a .run() method for simple execution, but for production
    # use, an ASGI server is recommended.
    print("Starting FastMCP SearXNG Search Server...")
    print(f"SearXNG Instance URL: {SEARXNG_INSTANCE_URL}")
    mcp.run()