Initial commit: DuckDuckGo MCP server

Carved out from the google-mcp monorepo. MCP server exposing
DuckDuckGo search over SearXNG, with the shared search library.

.env.example

@@ -0,0 +1,2 @@
+# SearXNG secret key (generate: python -c "import secrets; print(secrets.token_hex(32))")
+SEARXNG_SECRET=

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Jarian Cottingham
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

conftest.py

@@ -0,0 +1,2 @@
+# Ensures the repo root is on sys.path so `lib` and the server packages
+# are importable in tests without installation.

docker-compose.yml

@@ -0,0 +1,56 @@
+services:
+  searxng:
+    image: searxng/searxng:latest
+    container_name: searxng
+    expose:
+      - "8080"
+    volumes:
+      - ./searxng-settings.yml:/etc/searxng/settings.yml:ro
+    environment:
+      - SEARXNG_BASE_URL=http://localhost:8080/
+      - SEARXNG_SECRET=${SEARXNG_SECRET:-}
+      - UWSGI_WORKERS=4
+      - UWSGI_THREADS=4
+    restart: unless-stopped
+    stop_grace_period: 5s
+    networks:
+      - mcp-internal
+    healthcheck:
+      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8080/healthz')"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 15s
+
+  duckduckgo-search:
+    build:
+      context: .
+      dockerfile: duckduckgo-mcp/Dockerfile
+    container_name: duckduckgo-mcp
+    ports:
+      - "3002:3002"
+    environment:
+      - RATE_LIMIT_SECONDS=3
+      - MCP_PORT=3002
+      - SEARXNG_URL=http://searxng:8080
+      - PYTHONUNBUFFERED=1
+    depends_on:
+      searxng:
+        condition: service_healthy
+    restart: unless-stopped
+    stop_grace_period: 5s
+    networks:
+      - mcp-internal
+      - mcp-external
+    healthcheck:
+      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:3002/health')"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 10s
+
+networks:
+  mcp-internal:
+    internal: true
+  mcp-external:
+    driver: bridge

duckduckgo-mcp/Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.12-slim
+
+ENV DEBIAN_FRONTEND=noninteractive
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY lib/ ./lib/
+COPY duckduckgo-mcp/ ./duckduckgo-mcp/
+
+ENV PYTHONPATH=/app
+ENV RATE_LIMIT_SECONDS=3
+ENV SEARXNG_URL=http://searxng:8080
+ENV MCP_PORT=3002
+
+CMD ["python", "-m", "duckduckgo-mcp.server"]

duckduckgo-mcp/server.py

