fix: address PR review -- defensive formatting, input validation, error sanitization

- api_client: fix docstring (warns -> raises), add asyncio.Lock for safe lazy init
- formatters: guard against None/non-numeric score, use .get() for all dict access
- handlers: sanitize ValueError (no longer leaks internals), log errors server-side
- handlers: add _clamp_max_results to validate and bound top_k to [1, 100]
- tools: add minimum/maximum constraints to max_results schema
- tests: mock api_get in test_none_arguments_handled (no real HTTP)
- tests: add 8 new tests (score=None, clamp bounds, ValueError sanitized, schema bounds)
- tests: extract sys.path to conftest.py, remove from all test files
- .env.example: alphabetical key ordering (dotenv-linter compliance)

Skipped: N3 (env var rename -- deployment concern), N5 (emoji regex -- overkill)

45 tests pass.

Author: DevanshuNEU

mcp-server/.env.example

@@ -1,3 +1,3 @@
 # Backend API Configuration
-BACKEND_API_URL=http://localhost:8000
 API_KEY=your-api-key-here
+BACKEND_API_URL=http://localhost:8000

mcp-server/api_client.py

@@ -2,7 +2,9 @@
 
 Uses a module-level client to avoid creating new TCP connections per tool call.
 The client is initialized lazily on first use and reused for all subsequent calls.
+Concurrent access is serialized via asyncio.Lock to prevent duplicate clients.
 """
+import asyncio
 from typing import Any, Optional
 
 import httpx
@@ -12,10 +14,14 @@
 
 # Persistent client reused across all tool calls
 _client: Optional[httpx.AsyncClient] = None
+_client_lock: asyncio.Lock = asyncio.Lock()
 
 
 def _get_headers() -> dict[str, str]:
-    """Build auth headers. Warns if no API key is configured."""
+    """Return Authorization header with the configured API_KEY.
+
+    Raises ValueError if API_KEY is empty or unset.
+    """
     if not API_KEY:
         raise ValueError(
             "No API_KEY configured. Set API_KEY in .env or environment."
@@ -26,12 +32,13 @@ def _get_headers() -> dict[str, str]:
 async def get_client() -> httpx.AsyncClient:
     """Get or create the persistent HTTP client."""
     global _client
-    if _client is None or _client.is_closed:
-        _client = httpx.AsyncClient(
-            base_url=BACKEND_API_URL,
-            timeout=120.0,
-            headers=_get_headers(),
-        )
+    async with _client_lock:
+        if _client is None or _client.is_closed:
+            _client = httpx.AsyncClient(
+                base_url=BACKEND_API_URL,
+                timeout=120.0,
+                headers=_get_headers(),
+            )
     return _client
 
 

mcp-server/formatters.py

@@ -20,7 +20,11 @@ def format_search_results(result: dict) -> str:
         return output + "No results found.\n"
 
     for idx, res in enumerate(result["results"], 1):
-        score = res.get("score", 0) * 100
+        score_raw = res.get("score")
+        try:
+            score = float(score_raw) * 100
+        except (TypeError, ValueError):
+            score = 0
         name = res.get("name", "unknown")
         file_path = res.get("file_path", "unknown")
         lang = res.get("language", "unknown")
@@ -97,7 +101,7 @@ def format_dependency_graph(result: dict) -> str:
     if high_import:
         output += "## Files with Most Imports\n\n"
         for f in sorted(high_import, key=lambda x: x.get("imports", 0), reverse=True)[:5]:
-            output += f"- `{f['id']}` - imports {f['imports']} files\n"
+            output += f"- `{f.get('id', '<unknown>')}` - imports {f.get('imports', 0)} files\n"
 
     return output
 
@@ -115,14 +119,14 @@ def format_code_style(result: dict) -> str:
     if naming:
         output += "## Function Naming Conventions\n\n"
         for conv, info in naming.items():
-            output += f"- **{conv}:** {info['percentage']} ({info['count']} functions)\n"
+            output += f"- **{conv}:** {info.get('percentage', '?')} ({info.get('count', 0)} functions)\n"
         output += "\n"
 
     top_imports = result.get("top_imports")
     if top_imports:
         output += "## Most Common Imports\n\n"
         for item in top_imports[:10]:
-            output += f"- `{item['module']}` (used {item['count']}x)\n"
+            output += f"- `{item.get('module', '<unknown>')}` (used {item.get('count', 0)}x)\n"
 
     return output
 
@@ -168,7 +172,7 @@ def format_repository_insights(result: dict) -> str:
     if critical:
         output += "## Most Critical Files\n"
         for item in critical[:5]:
-            output += f"- `{item['file']}` ({item['dependents']} dependents)\n"
+            output += f"- `{item.get('file', '<unknown>')}` ({item.get('dependents', 0)} dependents)\n"
 
     return output
 

mcp-server/handlers.py

@@ -4,11 +4,14 @@
 Each handler follows the same pattern: call API, format response.
 Error handling is centralized in call_tool() so individual handlers stay clean.
 """
