refactor(gatekeeper): Drop Chat Completions API from OpenAI client

Author: subpop

src/linux_mcp_server/gatekeeper/openai_client.py

@@ -1,19 +1,15 @@
-"""OpenAI Responses and Chat Completions clients for the gatekeeper."""
+"""OpenAI Responses API client for the gatekeeper."""
 
 import os
 
 from typing import Any
-from urllib.parse import urlparse
 
 from linux_mcp_server.config import CONFIG
 from linux_mcp_server.config import ReasoningEffort
 from linux_mcp_server.gatekeeper.http_utils import DEFAULT_TIMEOUT_SECONDS
-from linux_mcp_server.gatekeeper.http_utils import GatekeeperHTTPError
 from linux_mcp_server.gatekeeper.http_utils import normalize_model_id
 from linux_mcp_server.gatekeeper.http_utils import post_json
-from linux_mcp_server.gatekeeper.schema import openai_response_format
 from linux_mcp_server.gatekeeper.schema import openai_text_format
-from linux_mcp_server.gatekeeper.usage import extract_openai_chat_completions_usage
 from linux_mcp_server.gatekeeper.usage import extract_openai_responses_usage
 from linux_mcp_server.models import GatekeeperCompletion
 
@@ -43,26 +39,6 @@ def _get_openai_base_url() -> str:
     return (configured or os.environ.get("OPENAI_API_BASE") or OPENAI_DEFAULT_BASE_URL).rstrip("/")
 
 