@@ -0,0 +1,313 @@
+import asyncio
+import logging
+import os
+import signal
+import sys
+from collections import defaultdict
+from contextlib import asynccontextmanager
+from datetime import datetime, timezone
+
+import httpx
+from mcp.server import Server
+from mcp.server.sse import SseServerTransport
+from mcp.types import TextContent, Tool
+from starlette.applications import Starlette
+from starlette.middleware import Middleware
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.routing import Mount, Route
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    stream=sys.stderr,
+)
+logger = logging.getLogger("duckduckgo-mcp")
+
+app = Server("duckduckgo-search")
+PORT = int(os.environ.get("MCP_PORT", "3002"))
+SEARXNG_URL = os.environ.get("SEARXNG_URL", "http://searxng:8080")
+RATE_LIMIT_SECONDS = float(os.environ.get("RATE_LIMIT_SECONDS", "3"))
+
+DEFAULT_RESULTS = 10
+MAX_RESULTS = 20
+MAX_QUERY_LEN = 500
+
+
+def parse_num_results(value) -> int:
+    """Parse a result-count value, clamped to 1..MAX_RESULTS."""
+    try:
+        num = int(value)
+    except (TypeError, ValueError):
+        return DEFAULT_RESULTS
+    return max(1, min(num, MAX_RESULTS))
+
+
+# --- HTTPX singleton (#8) ---
+
+_httpx_client: httpx.AsyncClient | None = None
+
+
+async def get_httpx_client() -> httpx.AsyncClient:
+    global _httpx_client
+    if _httpx_client is None:
+        _httpx_client = httpx.AsyncClient(
+            timeout=httpx.Timeout(15.0),
+            limits=httpx.Limits(max_connections=32, max_keepalive_connections=16),
+        )
+    return _httpx_client
+
+
+async def close_httpx_client():
+    global _httpx_client
+    if _httpx_client is not None:
+        await _httpx_client.aclose()
+        _httpx_client = None
+
+
+@asynccontextmanager
+async def lifespan(app):
+    logger.info("HTTPX client initialized")
+    try:
+        yield
+    finally:
+        await close_httpx_client()
+        logger.info("HTTPX client closed")
+
+
+# --- Rate limiter ---
+
+class RateLimiter:
+    def __init__(self, min_interval: float):
+        self.min_interval = min_interval
+        self._last_request = None
+        self._lock = asyncio.Lock()
+
+    async def acquire(self):
+        async with self._lock:
+            now = asyncio.get_running_loop().time()
+            if self._last_request:
+                elapsed = now - self._last_request
+                wait = self.min_interval - elapsed
+                if wait > 0:
+                    logger.info(f"Rate limit: waiting {wait:.1f}s")
+                    await asyncio.sleep(wait)
+            self._last_request = asyncio.get_running_loop().time()
+
+
+rate_limiter = RateLimiter(RATE_LIMIT_SECONDS)
+
+
+# --- Per-IP rate limiting middleware (#3) ---
+
+class RateLimitMiddleware(BaseHTTPMiddleware):
+    """Per-IP rate limiting for HTTP endpoints."""
+
+    def __init__(self, app, max_requests: int = 30, window_seconds: int = 60):
+        super().__init__(app)
+        self.max_requests = max_requests
+        self.window_seconds = window_seconds
+        self._requests: dict[str, list[float]] = defaultdict(list)
+        self._lock = asyncio.Lock()
+
+    async def dispatch(self, request: Request, call_next):
+        if request.url.path == "/health":
+            return await call_next(request)
+
+        client_ip = request.client.host if request.client else "unknown"
+        now = datetime.now(timezone.utc).timestamp()
+        cutoff = now - self.window_seconds
+
+        async with self._lock:
+            if len(self._requests) > 1000:
+                for ip in [
+                    ip
+                    for ip, ts in self._requests.items()
+                    if not ts or ts[-1] <= cutoff
+                ]:
+                    del self._requests[ip]
+            timestamps = self._requests[client_ip]
+            self._requests[client_ip] = [t for t in timestamps if t > cutoff]
+            if len(self._requests[client_ip]) >= self.max_requests:
+                return JSONResponse(
+                    {"error": "Rate limit exceeded. Try again later."},
+                    status_code=429,
+                )
+            self._requests[client_ip].append(now)
+
+        response = await call_next(request)
+        return response
+
+
+# --- Search logic ---
+
+async def do_search(query: str, num_results: int, engine: str):
+    """Shared search logic."""
+    await rate_limiter.acquire()
+
+    client = await get_httpx_client()
+    resp = await client.get(
+        f"{SEARXNG_URL}/search",
+        params={
+            "q": query,
+            "format": "json",
+            "engines": engine,
+            "categories": "general",
+            "language": "en",
+        },
+    )
+    resp.raise_for_status()
+    data = resp.json()
+
+    results = data.get("results", [])[:num_results]
+    return [
+        {
+            "title": r.get("title", ""),
+            "url": r.get("url", ""),
+            "snippet": r.get("content", "")[:200],
+        }
+        for r in results
+    ]
+
+
+# --- HTTP endpoint ---
+
+async def search_http(request: Request):
+    query = request.query_params.get("q", "").strip()
+    if not query:
+        return JSONResponse({"error": "'q' parameter required"}, status_code=400)
+    if len(query) > MAX_QUERY_LEN:
+        return JSONResponse(
+            {"error": f"'q' must be {MAX_QUERY_LEN} characters or fewer"},
+            status_code=400,
+        )
+
+    num = parse_num_results(request.query_params.get("num"))
+    try:
+        results = await do_search(query, num, "duckduckgo")
+        return JSONResponse({"query": query, "results": results})
+    except Exception:
+        logger.exception("Search failed")
+        return JSONResponse({"error": "Search service unavailable"}, status_code=503)
+
+
+# --- Health check (#6) ---
+
+async def health_check(request: Request):
+    return JSONResponse({"status": "ok", "service": "duckduckgo-mcp"})
+
+
+# --- MCP tool handlers ---
+
+@app.list_tools()
+async def list_tools() -> list[Tool]:
+    return [
+        Tool(
+            name="duckduckgo_search",
+            description="Search DuckDuckGo and return results with titles, URLs, and snippets.",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "query": {"type": "string", "description": "The search query."},
+                    "num_results": {
+                        "type": "integer",
+                        "description": "Maximum number of results (default 10).",
+                        "default": 10,
+                    },
+                },
+                "required": ["query"],
+            },
+        )
+    ]
+
+
+@app.call_tool()
+async def call_tool(name: str, arguments: dict) -> list[TextContent]:
+    if name != "duckduckgo_search":
+        raise ValueError(f"Unknown tool: {name}")
+
+    query = arguments.get("query", "")
+    if not isinstance(query, str) or not query:
+        return [TextContent(type="text", text="Error: 'query' is required.")]
+    if len(query) > MAX_QUERY_LEN:
+        return [
+            TextContent(
+                type="text",
+                text=f"Error: query must be {MAX_QUERY_LEN} characters or fewer.",
+            )
+        ]
+
+    num_results = parse_num_results(arguments.get("num_results"))
+
+    try:
+        results = await do_search(query, num_results, "duckduckgo")
+    except Exception:
+        logger.exception("Search failed")
+        return [TextContent(type="text", text="Search failed: service unavailable")]
+
+    if not results:
+        return [TextContent(type="text", text=f"No results found for: {query}")]
+
+    lines = [f"Search results for: {query}\n"]
+    for i, r in enumerate(results, 1):
+        lines.append(f"{i}. {r['title']}")
+        lines.append(f"   URL: {r['url']}")
+        if r["snippet"]:
+            lines.append(f"   {r['snippet']}")
+        lines.append("")
+
+    return [TextContent(type="text", text="\n".join(lines))]
+
+
+# --- SSE MCP transport ---
+
+sse = SseServerTransport("/messages/")
+
+
+async def handle_sse(request):
+    async with sse.connect_sse(
+        request.scope, request.receive, request._send
+    ) as (read_stream, write_stream):
+        await app.run(
+            read_stream,
+            write_stream,
+            app.create_initialization_options(),
+        )
+    return Response()
+
+
+starlette_app = Starlette(
+    debug=False,
+    lifespan=lifespan,
+    routes=[
+        Route("/health", endpoint=health_check),
+        Route("/search", endpoint=search_http),
+        Route("/sse", endpoint=handle_sse),
+        Mount("/messages/", app=sse.handle_post_message),
+    ],
+    middleware=[Middleware(RateLimitMiddleware)],
+)
+
+
+async def main():
+    logger.info(f"DuckDuckGo MCP server ready on port {PORT}")
+
+    import uvicorn
+
+    config = uvicorn.Config(starlette_app, host="0.0.0.0", port=PORT, log_level="info")
+    server = uvicorn.Server(config)
+
+    loop = asyncio.get_running_loop()
+
+    def handle_signal():
+        server.should_exit = True
+
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        loop.add_signal_handler(sig, handle_signal)
+
+    await server.serve()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

