initial commit

SEARCH_SKILL.md

@@ -0,0 +1,57 @@
+# Search API
+
+Simple HTTP search endpoints. Returns JSON.
+
+## Endpoints
+
+| Service | URL |
+|---------|-----|
+| Google | `http://192.168.8.128:3001/search` |
+| DuckDuckGo | `http://192.168.8.128:3002/search` |
+
+## Usage
+
+```
+GET /search?q=your+query&num=5
+```
+
+**Params:**
+- `q` (required) — search query
+- `num` (optional) — max results, default 10, max 20
+
+## Example
+
+```bash
+curl "http://192.168.8.128:3001/search?q=Python+async+best+practices&num=5"
+```
+
+**Response:**
+```json
+{
+  "query": "Python async best practices",
+  "results": [
+    {
+      "title": "Python asyncio documentation",
+      "url": "https://docs.python.org/3/library/asyncio.html",
+      "snippet": "The asyncio library provides infrastructure for writing concurrent code..."
+    },
+    {
+      "title": "Best Practices for asyncio - Real Python",
+      "url": "https://realpython.com/async-io-python/",
+      "snippet": "When working with asyncio, follow these patterns for clean concurrent code..."
+    }
+  ]
+}
+```
+
+## Rate Limits
+
+- Google: 5s between requests
+- DuckDuckGo: 3s between requests
+
+## Tips
+
+- Use Google (`:3001`) for general web search
+- Use DuckDuckGo (`:3002`) for different ranking or privacy-focused results
+- Keep queries concise (2-5 words)
+- If one engine returns poor results, try the other

USAGE.md