+import logging
 from typing import Any
 
 import httpx
 import mcp.types as types
 
+logger = logging.getLogger(__name__)
+
 from api_client import api_get, api_post
 from formatters import (
     format_codebase_dna,
@@ -21,12 +24,22 @@
 )
 
 
+def _clamp_max_results(raw: Any) -> int:
+    """Validate and clamp max_results to [1, 100]."""
+    try:
+        value = int(raw)
+    except (TypeError, ValueError):
+        return 10
+    return max(1, min(value, 100))
+
+
 async def _handle_search(args: dict[str, Any]) -> str:
     # Map tool schema's max_results to v2 API's top_k
+    top_k = _clamp_max_results(args.get("max_results", 10))
     payload = {
         "query": args["query"],
         "repo_id": args["repo_id"],
-        "top_k": args.get("max_results", 10),
+        "top_k": top_k,
         "use_reranking": True,
     }
     result = await api_post("/search/v2", json=payload)
@@ -89,7 +102,9 @@ def _safe_error_message(tool_name: str, args: dict[str, Any], error: Exception)
     if isinstance(error, httpx.ConnectError):
         return f"Cannot connect to backend for tool '{tool_name}'. Is the server running?"
     if isinstance(error, ValueError):
-        return str(error)
+        logger.warning("ValueError in tool '%s' (repo: %s): %s", tool_name, repo_id, error)
+        return f"Tool input error for '{tool_name}' (repo: {repo_id})"
+    logger.exception("Unexpected error in tool '%s' (repo: %s)", tool_name, repo_id)
     return f"Unexpected error in tool '{tool_name}' (repo: {repo_id})"
 
 

mcp-server/tests/conftest.py

@@ -0,0 +1,8 @@
+"""Shared test configuration.
+
+Adds the mcp-server root to sys.path so tests can import modules directly.
+"""
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

mcp-server/tests/test_config.py

@@ -1,9 +1,5 @@
 """Tests for MCP server configuration."""
 import pytest
-import sys
-import os
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 from config import API_PREFIX, SERVER_NAME, SERVER_VERSION, BACKEND_API_URL
 

mcp-server/tests/test_formatters.py

@@ -4,11 +4,6 @@
 without any mocking or network calls.
 """
 import pytest
-import sys
-import os
-
-# Add parent directory to path so we can import the modules
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 from formatters import (
     format_search_results,
@@ -69,6 +64,15 @@ def test_v1_fallback(self):
         assert "Found 0 results" in output
         assert "(v1)" in output
 
+    def test_none_score_handled(self):
+        """score=None should not crash the formatter."""
+        result = {"total": 1, "cached": False, "search_version": "v2", "results": [{
+            "name": "test", "file_path": "t.py", "code": "pass",
+            "language": "python", "score": None, "line_start": 1, "line_end": 1,
+        }]}
+        output = format_search_results(result)
+        assert "0% match" in output
+
     def test_no_emoji_in_output(self):
         """CLAUDE.md violation check: no emojis anywhere in formatted output."""
         result = {"total": 1, "cached": True, "search_version": "v2", "results": [{

mcp-server/tests/test_handlers.py

@@ -4,16 +4,11 @@
 dispatch logic and error handling without network calls.
 """
 import pytest