lib/__init__.py

@@ -0,0 +1,25 @@
+"""Shared search utilities.
+
+Playwright-based scrapers are imported lazily so the package (and its
+rate limiter) works without the optional ``playwright`` dependency.
+"""
+
+from lib.rate_limiter import RateLimiter
+
+__all__ = ["PlaywrightManager", "RateLimiter", "GoogleSearch", "DuckDuckGoSearch"]
+
+
+def __getattr__(name):
+    if name == "PlaywrightManager":
+        from lib.playwright_manager import PlaywrightManager
+
+        return PlaywrightManager
+    if name == "GoogleSearch":
+        from lib.google_search import GoogleSearch
+
+        return GoogleSearch
+    if name == "DuckDuckGoSearch":
+        from lib.duckduckgo_search import DuckDuckGoSearch
+
+        return DuckDuckGoSearch
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

lib/duckduckgo_search.py

@@ -0,0 +1,79 @@
+import asyncio
+import logging
+from typing import Any, Dict, List
+
+from lib.playwright_manager import PlaywrightManager
+from lib.rate_limiter import RateLimiter
+
+logger = logging.getLogger(__name__)
+
+DDG_URL = "https://duckduckgo.com/html/"
+
+
+class DuckDuckGoSearch:
+    """
+    Search DuckDuckGo using Playwright headless browser.
+    Falls back to lightweight HTTP mode if available.
+    """
+
+    def __init__(self, min_interval: float = 3.0):
+        self.rate_limiter = RateLimiter(min_interval_seconds=min_interval)
+
+    async def search(
+        self, query: str, num_results: int = 10
+    ) -> List[Dict[str, Any]]:
+        await self.rate_limiter.acquire()
+
+        page = None
+        try:
+            page = await PlaywrightManager.get_page()
+            logger.info(f"Navigating to DuckDuckGo search: {query}")
+
+            await page.goto(DDG_URL, wait_until="domcontentloaded", timeout=30000)
+
+            await page.fill('input[name="q"]', query)
+            await page.keyboard.press("Enter")
+
+            await page.wait_for_selector(
+                "#rweb-results .result", state="attached", timeout=15000
+            )
+            await asyncio.sleep(1)
+
+            results = await self._parse_page(page, num_results)
+            logger.info(
+                f"DuckDuckGo search returned {len(results)} results for: {query}"
+            )
+            return results
+
+        except Exception as e:
+            logger.error(f"DuckDuckGo search failed for '{query}': {e}")
+            raise RuntimeError(f"DuckDuckGo search failed: {e}") from e
+        finally:
+            if page:
+                await PlaywrightManager.close_page(page)
+
+    async def _parse_page(
+        self, page, max_results: int
+    ) -> List[Dict[str, Any]]:
+        results = await page.evaluate(
+            """() => {
+                const items = document.querySelectorAll('#rweb-results .result');
+                const results = [];
+                for (const item of items) {
+                    const a = item.querySelector('.result__a');
+                    const h2 = item.querySelector('.result__title');
+                    const snippetEl = item.querySelector('.result__snippet');
+                    if (!a || !h2) continue;
+
+                    const url = a.href || '';
+                    const title = h2.textContent?.trim() || '';
+                    const snippet = snippetEl?.textContent?.trim() || '';
+
+                    if (title) {
+                        results.push({ title, url, snippet });
+                    }
+                }
+                return results;
+            }"""
+        )
+        return results[:max_results]

