@@ -129,6 +129,50 @@ def extract_propagated_headers(
     return propagated
 
 
+def strip_bearer_prefix(value: str) -> str:
+    """Strip a leading ``Bearer`` scheme from an Authorization header value.
+
+    OGX's MCP tool ``authorization`` field expects a raw token/credential and
+    unconditionally prepends its own ``Bearer `` prefix before forwarding the
+    request to the downstream MCP server. Values built by this module (e.g.
+    the resolved Kubernetes token, or an ``Authorization`` header propagated
+    verbatim from the incoming request) already include the ``Bearer`` scheme
+    since that's the correct on-the-wire header representation. Passing such
+    a value straight through to OGX's ``authorization`` field would result in
+    a duplicated ``Bearer Bearer <token>`` header at the downstream MCP
+    server, so callers that populate that field must strip the scheme first.
+
+    Args:
+        value: The raw Authorization header value, with or without a
+            ``Bearer`` scheme.
+
+    Returns:
+        The value with a leading ``Bearer`` scheme (exact casing) and its
+        separating whitespace removed. Values without the scheme, including
+        those with a differently-cased scheme (e.g. ``bearer``), are
+        returned unchanged.
+    """
+    return value.removeprefix("Bearer ")
+
+
+def ensure_bearer_prefix(value: str) -> str:
+    """Ensure an Authorization header value carries the ``Bearer`` scheme.
+
+    Idempotent counterpart to :func:`strip_bearer_prefix`: a value that
+    already has the exact ``Bearer`` scheme is not double-prefixed, while a
+    raw token (or a value with a differently-cased scheme, e.g. ``bearer``,
+    which is treated as a raw token) gets the ``Bearer`` scheme added.
+
+    Args:
+        value: A raw token, or a value that may already start with the
+            exact ``Bearer`` scheme.
+
+    Returns:
+        The value guaranteed to start with a single ``Bearer `` prefix.
+    """
+    return f"Bearer {strip_bearer_prefix(value)}"
+
+
 def find_unresolved_auth_headers(
     configured: Mapping[str, str],
     resolved: Mapping[str, str],
 
@@ -15,7 +15,7 @@
 from configuration import AppConfig
 from log import get_logger
 from models.api.responses.error import UnauthorizedResponse
-from utils.mcp.mcp_headers import McpHeaders, build_mcp_headers
+from utils.mcp.mcp_headers import McpHeaders, build_mcp_headers, ensure_bearer_prefix
 
 logger = get_logger(__name__)
 
@@ -57,14 +57,9 @@ async def check_mcp_auth(
     for mcp_server in configuration.mcp_servers:
         headers = complete_headers.get(mcp_server.name, {})
         auth_header = headers.get("Authorization")
-        if auth_header is not None:
-            authorization = (
-                auth_header
-                if auth_header.startswith("Bearer ")
-                else f"Bearer {auth_header}"
-            )
-        else:
-            authorization = None
+        authorization = (
+            ensure_bearer_prefix(auth_header) if auth_header is not None else None
+        )
 
         if (
             authorization
 
@@ -11,6 +11,7 @@
 
 from log import get_logger
 from models.common.tools import ListedMcpTool
+from utils.mcp.mcp_headers import ensure_bearer_prefix
 
 logger = get_logger(__name__)
 
@@ -106,12 +107,8 @@ def _prepare_mcp_request_headers(headers: dict[str, str]) -> dict[str, str]:
     """
     prepared = dict(headers)
     for header_name, value in list(prepared.items()):
-        if (
-            header_name.lower() == "authorization"
-            and value
-            and not value.startswith("Bearer ")
-        ):
-            prepared[header_name] = f"Bearer {value}"
+        if header_name.lower() == "authorization" and value:
+            prepared[header_name] = ensure_bearer_prefix(value)
     return prepared
 
 
 
@@ -118,6 +118,7 @@
     McpHeaders,
     build_mcp_headers,
     find_unresolved_auth_headers,
+    strip_bearer_prefix,
 )
 from utils.model_list import parse_model_list_response
 from utils.otel_tracing import (
@@ -781,6 +782,13 @@ async def get_mcp_tools(
             continue
 
         authorization = headers.pop("Authorization", None)
+        if authorization is not None:
+            # OGX's "authorization" field expects a raw token/credential and
+            # prepends its own "Bearer " scheme downstream; strip any scheme
+            # already present (e.g. from Kubernetes auth or a propagated
+            # request header) to avoid a duplicated "Bearer Bearer <token>"
+            # header at the MCP server.
+            authorization = strip_bearer_prefix(authorization)
 
         require_approval = (
             mcp_server.require_approval
@@ -860,6 +868,10 @@ def apply_mcp_headers_to_explicit_tools(
             continue
 
         authorization = headers.pop("Authorization", None)
+        if authorization is not None:
+            # See the equivalent comment in get_mcp_tools: OGX prepends its
+            # own "Bearer " scheme, so strip any scheme already present here.
+            authorization = strip_bearer_prefix(authorization)
         out.append(
             mcp_tool.model_copy(
                 update={
 
@@ -11,8 +11,10 @@
 from utils.mcp import mcp_headers
 from utils.mcp.mcp_headers import (
     build_server_headers,
+    ensure_bearer_prefix,
     extract_propagated_headers,
     find_unresolved_auth_headers,
+    strip_bearer_prefix,
 )
 
 
@@ -335,6 +337,49 @@ def test_empty_resolved_returns_all_configured(self) -> None:
         assert sorted(result) == ["Authorization", "X-Api-Key"]
 
 
+class TestStripBearerPrefix:
+    """Test cases for strip_bearer_prefix function."""
+
+    def test_strips_bearer_prefix(self) -> None:
+        """Test that a leading 'Bearer ' scheme is removed."""
+        assert strip_bearer_prefix("Bearer abc123") == "abc123"
+
+    def test_does_not_strip_differently_cased_prefix(self) -> None:
+        """Test that only the exact 'Bearer ' scheme is matched, not other casings."""
+        assert strip_bearer_prefix("bearer abc123") == "bearer abc123"
+        assert strip_bearer_prefix("BEARER abc123") == "BEARER abc123"
+        assert strip_bearer_prefix("BeArEr abc123") == "BeArEr abc123"
+
+    def test_value_without_prefix_is_unchanged(self) -> None:
+        """Test that a raw token without a scheme is returned as-is."""
+        assert strip_bearer_prefix("abc123") == "abc123"
+
+    def test_empty_string_is_unchanged(self) -> None:
+        """Test that an empty string is returned unchanged."""
+        assert not strip_bearer_prefix("")
+
+    def test_only_strips_leading_prefix(self) -> None:
+        """Test that only the first 'Bearer ' scheme is stripped, not repeated ones."""
+        assert strip_bearer_prefix("Bearer Bearer abc123") == "Bearer abc123"
+
+
+class TestEnsureBearerPrefix:
+    """Test cases for ensure_bearer_prefix function."""
+
+    def test_adds_bearer_prefix_to_raw_token(self) -> None:
+        """Test that a raw token gets the 'Bearer ' scheme added."""
+        assert ensure_bearer_prefix("abc123") == "Bearer abc123"
+
+    def test_does_not_double_prefix_already_scoped_value(self) -> None:
+        """Test that a value already carrying 'Bearer ' is left unchanged."""
+        assert ensure_bearer_prefix("Bearer abc123") == "Bearer abc123"
+
+    def test_double_prefixes_differently_cased_scheme(self) -> None:
+        """Test that a differently-cased scheme is treated as a raw token and prefixed."""
+        assert ensure_bearer_prefix("bearer abc123") == "Bearer bearer abc123"
+        assert ensure_bearer_prefix("BEARER abc123") == "Bearer BEARER abc123"
+
+
 class TestBuildServerHeaders:
     """Test cases for build_server_headers function."""
 
 
@@ -485,7 +485,10 @@ async def test_get_mcp_tools_with_kubernetes_auth(
         mocker.patch("utils.responses.configuration", mock_config)
         tools_k8s = await get_mcp_tools(token="user-k8s-token")
         assert len(tools_k8s) == 1
-        assert tools_k8s[0].authorization == "Bearer user-k8s-token"
+        # The Bearer scheme is stripped here since OGX adds its own "Bearer "
+        # prefix before forwarding the token to the MCP server; passing the
+        # scheme through would result in a duplicated "Bearer Bearer <token>".
+        assert tools_k8s[0].authorization == "user-k8s-token"
 
     @pytest.mark.asyncio
     async def test_get_mcp_tools_with_mcp_headers(self, mocker: MockerFixture) -> None:
@@ -623,7 +626,7 @@ async def test_get_mcp_tools_with_mixed_headers(
 
         tools = await get_mcp_tools(token="k8s-token", mcp_headers=mcp_headers)
         assert len(tools) == 1
-        assert tools[0].authorization == "Bearer k8s-token"
+        assert tools[0].authorization == "k8s-token"
         assert tools[0].headers == {
             "X-API-Key": "secret-api-key",
             "X-Custom": "client-custom-value",
@@ -708,6 +711,40 @@ async def test_get_mcp_tools_with_propagated_headers(
             "x-request-id": "req-456",
         }
 
+    @pytest.mark.asyncio
+    async def test_get_mcp_tools_propagated_authorization_strips_bearer_scheme(
+        self, mocker: MockerFixture
+    ) -> None:
+        """Regression test for a duplicated 'Bearer Bearer <token>' MCP header.
+
+        When an MCP server allowlists the incoming ``Authorization`` header
+        via ``headers: [Authorization]`` (no ``authorization_headers``
+        config), the propagated value already includes the ``Bearer``
+        scheme. Since OGX's MCP ``authorization`` field expects a raw token
+        and adds its own ``Bearer `` prefix, the scheme must be stripped
+        before it is assigned to that field, or the downstream MCP server
+        receives ``Authorization: Bearer Bearer <token>``.
+        """
+        servers = [
+            ModelContextProtocolServer(
+                name="subscription-watch",
+                url="http://subscription-watch:8080",
+                headers=["Authorization"],
+                provider_id="provider",
+            ),
+        ]
+        mock_config = mocker.Mock()
+        mock_config.mcp_servers = servers
+        mocker.patch("utils.responses.configuration", mock_config)
+
+        request_headers = {"Authorization": "Bearer user-jwt-token"}
+        tools = await get_mcp_tools(
+            token=None, mcp_headers=None, request_headers=request_headers
+        )
+        assert len(tools) == 1
+        assert tools[0].authorization == "user-jwt-token"
+        assert tools[0].headers is None
+
     @pytest.mark.asyncio
     async def test_get_mcp_tools_propagated_headers_do_not_overwrite_auth_headers(
         self, tmp_path: Path, mocker: MockerFixture
@@ -815,7 +852,7 @@ async def test_get_mcp_tools_propagated_headers_additive_with_mcp_headers(
             token=None, mcp_headers=mcp_hdrs, request_headers=request_headers
         )
         assert len(tools) == 1
-        assert tools[0].authorization == "Bearer client-token"
+        assert tools[0].authorization == "client-token"
         assert tools[0].headers == {"x-rh-identity": "identity-value"}
 
     @pytest.mark.asyncio
@@ -934,6 +971,48 @@ async def test_apply_mcp_headers_preserves_type(
         dumped = out[0].model_dump(exclude_unset=True)
         assert dumped.get("type") == "mcp"
 
+    def test_apply_mcp_headers_strips_bearer_from_propagated_authorization(
+        self, mocker: MockerFixture
+    ) -> None:
+        """apply_mcp_headers_to_explicit_tools must strip the Bearer scheme.
+
+        Same regression as ``test_get_mcp_tools_propagated_authorization_strips_bearer_scheme``
+        but for the explicit-tools path: a propagated ``Authorization`` header
+        must not retain its ``Bearer`` scheme when assigned to OGX's
+        ``authorization`` field, or the MCP server receives a duplicated
+        ``Bearer Bearer <token>`` header.
+        """
+        from utils.responses import (  # pylint: disable=import-outside-toplevel
+            apply_mcp_headers_to_explicit_tools,
+        )
+
+        servers = [
+            ModelContextProtocolServer(
+                name="subscription-watch",
+                url="http://subscription-watch:8080",
+                headers=["Authorization"],
+                provider_id="mcp",
+            ),
+        ]
+        mock_config = mocker.Mock()
+        mock_config.mcp_servers = servers
+        mocker.patch("utils.responses.configuration", mock_config)
+
+        explicit = InputToolMCP(
+            server_label="subscription-watch",
+            server_url="http://subscription-watch:8080",
+        )
+
+        out = apply_mcp_headers_to_explicit_tools(
+            [explicit],
+            token=None,
+            mcp_headers=None,
+            request_headers={"Authorization": "Bearer user-jwt-token"},
+        )
+
+        assert len(out) == 1
+        assert out[0].authorization == "user-jwt-token"
+
 
 class TestGetTopicSummary:
     """Tests for get_topic_summary function."""