-import sys
-import os
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
 from unittest.mock import AsyncMock, patch
 import httpx
 import mcp.types as types
 
-from handlers import call_tool, _safe_error_message
+from handlers import call_tool, _safe_error_message, _clamp_max_results
 
 
 # -- Dispatch --
@@ -68,11 +63,35 @@ async def test_dna_calls_correct_endpoint(self, mock_get):
         assert "/repos/r1/dna" in call_path
 
     @pytest.mark.asyncio
-    async def test_none_arguments_handled(self):
+    @patch("handlers.api_get", new_callable=AsyncMock)
+    async def test_none_arguments_handled(self, mock_get):
         """call_tool(name, None) should not crash."""
+        mock_get.return_value = {"repositories": []}
         result = await call_tool("list_repositories", None)
-        # Will fail on network, but should not crash on None args
         assert len(result) == 1
+        assert "No repositories indexed" in result[0].text
+
+
+# -- Input validation --
+
+class TestClampMaxResults:
+    def test_default_on_none(self):
+        assert _clamp_max_results(None) == 10
+
+    def test_default_on_string(self):
+        assert _clamp_max_results("abc") == 10
+
+    def test_clamps_zero_to_one(self):
+        assert _clamp_max_results(0) == 1
+
+    def test_clamps_negative(self):
+        assert _clamp_max_results(-5) == 1
+
+    def test_clamps_over_max(self):
+        assert _clamp_max_results(500) == 100
+
+    def test_valid_value_passes(self):
+        assert _clamp_max_results(25) == 25
 
 
 # -- Error handling --
@@ -99,11 +118,14 @@ def test_connect_error(self):
         msg = _safe_error_message("search_code", {}, error)
         assert "Cannot connect" in msg
 
-    def test_value_error_passthrough(self):
-        """ValueError messages are user-facing (e.g. missing API key)."""
+    def test_value_error_sanitized(self):
+        """ValueError should not leak internal details."""
         error = ValueError("No API_KEY configured")
-        msg = _safe_error_message("search_code", {}, error)
-        assert "No API_KEY configured" in msg
+        msg = _safe_error_message("search_code", {"repo_id": "r1"}, error)
+        assert "Tool input error" in msg
+        assert "search_code" in msg
+        # Internal message should NOT be in output
+        assert "No API_KEY" not in msg
 
     def test_generic_error_hides_details(self):
         error = RuntimeError("internal traceback info")

mcp-server/tests/test_tools.py

@@ -4,10 +4,6 @@
 and that adding/removing tools doesn't break the schema contract.
 """
 import pytest
-import sys
-import os
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 from tools import get_tool_schemas
 
@@ -62,6 +58,14 @@ def test_repo_tools_require_repo_id(self):
             required = schemas[name].inputSchema.get("required", [])
             assert "repo_id" in required, f"{name} should require repo_id"
 
+    def test_search_max_results_bounded(self):
+        """max_results schema should have min/max to prevent invalid searches."""
+        schemas = {t.name: t for t in get_tool_schemas()}
+        max_results = schemas["search_code"].inputSchema["properties"]["max_results"]
+        assert max_results["type"] == "integer"
+        assert max_results["minimum"] >= 1
+        assert max_results["maximum"] > max_results["minimum"]
+
     def test_list_repos_has_no_required_fields(self):
         schemas = {t.name: t for t in get_tool_schemas()}
         list_repos = schemas["list_repositories"]

mcp-server/tools.py

@@ -36,6 +36,8 @@ def get_tool_schemas() -> list[types.Tool]:
                         "type": "integer",
                         "description": "Maximum number of results (default: 10)",
                         "default": 10,
+                        "minimum": 1,
+                        "maximum": 100,
                     },
                 },
                 "required": ["query", "repo_id"],