refactor: reorganize provider-specific funcions

Move provider-specific methods out of http_utils into their relevant
client file.

Author: subpop

src/linux_mcp_server/gatekeeper/anthropic_client.py

@@ -1,18 +1,44 @@
 """Anthropic Messages API client for the gatekeeper."""
 
+import os
+
 from typing import Any
 
 from linux_mcp_server.config import CONFIG
-from linux_mcp_server.gatekeeper.http_utils import ANTHROPIC_API_URL
-from linux_mcp_server.gatekeeper.http_utils import ANTHROPIC_API_VERSION
-from linux_mcp_server.gatekeeper.http_utils import ANTHROPIC_DEFAULT_MAX_TOKENS
-from linux_mcp_server.gatekeeper.http_utils import anthropic_thinking_block
+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 get_anthropic_api_key
 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.llm import GatekeeperCompletion
 from linux_mcp_server.gatekeeper.schema import anthropic_output_config
+from linux_mcp_server.models import GatekeeperCompletion
+
+
+ANTHROPIC_API_URL = "https://api.anthropic.com/v1/messages"
+ANTHROPIC_API_VERSION = "2023-06-01"
+ANTHROPIC_DEFAULT_MAX_TOKENS = 4096
+
+
+def _get_anthropic_api_key() -> str:
+    api_key = os.environ.get("ANTHROPIC_API_KEY")
+    if not api_key:
+        raise ValueError("ANTHROPIC_API_KEY is required for Anthropic gatekeeper provider.")
+    return api_key
+
+
+def _anthropic_thinking_block(reasoning_effort: ReasoningEffort | None) -> dict[str, Any] | None:
+    if reasoning_effort is None or reasoning_effort in {ReasoningEffort.NONE, ReasoningEffort.DEFAULT}:
+        return None
+    budget_by_effort = {
+        ReasoningEffort.MINIMAL: 1024,
+        ReasoningEffort.LOW: 4096,
+        ReasoningEffort.MEDIUM: 8192,
+        ReasoningEffort.HIGH: 16384,
+        ReasoningEffort.XHIGH: 32768,
+    }
+    budget = budget_by_effort.get(reasoning_effort)
+    if budget is None:
+        return None
+    return {"type": "enabled", "budget_tokens": budget}
 
 
 def build_messages_body(prompt: str, *, include_model: bool) -> dict[str, Any]:
@@ -25,7 +51,7 @@ def build_messages_body(prompt: str, *, include_model: bool) -> dict[str, Any]:
         body["model"] = normalize_model_id(CONFIG.gatekeeper.model or "")
     if CONFIG.gatekeeper.structured_output:
         body["output_config"] = anthropic_output_config()
-    thinking = anthropic_thinking_block(CONFIG.gatekeeper.reasoning_effort)
+    thinking = _anthropic_thinking_block(CONFIG.gatekeeper.reasoning_effort)
     if thinking is not None:
         body["thinking"] = thinking
     return body
@@ -42,7 +68,7 @@ def extract_messages_text(response: dict[str, Any]) -> str:
 
 def complete_anthropic(prompt: str, *, timeout: int = DEFAULT_TIMEOUT_SECONDS) -> GatekeeperCompletion:
     headers = {
-        "x-api-key": get_anthropic_api_key(),
+        "x-api-key": _get_anthropic_api_key(),
         "anthropic-version": ANTHROPIC_API_VERSION,
         "Content-Type": "application/json",
     }

src/linux_mcp_server/gatekeeper/gemini_client.py

@@ -1,24 +1,47 @@
 """Gemini generateContent client for the gatekeeper."""
 
+import os
+
 from typing import Any
 
 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 gemini_thinking_level
-from linux_mcp_server.gatekeeper.http_utils import get_google_api_key
-from linux_mcp_server.gatekeeper.http_utils import GOOGLE_AI_BASE_URL
 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.llm import GatekeeperCompletion
 from linux_mcp_server.gatekeeper.schema import gemini_generation_config
+from linux_mcp_server.models import GatekeeperCompletion
+
+
+GOOGLE_AI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta"
+
+
+def _get_google_api_key() -> str:
+    api_key = os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY")
+    if not api_key:
+        raise ValueError("GOOGLE_API_KEY or GEMINI_API_KEY is required for Gemini direct backend.")
+    return api_key
+
+
+def _gemini_thinking_level(reasoning_effort: ReasoningEffort | None) -> str | None:
+    if reasoning_effort is None or reasoning_effort in {ReasoningEffort.NONE, ReasoningEffort.DEFAULT}:
+        return None
+    mapping = {
+        ReasoningEffort.MINIMAL: "MINIMAL",
+        ReasoningEffort.LOW: "LOW",
+        ReasoningEffort.MEDIUM: "MEDIUM",
+        ReasoningEffort.HIGH: "HIGH",
+        ReasoningEffort.XHIGH: "HIGH",
+    }
+    return mapping.get(reasoning_effort)
 
 
 def build_gemini_body(prompt: str) -> dict[str, Any]:
     generation_config = gemini_generation_config(
         temperature=CONFIG.gatekeeper.temperature,
         structured_output=CONFIG.gatekeeper.structured_output,
     )