@@ -0,0 +1,219 @@
+# MCP Search Services
+
+MCP (Model Context Protocol) servers that expose Google and DuckDuckGo search as tools for LLMs.
+
+## Quick Start
+
+```bash
+cd google-mcp
+docker compose up -d --build
+```
+
+This starts 3 containers:
+
+| Service | Port | Purpose |
+|---------|------|---------|
+| `google-mcp` | 3001 | MCP server with `google_search` tool |
+| `duckduckgo-mcp` | 3002 | MCP server with `duckduckgo_search` tool |
+| `searxng` | 8080 | Internal metasearch engine |
+
+## Architecture
+
+```
+LLM/MCP Client
+    │
+    ├── http://host:3001/sse  ──→ google-mcp  ──→ searxng ──→ Google
+    └── http://host:3002/sse  ──→ duckduckgo-mcp ──→ searxng ──→ DuckDuckGo
+```
+
+SearXNG handles all the actual search work behind the scenes. It aggregates from Google, Brave, Startpage, Wikipedia, and other engines with rotating user agents, so there's no single browser fingerprint to detect.
+
+## Connecting an LLM Client
+
+### Claude Desktop
+
+Add to `claude_desktop_config.json`:
+
+```json
+{
+  "mcpServers": {
+    "google-search": {
+      "command": "npx",
+      "args": ["-y", "@anthropic-ai/mcp-remote", "http://localhost:3001/sse"]
+    },
+    "duckduckgo-search": {
+      "command": "npx",
+      "args": ["-y", "@anthropic-ai/mcp-remote", "http://localhost:3002/sse"]
+    }
+  }
+}
+```
+
+### Cursor / Windsurf
+
+In your IDE's MCP settings, add:
+
+**Google Search**
+- Type: SSE
+- URL: `http://localhost:3001/sse`
+
+**DuckDuckGo Search**
+- Type: SSE
+- URL: `http://localhost:3002/sse`
+
+### VS Code with MCP extension
+
+Install the MCP extension, then add to workspace settings:
+
+```json
+{
+  "mcp": {
+    "servers": {
+      "google-search": {
+        "transport": {
+          "type": "sse",
+          "url": "http://localhost:3001/sse"
+        }
+      },
+      "duckduckgo-search": {
+        "transport": {
+          "type": "sse",
+          "url": "http://localhost:3002/sse"
+        }
+      }
+    }
+  }
+}
+```
+
+### Remote hosting
+
+If your machine has a public IP or domain, replace `localhost` with your host:
+
+```
+http://your-server-ip:3001/sse
+http://your-server-ip:3002/sse
+```
+
+## Available Tools
+
+### `google_search`
+
+Searches via Google engine through SearXNG.
+
+```json
+{
+  "name": "google_search",
+  "arguments": {
+    "query": "Python programming language",
+    "num_results": 5
+  }
+}
+```
+
+### `duckduckgo_search`
+
+Searches via DuckDuckGo engine through SearXNG.
+
+```json
+{
+  "name": "duckduckgo_search",
+  "arguments": {
+    "query": "Python programming language",
+    "num_results": 5
+  }
+}
+```
+
+Both tools return formatted text with numbered results containing title, URL, and snippet.
+
+## Configuration
+
+| Variable | Default | Description |
+|----------|---------|-------------|
+| `RATE_LIMIT_SECONDS` | 5 (Google) / 3 (DDG) | Minimum seconds between search requests |
+| `MCP_PORT` | 3001 / 3002 | Port the MCP server listens on |
+| `SEARXNG_URL` | `http://searxng:8080` | URL of the SearXNG instance |
+
+To change rate limits, edit `docker-compose.yml` and restart.
+
+## Management
+
+```bash
+# Start all services
+docker compose up -d
+
+# Stop all services
+docker compose down
+
+# View logs
+docker compose logs -f google-mcp
+docker compose logs -f duckduckgo-mcp
+
+# Rebuild after code changes
+docker compose up -d --build
+
+# Run integration test
+python3 integration_test.py
+```
+
+## Testing
+
+```bash
+# Unit tests
+PYTHONPATH=. python3 -m pytest tests/ -v
+
+# Full integration test (requires running containers)
+python3 integration_test.py
+```
+
+## Troubleshooting
+
+**No results returned** — SearXNG may be blocked on your network. Check which engines are working:
+```bash
+curl "http://localhost:8080/search?q=test&format=json" | python3 -m json.tool
+```
+
+**Services not starting** — Check for port conflicts:
+```bash
+docker compose ps
+docker compose logs
+```
+
+**Rate limiting too aggressive** — Lower `RATE_LIMIT_SECONDS` in docker-compose.yml.
+
+**Want more search engines?** — Edit `searxng-settings.yml` to enable/disable engines, then restart SearXNG:
+```bash
+docker compose restart searxng
+```
+
+## Project Structure
+
+```
+google-mcp/
+├── docker-compose.yml      # All 3 services
+├── searxng-settings.yml    # SearXNG engine config
+├── google-mcp/
+│   ├── Dockerfile
+│   └── server.py           # Google MCP server
+├── duckduckgo-mcp/
+│   ├── Dockerfile
+│   └── server.py           # DuckDuckGo MCP server
+├── lib/
+│   ├── playwright_manager.py  # (legacy, not used)
+│   ├── rate_limiter.py        # Per-service rate limiter
+│   ├── google_search.py       # (legacy, not used)
+│   └── duckduckgo_search.py   # (legacy, not used)
+├── tests/
+│   ├── test_rate_limiter.py
+│   └── test_search_parsing.py
+├── integration_test.py       # End-to-end MCP test
+└── requirements.txt
+```
+
+## Notes
+
+- **No Playwright in production** — The original design used Playwright headless browsers, but Google actively blocks automated browsers. SearXNG is the working solution: it's a self-hosted metasearch aggregator that handles browser rotation internally.
+- **CloakBrowser / playwright-cli** — If you need custom browser automation beyond search, those tools would pair well with this setup. SearXNG already solves the search detection problem.
+- **Rate limiting** — Each MCP server has its own rate limiter to avoid overwhelming SearXNG or the upstream engines.
+- **SearXNG as single point** — Both MCP servers share one SearXNG instance. This is fine for moderate usage. For heavy load, run multiple SearXNG instances behind a load balancer.

docker-compose.yml

