[MCP] Enforce usage of caller-supplied API key

src/mcp_components/tools.py

@@ -186,9 +186,10 @@ def fetch_search_data(params: dict[str, Any] | None) -> dict[str, Any]:
     if not api_key:
         raise RuntimeError("Error: Unable to access API key from request context")
 
+    # api_key set last so caller params can never override the trusted key.
     search_params = {
-        "api_key": api_key,
         "engine": "google_light",
         **(params or {}),
+        "api_key": api_key,
     }
     return serpapi.search(search_params).as_dict()

src/server.py

@@ -137,5 +137,4 @@ async def healthcheck_handler(request):
     host = os.getenv("MCP_HOST", "0.0.0.0")
     port = int(os.getenv("MCP_PORT", "8000"))
 
-    workers = int(os.getenv("WEB_CONCURRENCY", "4"))
-    uvicorn.run(starlette_app, host=host, port=port, ws="none", workers=workers)
+    uvicorn.run(starlette_app, host=host, port=port, ws="none")

tests/test_server.py

@@ -264,6 +264,35 @@ def capture(params):
     assert captured["engine"] == "google_news"
 
 
+async def test_search_ignores_caller_supplied_api_key(monkeypatch):
+    # Caller-supplied api_key must never override the authenticated key.
+    captured = {}
+
+    def capture(params):
+        captured.update(params)
+        return serp_results({})
+
+    use_request(monkeypatch, real_request(state={"api_key": "TRUSTED"}))
+    use_search(monkeypatch, capture)
+    await mcp_tools.search(params={"q": "x", "api_key": "CALLER_CONTROLLED"})
+    assert captured["api_key"] == "TRUSTED"
+    assert captured["q"] == "x"
+
+
+async def test_search_apps_ignore_caller_supplied_api_key(monkeypatch):
+    # App variants share fetch_search_data, so the same guard must hold.
+    captured = {}
+
+    def capture(params):
+        captured.update(params)
+        return serp_results(_SAMPLE_PAYLOAD)
+
+    use_request(monkeypatch, real_request(state={"api_key": "TRUSTED"}))
+    use_search(monkeypatch, capture)
+    await mcp_apps.search_table(params={"q": "x", "api_key": "CALLER_CONTROLLED"})
+    assert captured["api_key"] == "TRUSTED"
+
+
 @pytest.mark.parametrize(
     "status, fragment",
     [