lib/google_search.py

@@ -0,0 +1,163 @@
+import asyncio
+import logging
+import re
+import urllib.parse
+from typing import Any, Dict, List
+
+from lib.playwright_manager import PlaywrightManager
+from lib.rate_limiter import RateLimiter
+
+logger = logging.getLogger(__name__)
+
+GOOGLE_URL = "https://www.google.com/search"
+
+
+class GoogleSearch:
+    """
+    Search Google using Playwright headless browser with stealth mode.
+    Respects rate limits and cleans up page resources after each search.
+    """
+
+    def __init__(self, min_interval: float = 5.0):
+        self.rate_limiter = RateLimiter(min_interval_seconds=min_interval)
+
+    async def search(
+        self, query: str, num_results: int = 10, language: str = "en"
+    ) -> List[Dict[str, Any]]:
+        await self.rate_limiter.acquire()
+
+        page = None
+        try:
+            page = await PlaywrightManager.get_page(stealth=True)
+
+            url_params = (
+                f"{GOOGLE_URL}?q={urllib.parse.quote(query)}"
+                f"&num={min(num_results, 20)}&hl={language}"
+            )
+            logger.info(f"Navigating to Google search: {query}")
+
+            await page.goto(url_params, wait_until="domcontentloaded", timeout=30000)
+            await asyncio.sleep(3)
+
+            html = await page.content()
+            results = self._parse_results(html, num_results)
+            logger.info(f"Google search returned {len(results)} results for: {query}")
+            return results
+
+        except Exception as e:
+            logger.error(f"Google search failed for '{query}': {e}")
+            raise RuntimeError(f"Google search failed: {e}") from e
+        finally:
+            if page:
+                await PlaywrightManager.close_page(page)
+
+    def _parse_results(
+        self, html: str, max_results: int
+    ) -> List[Dict[str, Any]]:
+        results = []
+        seen_urls = set()
+
+        # Strategy 1: Parse /url?q= links (Google's redirect URLs)
+        url_pattern = re.findall(
+            r'<a[^>]*href="(/url\?q=([^&"]+)&[^"]*)"[^>]*>(.*?)</a>',
+            html,
+            re.DOTALL,
+        )
+
+        for _, raw_url, anchor_html in url_pattern:
+            if len(results) >= max_results:
+                break
+
+            url = self._clean_url(raw_url)
+            if not url or url in seen_urls or "google.com" in url:
+                continue
+            seen_urls.add(url)
+
+            title = re.sub(r'<[^>]+>', "", anchor_html).strip()
+            if not title or len(title) < 3:
+                continue
+
+            results.append({"title": title, "url": url, "snippet": ""})
+
+        # Strategy 2: If we got results, try to pair them with nearby snippets
+        if results:
+            results = self._add_snippets(html, results)
+        else:
+            # Strategy 3: Parse data-href attributes (modern Google)
+            data_hrefs = re.findall(r'data-href="(.*?)"', html)
+            for raw_url in data_hrefs:
+                if len(results) >= max_results:
+                    break
+                url = self._clean_url(raw_url)
+                if not url or url in seen_urls or "google.com" in url:
+                    continue
+                seen_urls.add(url)
+                results.append({"title": url, "url": url, "snippet": ""})
+
+        # Strategy 4: Parse from <h3> tags and nearby <a> tags
+        if not results:
+            results = self._parse_from_h3(html, max_results)
+
+        return results[:max_results]
+
+    def _clean_url(self, raw_url: str) -> str:
+        """Decode Google redirect URL."""
+        raw_url = urllib.parse.unquote(raw_url)
+        # Remove tracking parameters
+        url = re.sub(r"&[a-z_]+=.*$", "", raw_url)
+        url = re.sub(r"[?#].*$", "", url)
+        if not url.startswith("http"):
+            url = "https://" + url.lstrip("//")
+        return url.strip()
+
+    def _add_snippets(
+        self, html: str, results: List[Dict[str, Any]]
+    ) -> List[Dict[str, Any]]:
+        """Try to find snippets near result URLs."""
+        snippet_blocks = re.findall(
+            r'<span[^>]*>([\s\S]*?)</span>', html
+        )
+        for i, result in enumerate(results):
+            for block in snippet_blocks:
+                clean = re.sub(r'<[^>]+>', "", block).strip()
+                if not 50 < len(clean) < 300:
+                    continue
+                pos = html.find(clean)
+                if pos == -1:
+                    continue
+                window = html[max(0, pos - 500): pos + 500]
+                if result["url"] in window:
+                    result["snippet"] = clean[:200]
+                    break
+        return results
+
+    def _parse_from_h3(
+        self, html: str, max_results: int
+    ) -> List[Dict[str, Any]]:
+        """Parse results from h3 title tags and nearby links."""
+        results = []
+        h3_pattern = re.findall(
+            r'<h3[^>]*>([\s\S]*?)</h3>', html, re.DOTALL
+        )
+
+        for h3_content in h3_pattern:
+            if len(results) >= max_results:
+                break
+
+            # Extract title from h3
+            title = re.sub(r'<[^>]+>', "", h3_content).strip()
+            if not title or len(title) < 3:
+                continue
+
+            # Find the link inside the h3
+            link_match = re.search(r'href="([^"]+)"', h3_content)
+            url = ""
+            if link_match:
+                url = self._clean_url(link_match.group(1))
+
+            if url and "google.com" not in url:
+                results.append({"title": title, "url": url, "snippet": ""})
+            elif title:
+                results.append({"title": title, "url": url, "snippet": ""})
+
+        return results