@@ -0,0 +1,48 @@
+services:
+  searxng:
+    image: searxng/searxng:latest
+    container_name: searxng
+    ports:
+      - "8080:8080"
+    volumes:
+      - ./searxng-settings.yml:/etc/searxng/settings.yml:ro
+    environment:
+      - SEARXNG_BASE_URL=http://localhost:8080/
+      - UWSGI_WORKERS=4
+      - UWSGI_THREADS=4
+    restart: unless-stopped
+    stop_grace_period: 5s
+
+  google-search:
+    build:
+      context: .
+      dockerfile: google-mcp/Dockerfile
+    container_name: google-mcp
+    ports:
+      - "3001:3001"
+    environment:
+      - RATE_LIMIT_SECONDS=5
+      - MCP_PORT=3001
+      - SEARXNG_URL=http://searxng:8080
+      - PYTHONUNBUFFERED=1
+    depends_on:
+      - searxng
+    restart: unless-stopped
+    stop_grace_period: 5s
+
+  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
+    restart: unless-stopped
+    stop_grace_period: 5s

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,198 @@
+import asyncio
+import json
+import logging
+import os
+import signal
+import sys
+
+import httpx
+
+from mcp.server import Server
+from mcp.server.sse import SseServerTransport
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.routing import Mount, Route
+from mcp.types import Tool, TextContent
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)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"))
+
+
+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_event_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_event_loop().time()
+
+
+rate_limiter = RateLimiter(RATE_LIMIT_SECONDS)
+
+
+async def do_search(query: str, num_results: int, engine: str):
+    """Shared search logic."""
+    await rate_limiter.acquire()
+
+    async with httpx.AsyncClient(timeout=15) as 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", "")
+    if not query:
+        return JSONResponse({"error": "'q' parameter required"}, status_code=400)
+
+    num = min(int(request.query_params.get("num", "10")), 20)
+    try:
+        results = await do_search(query, num, "duckduckgo")
+        return JSONResponse({"query": query, "results": results})
+    except Exception as e:
+        logger.exception("Search failed")
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
+# --- 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 query:
+        return [TextContent(type="text", text="Error: 'query' is required.")]
+
+    num_results = min(int(arguments.get("num_results", 10)), 20)
+
+    try:
+        results = await do_search(query, num_results, "duckduckgo")
+    except Exception as e:
+        return [TextContent(type="text", text=f"Search failed: {e}")]
+
+    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=True,
+    routes=[
+        Route("/search", endpoint=search_http),
+        Route("/sse", endpoint=handle_sse),
+        Mount("/messages/", app=sse.handle_post_message),
+    ],
+)
+
+
+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())

google-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 google-mcp/ ./google-mcp/
+
+ENV PYTHONPATH=/app
+ENV RATE_LIMIT_SECONDS=5
+ENV SEARXNG_URL=http://searxng:8080
+ENV MCP_PORT=3001
+
+CMD ["python", "-m", "google-mcp.server"]

google-mcp/server.py