-    thinking_level = gemini_thinking_level(CONFIG.gatekeeper.reasoning_effort)
+    thinking_level = _gemini_thinking_level(CONFIG.gatekeeper.reasoning_effort)
     if thinking_level is not None:
         generation_config["thinkingConfig"] = {"thinkingLevel": thinking_level}
     return {
@@ -41,7 +64,7 @@ def extract_gemini_text(response: dict[str, Any]) -> str:
 
 def complete_gemini(prompt: str, *, timeout: int = DEFAULT_TIMEOUT_SECONDS) -> GatekeeperCompletion:
     model = normalize_model_id(CONFIG.gatekeeper.model or "")
-    api_key = get_google_api_key()
+    api_key = _get_google_api_key()
     response = post_json(
         provider="gemini",
         url=f"{GOOGLE_AI_BASE_URL}/models/{model}:generateContent?key={api_key}",

src/linux_mcp_server/gatekeeper/http_utils.py

@@ -1,29 +1,12 @@
 """Shared helpers for gatekeeper HTTP clients."""
 
-import os
-
 from typing import Any
-from typing import Literal
-from urllib.parse import urlparse
 
 import requests
 
-from linux_mcp_server.config import CONFIG
-from linux_mcp_server.config import ReasoningEffort
-
 
 DEFAULT_TIMEOUT_SECONDS = 120
 
-OPENAI_DEFAULT_BASE_URL = "https://api.openai.com/v1"
-OPENROUTER_DEFAULT_BASE_URL = "https://openrouter.ai/api/v1"
-
-ANTHROPIC_API_URL = "https://api.anthropic.com/v1/messages"
-ANTHROPIC_API_VERSION = "2023-06-01"
-ANTHROPIC_VERTEX_VERSION = "vertex-2023-10-16"
-ANTHROPIC_DEFAULT_MAX_TOKENS = 4096
-
-GOOGLE_AI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta"
-
 
 class GatekeeperHTTPError(RuntimeError):
     """Raised when an LLM provider returns an error response."""
@@ -50,136 +33,8 @@ def post_json(
     return response.json()
 
 
-def get_openai_base_url() -> str:
-    configured = CONFIG.gatekeeper.openai.base_url if CONFIG.gatekeeper.openai else None
-    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 normalize_model_id(model: str) -> str:
     for prefix in ("openai/", "anthropic/", "vertex_ai/", "gemini/"):
         if model.startswith(prefix):
             return model[len(prefix) :]
     return model
-
-
-def normalize_openrouter_model_id(model: str) -> str:
-    if model.startswith("openrouter/"):
-        return model[len("openrouter/") :]
-    return model
-
-
-def get_openrouter_base_url() -> str:
-    configured = CONFIG.gatekeeper.openrouter.base_url if CONFIG.gatekeeper.openrouter else None
-    return (configured or OPENROUTER_DEFAULT_BASE_URL).rstrip("/")
-
-
-def vertex_api_style(model: str) -> Literal["anthropic", "gemini", "openai_compatible"]:
-    normalized = normalize_model_id(model)
-    if normalized.startswith("claude"):
-        return "anthropic"
-    if normalized.startswith("gemini"):
-        return "gemini"
-    return "openai_compatible"
-
-
-def get_vertex_openapi_base_url() -> str:
-    cfg = CONFIG.gatekeeper.vertex_ai
-    if cfg and cfg.base_url:
-        return cfg.base_url.rstrip("/")
-    from linux_mcp_server.gatekeeper.gcp_auth import get_gcp_location
-    from linux_mcp_server.gatekeeper.gcp_auth import get_gcp_project
-
-    project = get_gcp_project()
-    location = get_gcp_location()
-    return f"https://aiplatform.googleapis.com/v1/projects/{project}/locations/{location}/endpoints/openapi"
-
-
-def vertex_auth_headers() -> dict[str, str]:
-    from linux_mcp_server.gatekeeper.gcp_auth import get_gcp_access_token
-
-    return {"Authorization": f"Bearer {get_gcp_access_token()}"}
-
-
-def openai_reasoning_block(reasoning_effort: ReasoningEffort | None) -> dict[str, Any] | None:
-    if reasoning_effort is None or reasoning_effort == ReasoningEffort.DEFAULT:
-        return None
-    return {"effort": reasoning_effort.value}
-
-
-def openrouter_reasoning_block(reasoning_effort: ReasoningEffort | None) -> dict[str, Any] | None:
-    if reasoning_effort is None or reasoning_effort == ReasoningEffort.DEFAULT:
-        return None
-    if reasoning_effort == ReasoningEffort.NONE:
-        return {"enabled": False}
-    return {"enabled": True, "effort": reasoning_effort.value}
-
-
-def anthropic_thinking_block(reasoning_effort: ReasoningEffort | None) -> dict[str, Any] | None:
-    if reasoning_effort is None or reasoning_effort in {ReasoningEffort.NONE, ReasoningEffort.DEFAULT}:
-        return None
-    budget_by_effort = {
-        ReasoningEffort.MINIMAL: 1024,
-        ReasoningEffort.LOW: 4096,
-        ReasoningEffort.MEDIUM: 8192,
-        ReasoningEffort.HIGH: 16384,
-        ReasoningEffort.XHIGH: 32768,
-    }
-    budget = budget_by_effort.get(reasoning_effort)
-    if budget is None:
-        return None
-    return {"type": "enabled", "budget_tokens": budget}
-
-
-def gemini_thinking_level(reasoning_effort: ReasoningEffort | None) -> str | None:
-    if reasoning_effort is None or reasoning_effort in {ReasoningEffort.NONE, ReasoningEffort.DEFAULT}:
-        return None
-    mapping = {
-        ReasoningEffort.MINIMAL: "MINIMAL",
-        ReasoningEffort.LOW: "LOW",
-        ReasoningEffort.MEDIUM: "MEDIUM",
-        ReasoningEffort.HIGH: "HIGH",
-        ReasoningEffort.XHIGH: "HIGH",
-    }
-    return mapping.get(reasoning_effort)
-
-
-def get_openai_api_key() -> str:
-    api_key = os.environ.get("OPENAI_API_KEY")
-    if not api_key:
-        raise ValueError("OPENAI_API_KEY is required for OpenAI gatekeeper provider.")
-    return api_key
-
-
-def get_openrouter_api_key() -> str:
-    api_key = os.environ.get("OPENROUTER_API_KEY")
-    if not api_key:
-        raise ValueError("OPENROUTER_API_KEY is required for OpenRouter gatekeeper provider.")
-    return api_key
-
-
-def openrouter_auth_headers() -> dict[str, str]:
-    return {"Authorization": f"Bearer {get_openrouter_api_key()}"}
-
-
-def get_anthropic_api_key() -> str:
-    api_key = os.environ.get("ANTHROPIC_API_KEY")
-    if not api_key:
-        raise ValueError("ANTHROPIC_API_KEY is required for Anthropic gatekeeper provider.")
-    return api_key
-
-
-def get_google_api_key() -> str:
-    api_key = os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY")
-    if not api_key:
-        raise ValueError("GOOGLE_API_KEY or GEMINI_API_KEY is required for Gemini direct backend.")
-    return api_key
-
-
-def openai_auth_headers() -> dict[str, str]:
-    return {"Authorization": f"Bearer {get_openai_api_key()}"}

src/linux_mcp_server/gatekeeper/llm.py

@@ -2,27 +2,19 @@
 
 import logging
 
-from pydantic import BaseModel
-
 from linux_mcp_server.config import CONFIG
 from linux_mcp_server.config import GatekeeperProvider
 from linux_mcp_server.gatekeeper.anthropic_client import complete_anthropic
 from linux_mcp_server.gatekeeper.gemini_client import complete_gemini
 from linux_mcp_server.gatekeeper.openai_client import complete_openai
 from linux_mcp_server.gatekeeper.openrouter_client import complete_openrouter
 from linux_mcp_server.gatekeeper.vertex_ai_client import complete_vertex_ai
+from linux_mcp_server.models import GatekeeperCompletion
 
 
 logger = logging.getLogger("linux-mcp-server")
 
 
-class GatekeeperCompletion(BaseModel):
-    text: str
-    prompt_tokens: int = 0
-    completion_tokens: int = 0
-    usage_cost: float | None = None
-
-
 def _infer_provider_from_model(model: str) -> GatekeeperProvider:
     if model.startswith("openrouter/"):
         return GatekeeperProvider.OPENROUTER