-def _prefers_openai_chat_completions(base_url: str) -> bool:
-    """Hosts known to expose only Chat Completions, not the Responses API."""
-    path = urlparse(base_url).path or ""
-    return "/endpoints/openapi" in path
-
-
-def _openai_template_kwargs() -> dict[str, Any]:
-    if CONFIG.gatekeeper.openai is None:
-        return {}
-    return CONFIG.gatekeeper.openai.template_kwargs
-
-
-def _apply_chat_completions_extras(body: dict[str, Any]) -> dict[str, Any]:
-    """Merge template_kwargs into Chat Completions bodies (llama.cpp, etc.)."""
-    template_kwargs = _openai_template_kwargs()
-    if template_kwargs:
-        body["chat_template_kwargs"] = template_kwargs
-    return body
-
-
 def _build_responses_body(prompt: str, *, max_tokens: int) -> dict[str, Any]:
     body: dict[str, Any] = {
         "model": normalize_model_id(CONFIG.gatekeeper.model or ""),
@@ -79,21 +55,6 @@ def _build_responses_body(prompt: str, *, max_tokens: int) -> dict[str, Any]:
     return body
 
 
-def build_chat_completions_body(prompt: str, *, max_tokens: int) -> dict[str, Any]:
-    body: dict[str, Any] = {
-        "model": normalize_model_id(CONFIG.gatekeeper.model or ""),
-        "messages": [{"role": "user", "content": prompt}],
-        "max_completion_tokens": max_tokens,
-        "temperature": CONFIG.gatekeeper.temperature,
-    }
-    if CONFIG.gatekeeper.structured_output:
-        body["response_format"] = openai_response_format()
-    reasoning_effort = CONFIG.gatekeeper.reasoning_effort
-    if reasoning_effort is not None:
-        body["reasoning_effort"] = reasoning_effort.value
-    return _apply_chat_completions_extras(body)
-
-
 def _extract_responses_text(response: dict[str, Any]) -> str:
     output_text = response.get("output_text")
     if isinstance(output_text, str) and output_text.strip():
@@ -114,15 +75,6 @@ def _extract_responses_text(response: dict[str, Any]) -> str:
     return "".join(chunks).strip()
 
 
-def extract_chat_completions_text(response: dict[str, Any]) -> str:
-    choices = response.get("choices", [])
-    if not choices:
-        return ""
-    message = choices[0].get("message", {})
-    content = message.get("content")
-    return (content or "").strip() if isinstance(content, str) else ""
-
-
 async def complete_openai(
     prompt: str, *, max_tokens: int, timeout: int = DEFAULT_TIMEOUT_SECONDS
 ) -> GatekeeperCompletion:
@@ -131,37 +83,16 @@ async def complete_openai(
         **_openai_auth_headers(),
         "Content-Type": "application/json",
     }
-
-    # Try the Responses API first, falling back to Chat Completions if it's not available.
-    if not _prefers_openai_chat_completions(base_url):
-        try:
-            response = await post_json(
-                provider="openai",
-                url=f"{base_url}/responses",
-                headers=headers,
-                body=_build_responses_body(prompt, max_tokens=max_tokens),
-                timeout=timeout,
-            )
-            usage = extract_openai_responses_usage(response)
-            return GatekeeperCompletion(
-                text=_extract_responses_text(response),
-                prompt_tokens=usage.input_tokens,
-                completion_tokens=usage.output_tokens,
-            )
-        except GatekeeperHTTPError as exc:
-            if exc.status_code not in {404, 405}:
-                raise
-
     response = await post_json(
         provider="openai",
-        url=f"{base_url}/chat/completions",
+        url=f"{base_url}/responses",
         headers=headers,
-        body=build_chat_completions_body(prompt, max_tokens=max_tokens),
+        body=_build_responses_body(prompt, max_tokens=max_tokens),
         timeout=timeout,
     )
-    usage = extract_openai_chat_completions_usage(response)
+    usage = extract_openai_responses_usage(response)
     return GatekeeperCompletion(
-        text=extract_chat_completions_text(response),
+        text=_extract_responses_text(response),
         prompt_tokens=usage.input_tokens,
         completion_tokens=usage.output_tokens,
     )

src/linux_mcp_server/gatekeeper/vertex_ai_client.py

@@ -12,11 +12,11 @@
 from linux_mcp_server.gatekeeper.http_utils import DEFAULT_TIMEOUT_SECONDS
 from linux_mcp_server.gatekeeper.http_utils import normalize_model_id
 from linux_mcp_server.gatekeeper.http_utils import post_json
-from linux_mcp_server.gatekeeper.openai_client import build_chat_completions_body
-from linux_mcp_server.gatekeeper.openai_client import extract_chat_completions_text
+from linux_mcp_server.gatekeeper.openai_client import _build_responses_body
+from linux_mcp_server.gatekeeper.openai_client import _extract_responses_text
 from linux_mcp_server.gatekeeper.usage import extract_anthropic_usage
 from linux_mcp_server.gatekeeper.usage import extract_gemini_usage
-from linux_mcp_server.gatekeeper.usage import extract_openai_chat_completions_usage
+from linux_mcp_server.gatekeeper.usage import extract_openai_responses_usage
 from linux_mcp_server.models import GatekeeperCompletion
 
 
@@ -104,14 +104,14 @@ async def _complete_openai_compatible_on_vertex(prompt: str, *, max_tokens: int,
     base_url = _get_vertex_openapi_base_url()
     response = await post_json(
         provider="openai",
-        url=f"{base_url}/chat/completions",
+        url=f"{base_url}/responses",
         headers={**_vertex_auth_headers(), "Content-Type": "application/json"},
-        body=build_chat_completions_body(prompt, max_tokens=max_tokens),
+        body=_build_responses_body(prompt, max_tokens=max_tokens),
         timeout=timeout,
     )
-    usage = extract_openai_chat_completions_usage(response)
+    usage = extract_openai_responses_usage(response)
     return GatekeeperCompletion(
-        text=extract_chat_completions_text(response),
+        text=_extract_responses_text(response),
         prompt_tokens=usage.input_tokens,
         completion_tokens=usage.output_tokens,
     )

tests/gatekeeper/test_check_run_script.py

@@ -232,7 +232,7 @@ async def test_vertex_ai_provider_uses_openapi_base_url(self, mocker):
         mock_post = mocker.patch(
             "linux_mcp_server.gatekeeper.vertex_ai_client.post_json",
             new_callable=mocker.AsyncMock,
-            return_value={"choices": [{"message": {"content": '{"status": "OK"}'}}]},
+            return_value={"output_text": '{"status": "OK", "detail": ""}'},
         )
         mocker.patch.object(
             CONFIG,
@@ -251,5 +251,5 @@ async def test_vertex_ai_provider_uses_openapi_base_url(self, mocker):
 
         await check_run_script(description="test", script_type="bash", script="echo hi", readonly=True)
 
-        assert mock_post.call_args.kwargs["url"].endswith("/chat/completions")
+        assert mock_post.call_args.kwargs["url"].endswith("/responses")
         assert mock_post.call_args.kwargs["headers"]["Authorization"] == "Bearer gcp-token"

tests/gatekeeper/test_openai_client.py

@@ -8,23 +8,6 @@
 from linux_mcp_server.gatekeeper import openai_client
 from linux_mcp_server.gatekeeper.http_utils import GatekeeperHTTPError
 from linux_mcp_server.gatekeeper.openai_client import _openai_reasoning_block
-from linux_mcp_server.gatekeeper.openai_client import _prefers_openai_chat_completions
-
-
-@pytest.mark.parametrize(
-    "base_url,expected",
-    [
-        ("https://api.openai.com/v1", False),
-        ("http://localhost:11434/v1", False),
-        ("https://example.com/v1", False),
-        (
-            "https://aiplatform.googleapis.com/v1/projects/p/locations/global/endpoints/openapi",
-            True,
-        ),
-    ],
-)
-def test_prefers_openai_chat_completions(base_url, expected):
-    assert _prefers_openai_chat_completions(base_url) is expected
 
 
 def test_openai_reasoning_block_none():
@@ -71,7 +54,7 @@ async def test_complete_openai_uses_responses_api(self, gatekeeper_config, mocke
         assert body["reasoning"] == {"effort": "low"}
         assert body["text"]["format"]["type"] == "json_schema"
 
-    async def test_complete_openai_uses_responses_api_for_ollama(self, gatekeeper_config, mocker):
+    async def test_complete_openai_uses_responses_api_for_custom_base_url(self, gatekeeper_config, mocker):
         gatekeeper_config.openai = OpenAIGatekeeperConfig(base_url="http://localhost:11434/v1")
         mock_post = mocker.patch(
             "linux_mcp_server.gatekeeper.openai_client.post_json",
@@ -84,30 +67,15 @@ async def test_complete_openai_uses_responses_api_for_ollama(self, gatekeeper_co
         assert result.text == '{"status": "OK", "detail": ""}'
         assert mock_post.call_args.kwargs["url"] == "http://localhost:11434/v1/responses"
 
-    async def test_complete_openai_falls_back_to_chat_completions_on_404(self, gatekeeper_config, mocker):
-        gatekeeper_config.openai = OpenAIGatekeeperConfig(base_url="https://models.example.com/v1")
-        mock_post = mocker.patch(
+    async def test_complete_openai_propagates_responses_api_errors(self, gatekeeper_config, mocker):
+        mocker.patch(
             "linux_mcp_server.gatekeeper.openai_client.post_json",
             new_callable=mocker.AsyncMock,
-            side_effect=[
-                GatekeeperHTTPError("openai", 404, "not found"),
-                {
-                    "choices": [{"message": {"content": '{"status": "OK", "detail": ""}'}}],
-                    "usage": {"prompt_tokens": 9, "completion_tokens": 2},
-                },
-            ],
+            side_effect=GatekeeperHTTPError("openai", 404, "not found"),
         )
 
-        result = await openai_client.complete_openai("prompt", max_tokens=8000)
-
-        assert result.text == '{"status": "OK", "detail": ""}'
-        assert result.prompt_tokens == 9
-        assert result.completion_tokens == 2
-        assert mock_post.call_args_list[0].kwargs["url"] == "https://models.example.com/v1/responses"
-        assert mock_post.call_args_list[1].kwargs["url"] == "https://models.example.com/v1/chat/completions"
-        body = mock_post.call_args_list[1].kwargs["body"]
-        assert body["response_format"]["type"] == "json_schema"
-        assert body["reasoning_effort"] == "low"
+        with pytest.raises(GatekeeperHTTPError, match="not found"):
+            await openai_client.complete_openai("prompt", max_tokens=8000)
 
     async def test_structured_output_disabled(self, gatekeeper_config, mocker):
         gatekeeper_config.structured_output = False
@@ -121,22 +89,3 @@ async def test_structured_output_disabled(self, gatekeeper_config, mocker):
 
         body = mock_post.call_args.kwargs["body"]
         assert "text" not in body
-
-    async def test_template_kwargs_on_chat_completions(self, gatekeeper_config, mocker):
-        gatekeeper_config.openai = OpenAIGatekeeperConfig(
-            base_url="https://models.example.com/v1",
-            template_kwargs={"enable_thinking": False},
-        )
-        mock_post = mocker.patch(
-            "linux_mcp_server.gatekeeper.openai_client.post_json",
-            new_callable=mocker.AsyncMock,
-            side_effect=[
-                GatekeeperHTTPError("openai", 404, "not found"),
-                {"choices": [{"message": {"content": '{"status": "OK"}'}}]},
-            ],
-        )
-
-        await openai_client.complete_openai("prompt", max_tokens=8000)
-
-        body = mock_post.call_args_list[1].kwargs["body"]
-        assert body["chat_template_kwargs"] == {"enable_thinking": False}

tests/gatekeeper/test_vertex_ai_client.py

@@ -91,13 +91,13 @@ async def test_complete_openai_compatible_on_vertex(self, gatekeeper_config, moc
         mock_post = mocker.patch(
             "linux_mcp_server.gatekeeper.vertex_ai_client.post_json",
             new_callable=mocker.AsyncMock,
-            return_value={"choices": [{"message": {"content": '{"status": "OK"}'}}]},
+            return_value={"output_text": '{"status": "OK"}'},
         )
 
         result = await vertex_ai_client.complete_vertex_ai("prompt", max_tokens=8000)
 
         assert result.text == '{"status": "OK"}'
-        assert mock_post.call_args.kwargs["url"].endswith("/chat/completions")
+        assert mock_post.call_args.kwargs["url"].endswith("/responses")
         assert mock_post.call_args.kwargs["headers"]["Authorization"] == "Bearer gcp-token"
 
     async def test_custom_openapi_base_url(self, gatekeeper_config, mocker):
@@ -109,7 +109,7 @@ async def test_custom_openapi_base_url(self, gatekeeper_config, mocker):
         mock_post = mocker.patch(
             "linux_mcp_server.gatekeeper.vertex_ai_client.post_json",
             new_callable=mocker.AsyncMock,
-            return_value={"choices": [{"message": {"content": '{"status": "OK"}'}}]},
+            return_value={"output_text": '{"status": "OK"}'},
         )
 
         await vertex_ai_client.complete_vertex_ai("prompt", max_tokens=8000)