diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..69cc6fc --- /dev/null +++ b/.env.example @@ -0,0 +1,2 @@ +# SearXNG secret key (generate: python -c "import secrets; print(secrets.token_hex(32))") +SEARXNG_SECRET= diff --git a/docker-compose.yml b/docker-compose.yml index 4e291f4..f371a0f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -2,16 +2,25 @@ services: searxng: image: searxng/searxng:latest container_name: searxng - ports: - - "8080:8080" + 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/health')"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 15s google-search: build: @@ -26,9 +35,19 @@ services: - SEARXNG_URL=http://searxng:8080 - PYTHONUNBUFFERED=1 depends_on: - - searxng + 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:3001/health')"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 10s duckduckgo-search: build: @@ -43,6 +62,22 @@ services: - SEARXNG_URL=http://searxng:8080 - PYTHONUNBUFFERED=1 depends_on: - - searxng + searxng: + condition: service_healthy restart: unless-stopped - stop_grace_period: 5s \ No newline at end of file + 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 diff --git a/duckduckgo-mcp/server.py b/duckduckgo-mcp/server.py index 9271e77..d6a8be7 100644 --- a/duckduckgo-mcp/server.py +++ b/duckduckgo-mcp/server.py @@ -4,12 +4,16 @@ 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 starlette.applications import Starlette +from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request from starlette.responses import JSONResponse, Response from starlette.routing import Mount, Route @@ -17,7 +21,7 @@ from mcp.types import Tool, TextContent logging.basicConfig( level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + format="%(asctime)s [%(levelname)s] %(message)s", stream=sys.stderr, ) logger = logging.getLogger("duckduckgo-mcp") @@ -28,6 +32,40 @@ SEARXNG_URL = os.environ.get("SEARXNG_URL", "http://searxng:8080") RATE_LIMIT_SECONDS = float(os.environ.get("RATE_LIMIT_SECONDS", "3")) +# --- 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 @@ -36,36 +74,72 @@ class RateLimiter: async def acquire(self): async with self._lock: - now = asyncio.get_event_loop().time() + 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_event_loop().time() + 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() + + async with self._lock: + timestamps = self._requests[client_ip] + cutoff = now - self.window_seconds + 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() - 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() + 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 [ @@ -91,7 +165,13 @@ async def search_http(request: Request): return JSONResponse({"query": query, "results": results}) except Exception as e: logger.exception("Search failed") - return JSONResponse({"error": str(e)}, status_code=500) + 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 --- @@ -132,7 +212,8 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: try: results = await do_search(query, num_results, "duckduckgo") except Exception as e: - return [TextContent(type="text", text=f"Search failed: {e}")] + 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}")] @@ -166,12 +247,14 @@ async def handle_sse(request): starlette_app = Starlette( - debug=True, + debug=False, routes=[ + Route("/health", endpoint=health_check), Route("/search", endpoint=search_http), Route("/sse", endpoint=handle_sse), Mount("/messages/", app=sse.handle_post_message), ], + middleware=[RateLimitMiddleware], ) @@ -195,4 +278,4 @@ async def main(): if __name__ == "__main__": - asyncio.run(main()) \ No newline at end of file + asyncio.run(main()) diff --git a/google-mcp/server.py b/google-mcp/server.py index 964df5d..d031876 100644 --- a/google-mcp/server.py +++ b/google-mcp/server.py @@ -4,12 +4,16 @@ 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 starlette.applications import Starlette +from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request from starlette.responses import JSONResponse, Response from starlette.routing import Mount, Route @@ -17,7 +21,7 @@ from mcp.types import Tool, TextContent logging.basicConfig( level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + format="%(asctime)s [%(levelname)s] %(message)s", stream=sys.stderr, ) logger = logging.getLogger("google-mcp") @@ -28,6 +32,40 @@ SEARXNG_URL = os.environ.get("SEARXNG_URL", "http://searxng:8080") RATE_LIMIT_SECONDS = float(os.environ.get("RATE_LIMIT_SECONDS", "5")) +# --- 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 @@ -36,36 +74,72 @@ class RateLimiter: async def acquire(self): async with self._lock: - now = asyncio.get_event_loop().time() + 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_event_loop().time() + 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() + + async with self._lock: + timestamps = self._requests[client_ip] + cutoff = now - self.window_seconds + 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() - 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() + 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 [ @@ -91,7 +165,13 @@ async def search_http(request: Request): return JSONResponse({"query": query, "results": results}) except Exception as e: logger.exception("Search failed") - return JSONResponse({"error": str(e)}, status_code=500) + return JSONResponse({"error": "Search service unavailable"}, status_code=503) + + +# --- Health check (#6) --- + +async def health_check(request: Request): + return JSONResponse({"status": "ok", "service": "google-mcp"}) # --- MCP tool handlers --- @@ -132,7 +212,8 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: try: results = await do_search(query, num_results, "google") except Exception as e: - return [TextContent(type="text", text=f"Search failed: {e}")] + 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}")] @@ -166,12 +247,14 @@ async def handle_sse(request): starlette_app = Starlette( - debug=True, + debug=False, routes=[ + Route("/health", endpoint=health_check), Route("/search", endpoint=search_http), Route("/sse", endpoint=handle_sse), Mount("/messages/", app=sse.handle_post_message), ], + middleware=[RateLimitMiddleware], ) @@ -195,4 +278,4 @@ async def main(): if __name__ == "__main__": - asyncio.run(main()) \ No newline at end of file + asyncio.run(main()) diff --git a/searxng-settings.yml b/searxng-settings.yml index 2753454..f719062 100644 --- a/searxng-settings.yml +++ b/searxng-settings.yml @@ -19,9 +19,9 @@ search: server: port: 8080 bind_address: "0.0.0.0" - secret_key: "mcp-search-secret-key-change-in-production" - limiter: false - image_proxy: false + secret_key: "${SEARXNG_SECRET}" + limiter: true + image_proxy: true method: "GET" engines: @@ -85,4 +85,4 @@ outgoing: 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 \ No newline at end of file + max_keepalive_connection: 20