@@ -0,0 +1,198 @@
+import asyncio
+import json
+import logging
+import os
+import signal
+import sys
+
+import httpx
+
+from mcp.server import Server
+from mcp.server.sse import SseServerTransport
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.routing import Mount, Route
+from mcp.types import Tool, TextContent
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    stream=sys.stderr,
+)
+logger = logging.getLogger("google-mcp")
+
+app = Server("google-search")
+PORT = int(os.environ.get("MCP_PORT", "3001"))
+SEARXNG_URL = os.environ.get("SEARXNG_URL", "http://searxng:8080")
+RATE_LIMIT_SECONDS = float(os.environ.get("RATE_LIMIT_SECONDS", "5"))
+
+
+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_event_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_event_loop().time()
+
+
+rate_limiter = RateLimiter(RATE_LIMIT_SECONDS)
+
+
+async def do_search(query: str, num_results: int, engine: str):
+    """Shared search logic."""
+    await rate_limiter.acquire()
+
+    async with httpx.AsyncClient(timeout=15) as 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", "")
+    if not query:
+        return JSONResponse({"error": "'q' parameter required"}, status_code=400)
+
+    num = min(int(request.query_params.get("num", "10")), 20)
+    try:
+        results = await do_search(query, num, "google")
+        return JSONResponse({"query": query, "results": results})
+    except Exception as e:
+        logger.exception("Search failed")
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
+# --- MCP tool handlers ---
+
+@app.list_tools()
+async def list_tools() -> list[Tool]:
+    return [
+        Tool(
+            name="google_search",
+            description="Search Google and return top 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, max 20).",
+                        "default": 10,
+                    },
+                },
+                "required": ["query"],
+            },
+        )
+    ]
+
+
+@app.call_tool()
+async def call_tool(name: str, arguments: dict) -> list[TextContent]:
+    if name != "google_search":
+        raise ValueError(f"Unknown tool: {name}")
+
+    query = arguments.get("query", "")
+    if not query:
+        return [TextContent(type="text", text="Error: 'query' is required.")]
+
+    num_results = min(int(arguments.get("num_results", 10)), 20)
+
+    try:
+        results = await do_search(query, num_results, "google")
+    except Exception as e:
+        return [TextContent(type="text", text=f"Search failed: {e}")]
+
+    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=True,
+    routes=[
+        Route("/search", endpoint=search_http),
+        Route("/sse", endpoint=handle_sse),
+        Mount("/messages/", app=sse.handle_post_message),
+    ],
+)
+
+
+async def main():
+    logger.info(f"Google 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())

integration_test.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+"""Integration test: connect to MCP servers via SSE and test search."""
+import asyncio
+import json
+import sys
+import aiohttp
+
+
+class SSEClient:
+    def __init__(self):
+        self.events = []
+        self.lock = asyncio.Lock()
+
+    async def add_event(self, data):
+        async with self.lock:
+            self.events.append(data)
+
+    async def wait_for_id(self, expected_id, timeout=30):
+        start = asyncio.get_event_loop().time()
+        while asyncio.get_event_loop().time() - start < timeout:
+            async with self.lock:
+                for evt in self.events:
+                    if isinstance(evt, dict) and evt.get("id") == expected_id:
+                        return evt
+            await asyncio.sleep(0.1)
+        return None
+
+
+async def test_server(name, host, port, tool_name, query):
+    print(f"\n{'='*60}")
+    print(f"Testing {name} at http://{host}:{port}")
+    print(f"{'='*60}")
+
+    base = f"http://{host}:{port}"
+    collector = SSEClient()
+    messages_endpoint = None
+
+    async with aiohttp.ClientSession() as session:
+        async with session.get(f"{base}/sse") as sse_resp:
+            if sse_resp.status != 200:
+                print(f"FAIL: SSE status {sse_resp.status}")
+                return False
+
+            async def read_sse():
+                try:
+                    async for line in sse_resp.content:
+                        line_str = line.decode().strip()
+                        if not line_str or not line_str.startswith("data:"):
+                            continue
+                        data_str = line_str[5:].strip()
+                        try:
+                            data = json.loads(data_str)
+                            await collector.add_event(data)
+                        except json.JSONDecodeError:
+                            nonlocal messages_endpoint
+                            messages_endpoint = base.rstrip("/") + data_str
+                            print(f"    Endpoint: {messages_endpoint}")
+                            await collector.add_event(data_str)
+                except Exception as e:
+                    print(f"    SSE error: {e}")
+
+            reader = asyncio.create_task(read_sse())
+
+            try:
+                for _ in range(50):
+                    if messages_endpoint:
+                        break
+                    await asyncio.sleep(0.1)
+
+                if not messages_endpoint:
+                    print("FAIL: No endpoint")
+                    return False
+
+                # Initialize
+                async with session.post(messages_endpoint, json={
+                    "jsonrpc": "2.0", "id": 1, "method": "initialize",
+                    "params": {"protocolVersion": "2024-11-05", "capabilities": {},
+                               "clientInfo": {"name": "test", "version": "1.0"}},
+                }) as resp:
+                    pass
+
+                init = await collector.wait_for_id(1, timeout=10)
+                if not init:
+                    print("FAIL: No init response")
+                    return False
+                print(f"    Protocol: {init['result']['protocolVersion']}")
+
+                # Initialized notification
+                async with session.post(messages_endpoint, json={
+                    "jsonrpc": "2.0", "method": "notifications/initialized",
+                }) as resp:
+                    pass
+
+                # List tools
+                async with session.post(messages_endpoint, json={
+                    "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {},
+                }) as resp:
+                    pass
+
+                tools = await collector.wait_for_id(2, timeout=10)
+                if tools:
+                    print(f"    Tools: {[t['name'] for t in tools.get('result', {}).get('tools', [])]}")
+
+                # Call search
+                print(f"    Searching: '{query}'")
+                async with session.post(messages_endpoint, json={
+                    "jsonrpc": "2.0", "id": 3, "method": "tools/call",
+                    "params": {"name": tool_name, "arguments": {"query": query, "num_results": 5}},
+                }) as resp:
+                    pass
+
+                result = await collector.wait_for_id(3, timeout=30)
+                if not result:
+                    print("FAIL: No search response")
+                    return False
+
+                content = result.get("result", {}).get("content", [])
+                if content:
+                    text = content[0].get("text", "")
+                    print(f"\n    --- RESULTS ---\n{text}\n    --- END ---")
+                    if "Search failed" in text:
+                        print("    PARTIAL: Error in response")
+                        return False
+                    if "No results" in text:
+                        print("    PARTIAL: No results")
+                        return False
+                    return True
+
+                print("FAIL: Empty response")
+                return False
+
+            finally:
+                reader.cancel()
+                try:
+                    await reader
+                except asyncio.CancelledError:
+                    pass
+
+
+async def main():
+    host = "localhost"
+    results = {}
+
+    # Wait for services to be ready
+    print("Waiting for services to start...")
+    await asyncio.sleep(5)
+
+    results["google"] = await test_server("Google", host, 3001, "google_search", "Python programming language")
+
+    await asyncio.sleep(2)
+
+    results["duckduckgo"] = await test_server("DuckDuckGo", host, 3002, "duckduckgo_search", "Python programming language")
+
+    print(f"\n{'='*60}")
+    print("SUMMARY")
+    print(f"{'='*60}")
+    for name, ok in results.items():
+        print(f"  {name}: {'PASS' if ok else 'FAIL'}")
+
+    sys.exit(0 if all(results.values()) else 1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

lib/__init__.py

@@ -0,0 +1,11 @@
+from lib.playwright_manager import PlaywrightManager
+from lib.rate_limiter import RateLimiter
+from lib.google_search import GoogleSearch
+from lib.duckduckgo_search import DuckDuckGoSearch
+
+__all__ = [
+    "PlaywrightManager",
+    "RateLimiter",
+    "GoogleSearch",
+    "DuckDuckGoSearch",
+]

lib/duckduckgo_search.py

@@ -0,0 +1,79 @@
+import asyncio
+import logging
+from typing import List, Dict, Any, Optional
+
+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,155 @@
+import asyncio
+import logging
+import re
+from typing import List, Dict, Any
+
+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={query}&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."""
+        import urllib.parse
+
+        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 50 < len(clean) < 300 and result["url"] in html[max(0, html.find(clean) - 500):html.find(clean) + 500]:
+                    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 async_playwright, Browser, BrowserContext, Page
+
+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 (ImportError, 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: "mcp-search-secret-key-change-in-production"
+  limiter: false
+  image_proxy: false
+  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,173 @@
+#!/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 json
+import subprocess
+import sys
+import time
+
+
+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"
+    module = service_name.replace("-", "_")
+
+    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())

tests/test_rate_limiter.py

@@ -0,0 +1,84 @@
+import asyncio
+import time
+
+import pytest
+import pytest_asyncio
+
+from lib.rate_limiter import RateLimiter
+
+
+@pytest.fixture
+def event_loop():
+    loop = asyncio.new_event_loop()
+    yield loop
+    loop.close()
+
+
+class TestRateLimiter:
+    """Test rate limiter enforces minimum delays between requests."""
+
+    def test_allows_first_request_immediately(self):
+        async def run():
+            limiter = RateLimiter(min_interval_seconds=0.5)
+            start = time.monotonic()
+            await limiter.acquire()
+            elapsed = time.monotonic() - start
+            assert elapsed < 0.1
+
+        asyncio.get_event_loop().run_until_complete(run())
+
+    def test_enforces_minimum_interval(self):
+        async def run():
+            limiter = RateLimiter(min_interval_seconds=0.3)
+            await limiter.acquire()
+
+            start = time.monotonic()
+            await limiter.acquire()
+            elapsed = time.monotonic() - start
+
+            assert elapsed >= 0.25
+
+        asyncio.get_event_loop().run_until_complete(run())
+
+    def test_consecutive_requests_space_correctly(self):
+        async def run():
+            interval = 0.2
+            limiter = RateLimiter(min_interval_seconds=interval)
+
+            times = []
+            for _ in range(5):
+                await limiter.acquire()
+                times.append(time.monotonic())
+
+            for i in range(1, len(times)):
+                gap = times[i] - times[i - 1]
+                assert gap >= interval * 0.8
+
+        asyncio.get_event_loop().run_until_complete(run())
+
+    def test_custom_interval(self):
+        async def run():
+            limiter = RateLimiter(min_interval_seconds=0.1)
+            await limiter.acquire()
+
+            start = time.monotonic()
+            await limiter.acquire()
+            elapsed = time.monotonic() - start
+
+            assert elapsed >= 0.05
+
+        asyncio.get_event_loop().run_until_complete(run())
+
+    def test_no_delay_after_long_pause(self):
+        async def run():
+            limiter = RateLimiter(min_interval_seconds=0.3)
+            await limiter.acquire()
+            await asyncio.sleep(0.5)
+
+            start = time.monotonic()
+            await limiter.acquire()
+            elapsed = time.monotonic() - start
+
+            assert elapsed < 0.1
+
+        asyncio.get_event_loop().run_until_complete(run())

tests/test_search_parsing.py

@@ -0,0 +1,73 @@
+import pytest
+
+
+class TestSearchResultFormatting:
+    """Test SearXNG result formatting."""
+
+    def test_format_single_result(self):
+        results = [
+            {
+                "title": "Python.org",
+                "url": "https://www.python.org/",
+                "content": "Python is a programming language",
+            }
+        ]
+        lines = []
+        for i, r in enumerate(results, 1):
+            lines.append(f"{i}. {r.get('title', 'N/A')}")
+            lines.append(f"   URL: {r.get('url', 'N/A')}")
+            snippet = r.get("content", "")[:200]
+            if snippet:
+                lines.append(f"   {snippet}")
+
+        assert len(lines) == 3
+        assert "1. Python.org" in lines[0]
+        assert "https://www.python.org/" in lines[1]
+        assert "Python is a programming language" in lines[2]
+
+    def test_format_multiple_results(self):
+        results = [
+            {"title": f"Result {i}", "url": f"https://example.com/{i}", "content": f"Content {i}"}
+            for i in range(5)
+        ]
+        lines = []
+        for i, r in enumerate(results, 1):
+            lines.append(f"{i}. {r.get('title', 'N/A')}")
+            lines.append(f"   URL: {r.get('url', 'N/A')}")
+            snippet = r.get("content", "")[:200]
+            if snippet:
+                lines.append(f"   {snippet}")
+
+        assert len(lines) == 15
+
+    def test_format_missing_fields(self):
+        results = [{"title": "No URL", "content": "Some content"}]
+        lines = []
+        for i, r in enumerate(results, 1):
+            lines.append(f"{i}. {r.get('title', 'N/A')}")
+            lines.append(f"   URL: {r.get('url', 'N/A')}")
+            snippet = r.get("content", "")[:200]
+            if snippet:
+                lines.append(f"   {snippet}")
+
+        assert "URL: N/A" in lines[1]
+
+    def test_format_empty_results(self):
+        results = []
+        lines = []
+        for i, r in enumerate(results, 1):
+            lines.append(f"{i}. {r.get('title', 'N/A')}")
+        assert lines == []
+
+    def test_snippet_truncation(self):
+        long_content = "x" * 500
+        results = [{"title": "Long", "url": "https://example.com", "content": long_content}]
+        lines = []
+        for i, r in enumerate(results, 1):
+            lines.append(f"{i}. {r.get('title', 'N/A')}")
+            lines.append(f"   URL: {r.get('url', 'N/A')}")
+            snippet = r.get("content", "")[:200]
+            if snippet:
+                lines.append(f"   {snippet}")
+
+        assert len(lines[2]) <= 203