lib/playwright_manager.py

@@ -0,0 +1,114 @@
+import asyncio
+import logging
+from typing import Optional
+
+from playwright.async_api import Browser, BrowserContext, Page, async_playwright
+
+logger = logging.getLogger(__name__)
+
+
+class PlaywrightManager:
+    """
+    Singleton-style manager for a single Playwright browser instance.
+    Ensures only ONE browser/process runs per container, with proper cleanup.
+    """
+
+    _instance: Optional["PlaywrightManager"] = None
+    _pw_cm = None
+    _pw = None
+    _browser: Optional[Browser] = None
+    _context: Optional[BrowserContext] = None
+    _initialized = False
+    _cleanup_lock = asyncio.Lock()
+
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+
+    @classmethod
+    async def initialize(cls, user_agent: Optional[str] = None, headless: bool = True):
+        if cls._initialized:
+            logger.debug("Playwright already initialized, reusing instance")
+            return cls._instance
+
+        cls._pw_cm = async_playwright()
+        cls._pw = await cls._pw_cm.__aenter__()
+        browser_args = {
+            "headless": headless,
+            "args": [
+                "--no-sandbox",
+                "--disable-setuid-sandbox",
+                "--disable-dev-shm-usage",
+                "--disable-gpu",
+                "--disable-blink-features=AutomationControlled",
+            ],
+        }
+        cls._browser = await cls._pw.chromium.launch(**browser_args)
+
+        ua = user_agent or (
+            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+            "AppleWebKit/537.36 (KHTML, like Gecko) "
+            "Chrome/131.0.0.0 Safari/131.0.0.0"
+        )
+        cls._context = await cls._browser.new_context(
+            user_agent=ua,
+            viewport={"width": 1920, "height": 1080},
+            locale="en-US",
+            timezone_id="America/New_York",
+            extra_http_headers={"Accept-Language": "en-US,en;q=0.9"},
+        )
+        cls._initialized = True
+        logger.info("Playwright browser initialized (single instance)")
+        return cls._instance
+
+    @classmethod
+    async def get_page(cls, stealth: bool = False) -> Page:
+        if not cls._initialized:
+            raise RuntimeError(
+                "PlaywrightManager not initialized. Call initialize() first."
+            )
+        page = await cls._context.new_page()
+        if stealth:
+            try:
+                from playwright_stealth import Stealth
+                stealth = Stealth()
+                await stealth.apply_stealth_async(page)
+                logger.debug("Stealth mode applied to page")
+            except Exception as e:
+                logger.warning(f"Could not apply stealth: {e}")
+        return page
+
+    @classmethod
+    async def close_page(cls, page: Page):
+        try:
+            await page.close()
+        except Exception as e:
+            logger.debug(f"Error closing page: {e}")
+
+    @classmethod
+    async def shutdown(cls):
+        async with cls._cleanup_lock:
+            if not cls._initialized:
+                return
+            logger.info("Shutting down Playwright browser...")
+            try:
+                if cls._context:
+                    await cls._context.close()
+                    cls._context = None
+                if cls._browser:
+                    await cls._browser.close()
+                    cls._browser = None
+                if cls._pw_cm:
+                    await cls._pw_cm.__aexit__(None, None, None)
+                    cls._pw = None
+                    cls._pw_cm = None
+            except Exception as e:
+                logger.error(f"Error during Playwright shutdown: {e}")
+            finally:
+                cls._initialized = False
+                cls._instance = None
+
+    @classmethod
+    def is_initialized(cls) -> bool:
+        return cls._initialized

