Adding noop-with-token to allow k8 authentication for MCP servers

src/app/endpoints/mcp_auth.py

@@ -5,6 +5,7 @@
 
 from fastapi import APIRouter, Depends, Request
 
+import constants
 from authentication import get_auth_dependency
 from authentication.interface import AuthTuple
 from authorization.middleware import authorize
@@ -77,7 +78,7 @@ async def get_mcp_client_auth_options(
         client_headers = [
             header_name
             for header_name, header_value in mcp_server.authorization_headers.items()
-            if header_value.strip() == "client"
+            if header_value.strip() == constants.MCP_AUTH_CLIENT
         ]
 
         if client_headers:

src/app/endpoints/query_v2.py

@@ -20,6 +20,7 @@
 )
 from llama_stack_client import AsyncLlamaStackClient
 
+import constants
 import metrics
 from app.endpoints.query import (
     query_endpoint_handler_base,
@@ -782,12 +783,12 @@ def get_mcp_tools(
     def _get_token_value(original: str, header: str) -> str | None:
         """Convert to header value."""
         match original:
-            case "kubernetes":
+            case constants.MCP_AUTH_KUBERNETES:
                 # use k8s token
                 if token is None or token == "":
                     return None
                 return f"Bearer {token}"
-            case "client":
+            case constants.MCP_AUTH_CLIENT:
                 # use client provided token
                 if mcp_headers is None:
                     return None

src/constants.py

@@ -122,6 +122,10 @@
 DEFAULT_JWT_UID_CLAIM = "user_id"
 DEFAULT_JWT_USER_NAME_CLAIM = "username"
 
+# MCP authorization header special values
+MCP_AUTH_KUBERNETES = "kubernetes"
+MCP_AUTH_CLIENT = "client"
+
 # default RAG tool value
 DEFAULT_RAG_TOOL = "knowledge_search"
 

src/models/config.py

@@ -1665,8 +1665,12 @@ def validate_mcp_auth_headers(self) -> Self:
         Validate MCP server authorization headers against authentication module.
 
         Removes any MCP server with authorization_headers="kubernetes" when the
-        authentication module is not "k8s". This prevents sending wrong credential
-        types to MCP servers.
+        authentication module is not "k8s" or "noop-with-token". This prevents sending
+        wrong credential types to MCP servers.
+
+        Note: "noop-with-token" should only be used for testing/development purposes.
+        When using "noop-with-token" with kubernetes authorization headers, a real
+        Kubernetes token must still be passed in the request headers.
 
         Returns:
             Self: The model instance after validation.
@@ -1680,15 +1684,28 @@ def validate_mcp_auth_headers(self) -> Self:
             is_valid = True
             if mcp_server.authorization_headers:
                 for value in mcp_server.authorization_headers.values():
-                    if value.strip() == "kubernetes" and auth_module != "k8s":
+                    if (
+                        value.strip() == constants.MCP_AUTH_KUBERNETES
+                        and auth_module
+                        not in [
+                            constants.AUTH_MOD_K8S,
+                            constants.AUTH_MOD_NOOP_WITH_TOKEN,
+                        ]
+                    ):
                         logger.warning(
                             "Removing MCP server '%s': has authorization_headers with "
-                            "value 'kubernetes' but authentication module is '%s' "
-                            "(not 'k8s'). Either change authentication.module to 'k8s' "
-                            "or update the MCP server's authorization_headers to use a "
-                            "file path or 'client'.",
+                            "value '%s' but authentication module is '%s' "
+                            "(not '%s' or '%s'). Either change authentication.module to "
+                            "'%s' or '%s' or update the MCP server's authorization_headers "
+                            "to use a file path or '%s'.",
                             mcp_server.name,
+                            constants.MCP_AUTH_KUBERNETES,
                             auth_module,
+                            constants.AUTH_MOD_K8S,
+                            constants.AUTH_MOD_NOOP_WITH_TOKEN,
+                            constants.AUTH_MOD_K8S,
+                            constants.AUTH_MOD_NOOP_WITH_TOKEN,
+                            constants.MCP_AUTH_CLIENT,
                         )
                         is_valid = False
                         break

src/utils/mcp_auth_headers.py

@@ -3,6 +3,8 @@
 import logging
 from pathlib import Path
 
+import constants
+
 logger = logging.getLogger(__name__)
 
 
@@ -39,16 +41,16 @@ def resolve_authorization_headers(
     for header_name, value in authorization_headers.items():
         value = value.strip()
         try:
-            if value == "kubernetes":
+            if value == constants.MCP_AUTH_KUBERNETES:
                 # Special case: Keep kubernetes keyword for later substitution
-                resolved[header_name] = "kubernetes"
+                resolved[header_name] = constants.MCP_AUTH_KUBERNETES
                 logger.debug(
                     "Header %s will use Kubernetes token (resolved at request time)",
                     header_name,
                 )
-            elif value == "client":
+            elif value == constants.MCP_AUTH_CLIENT:
                 # Special case: Keep client keyword for later substitution
-                resolved[header_name] = "client"
+                resolved[header_name] = constants.MCP_AUTH_CLIENT
                 logger.debug(
                     "Header %s will use client-provided token (resolved at request time)",
                     header_name,

tests/e2e/mock_jwks_server/server.py

@@ -7,6 +7,7 @@
 
 import json
 from http.server import HTTPServer, BaseHTTPRequestHandler
+from typing import Any
 
 # Static JWKS - pre-generated RSA public key
 JWKS = {
@@ -55,7 +56,7 @@ def _json_response(self, data: dict) -> None:
         self.end_headers()
         self.wfile.write(body)
 
-    def log_message(self, format: str, *args) -> None:
+    def log_message(self, format: str, *args: Any) -> None:
         """Suppress request logging."""