lib/rate_limiter.py

@@ -0,0 +1,32 @@
+import asyncio
+import logging
+import time
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+class RateLimiter:
+    """
+    Enforces minimum delay between requests to avoid throttling.
+    Thread-safe for async use within a single event loop.
+    """
+
+    def __init__(self, min_interval_seconds: float = 3.0):
+        self.min_interval = min_interval_seconds
+        self._last_request_time: Optional[float] = None
+        self._lock = asyncio.Lock()
+
+    async def acquire(self):
+        async with self._lock:
+            now = time.monotonic()
+            if self._last_request_time is not None:
+                elapsed = now - self._last_request_time
+                wait_time = self.min_interval - elapsed
+                if wait_time > 0:
+                    logger.info(
+                        f"Rate limiter: waiting {wait_time:.1f}s "
+                        f"(min interval: {self.min_interval}s)"
+                    )
+                    await asyncio.sleep(wait_time)
+            self._last_request_time = time.monotonic()

requirements.txt

@@ -0,0 +1,2 @@
+mcp>=1.0.0,<2.0.0
+httpx>=0.27.0

searxng-settings.yml

@@ -0,0 +1,88 @@
+use_default_settings: True
+
+general:
+  instance_name: "SearXNG"
+  debug: false
+  mail_to: admin@example.com
+  contact_url: false
+  donation_url: false
+  enable_metrics: true
+
+search:
+  safe_search: 0
+  autocomplete: "google"
+  default_lang: "en-US"
+  formats:
+    - html
+    - json
+
+server:
+  port: 8080
+  bind_address: "0.0.0.0"
+  secret_key: "${SEARXNG_SECRET}"
+  limiter: true
+  image_proxy: true
+  method: "GET"
+
+engines:
+  - name: google
+    engine: google
+    sorting: relevance
+    use_mobile_ui: false
+    max_pages: 10
+    enabled: true
+    weight: 2
+
+  - name: duckduckgo
+    engine: duckduckgo
+    use_mobile_ui: false
+    enabled: true
+    weight: 1
+
+  - name: brave
+    engine: brave
+    enabled: true
+    weight: 1
+
+  - name: wikipedia
+    engine: wikipedia
+    enabled: true
+    weight: 1
+
+  - name: bing
+    engine: bing
+    enabled: true
+    weight: 1
+
+  - name: startpage
+    engine: startpage
+    enabled: true
+    weight: 1
+
+  - name: qwant
+    engine: qwant
+    enabled: true
+    weight: 1
+
+  - name: mozillian_ahrefs
+    engine: mozillian_ahrefs
+    enabled: true
+
+  - name: library_genesis
+    engine: library_genesis
+    enabled: false
+
+  - name: google news
+    engine: google_news
+    enabled: true
+
+  - name: duckduckgo news
+    engine: duckduckgo_news
+    enabled: true
+
+outgoing:
+  request_timeout: 5.0
+  max_retries: 1
+  user_agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/131.0.0.0"
+  max_connection_limit: 128
+  max_keepalive_connection: 20

test_client.py

@@ -0,0 +1,176 @@
+#!/usr/bin/env python3
+"""
+Test client that connects to an MCP server over stdio and runs a search.
+Usage:
+  python test_client.py google       # tests google-mcp container
+  python test_client.py duckduckgo   # tests duckduckgo-mcp container
+"""
+
+import asyncio
+import subprocess
+import sys
+
+
+async def test_mcp_server(service_name: str):
+    container_name = f"{service_name}-mcp"
+    query = "Python programming language"
+    print(f"\n{'='*60}")
+    print(f"Testing {container_name} with query: '{query}'")
+    print(f"{'='*60}")
+
+    proc = subprocess.Popen(
+        [
+            "docker", "exec", "-i", container_name,
+            "python", "-c",
+            f"""
+import sys
+sys.path.insert(0, '/app')
+import asyncio
+from mcp.server.stdio import stdio_server
+from {service_name.replace('-', '_')}_mcp.server import app
+
+async def test():
+    async with stdio_server() as (read, write):
+        await asyncio.sleep(0.5)
+        init_req = json.loads(await read.__anext__())
+        # Send initialize response
+        init_result = {{
+            'jsonrpc': '2.0', 'id': init_req['id'],
+            'result': {{
+                'protocolVersion': '2024-11-05', 'capabilities': {{}},
+                'serverInfo': {{'name': 'test', 'version': '1.0'}},
+            }},
+        }}
+        await write.send(json.dumps(init_result))
+        # Client sends initialized notification
+        await read.__anext__()
+        # List tools
+        list_req = {{'jsonrpc': '2.0', 'id': 1, 'method': 'tools/list', 'params': {{}}}}
+        await write.send(json.dumps(list_req))
+        list_resp = json.loads(await read.__anext__())
+        print("TOOLS:", json.dumps(list_resp.get('result', {{}}).get('tools', []), indent=2))
+""",
+        ],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        stdin=subprocess.PIPE,
+    )
+
+    try:
+        stdout, stderr = proc.communicate(timeout=30)
+        print("STDOUT:", stdout.decode())
+        if stderr:
+            print("STDERR:", stderr.decode())
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        print("TIMEOUT: Test timed out after 30s")
+        return False
+
+    return proc.returncode == 0
+
+
+async def test_with_curl(service_name: str):
+    """Test using docker exec to run a Python test directly in the container."""
+    container_name = f"{service_name}-mcp"
+
+    test_script = f"""
+import sys, asyncio, os
+sys.path.insert(0, '/app')
+os.environ.setdefault('RATE_LIMIT_SECONDS', '1')
+
+from lib.playwright_manager import PlaywrightManager
+from lib.google_search import GoogleSearch
+from lib.duckduckgo_search import DuckDuckGoSearch
+
+async def test():
+    await PlaywrightManager.initialize()
+    try:
+        query = "Python programming language"
+        if "{service_name}" == "google":
+            engine = GoogleSearch(min_interval=1)
+            results = await engine.search(query, num_results=5)
+        else:
+            engine = DuckDuckGoSearch(min_interval=1)
+            results = await engine.search(query, num_results=5)
+
+        print(f"Query: {{query}}")
+        print(f"Results found: {{len(results)}}")
+        for i, r in enumerate(results, 1):
+            title = r.get('title', 'N/A')
+            url = r.get('url', 'N/A')
+            snippet = r.get('snippet', '')[:100]
+            print(f"  {{i}}. {{title}}")
+            print(f"     URL: {{url}}")
+            if snippet:
+                print(f"     {{snippet}}")
+
+        if results:
+            print("\\nSUCCESS: Got {{len(results)}} results")
+            sys.exit(0)
+        else:
+            print("\\nFAILURE: No results returned")
+            sys.exit(1)
+    except Exception as e:
+        print(f"ERROR: {{e}}")
+        import traceback
+        traceback.print_exc()
+        sys.exit(1)
+    finally:
+        await PlaywrightManager.shutdown()
+
+asyncio.run(test())
+"""
+
+    print(f"\n{'='*60}")
+    print(f"Testing {container_name}")
+    print(f"{'='*60}")
+
+    try:
+        result = subprocess.run(
+            [
+                "docker", "exec", "-i", container_name,
+                "python", "-c", test_script,
+            ],
+            capture_output=True,
+            text=True,
+            timeout=60,
+        )
+        print(result.stdout)
+        if result.stderr:
+            print("STDERR:", result.stderr)
+        return result.returncode == 0
+
+    except subprocess.TimeoutExpired:
+        print("TIMEOUT: Test timed out after 60s")
+        return False
+    except FileNotFoundError:
+        print("ERROR: docker not found")
+        return False
+
+
+async def main():
+    if len(sys.argv) < 2:
+        print("Usage: python test_client.py [google|duckduckgo|both]")
+        sys.exit(1)
+
+    target = sys.argv[1]
+    services = ["google", "duckduckgo"] if target == "both" else [target]
+
+    results = {}
+    for service in services:
+        success = await test_with_curl(service)
+        results[service] = success
+
+    print(f"\n{'='*60}")
+    print("SUMMARY")
+    print(f"{'='*60}")
+    for service, success in results.items():
+        status = "PASS" if success else "FAIL"
+        print(f"  {service}: {status}")
+
+    all_passed = all(results.values())
+    sys.exit(0 if all_passed else 1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())