fix(mcp): reject scheme-only API key authorization payloads

litellm/proxy/_experimental/mcp_server/outbound_credentials/adapter.py

@@ -401,6 +401,11 @@ def _usable_credential_value(auth_type: MCPAuthType, name: str, value: str) -> b
         return True
     if value.lower() in ("bearer", "basic", "token", "apikey"):
         return False
+    if auth_type == MCPAuth.api_key:
+        api_scheme: Final = value.split(None, 1)[0]
+        if api_scheme.lower() in ("bearer", "token", "apikey"):
+            api_credential: Final = strip_auth_scheme(value, api_scheme).strip()
+            return api_credential.lower() != api_scheme.lower()
     if auth_type in (MCPAuth.bearer_token, MCPAuth.token):
         scheme: Final = "Bearer" if auth_type == MCPAuth.bearer_token else "token"
         credential: Final = strip_auth_scheme(value, scheme).strip()

tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_adapter.py

@@ -7,6 +7,7 @@ maps each CredError onto its HTTP status. These pin the parity-critical mapping
 
 import base64
 from types import SimpleNamespace
+from typing import Final
 
 import pytest
 from fastapi import HTTPException
@@ -20,7 +21,9 @@ from litellm.proxy._experimental.mcp_server.outbound_credentials.adapter import
     raise_user_oauth_challenge,
     to_server_spec,
     to_subject,
+    validate_static_credential,
 )
+from litellm.proxy._experimental.mcp_server.outbound_credentials.result import Ok
 from litellm.proxy._experimental.mcp_server.outbound_credentials.types import (
     ApiKeyConfig,
     AuthorizationCodeConfig,
@@ -34,10 +37,28 @@ from litellm.proxy._experimental.mcp_server.outbound_credentials.types import (
     SharedKey,
     TokenExchangeConfig,
 )
-from litellm.types.mcp import MCPAuth, MCPTransport
+from litellm.types.mcp import MCPAuth, MCPAuthType, MCPTransport
 from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
 
+@pytest.mark.parametrize("auth_type,header,value", [
+    (MCPAuth.api_key, "Authorization", "Bearer fixture-key"),
+    (MCPAuth.api_key, "Authorization", "ApiKey fixture-key"),
+    (MCPAuth.api_key, "Authorization", "token fixture-key"),
+    (MCPAuth.api_key, "Authorization", "Bearer token"),
+    (MCPAuth.api_key, "Authorization", "opaque-key"),
+    (MCPAuth.api_key, "Authorization", "Custom Custom"),
+    (MCPAuth.api_key, "X-API-Key", "Bearer Bearer"),
+    (MCPAuth.api_key, "X-Custom", "ApiKey ApiKey"),
+    (MCPAuth.authorization, "Authorization", "Bearer Bearer"),
+])
+def test_static_credential_preserves_supported_api_key_and_raw_headers(
+    auth_type: MCPAuthType, header: str, value: str,
+) -> None:
+    result: Final = validate_static_credential(auth_type, {header: value}, upstream_token_header=header)
+    assert isinstance(result, Ok)
+
+
 def _server(**kwargs) -> MCPServer:
     return MCPServer(server_id="s", name="n", transport=MCPTransport.http, **kwargs)
 

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py

@@ -13767,7 +13767,10 @@ class TestProtectedCredentialPreparation:
         assert custom_slot is None or custom_slot not in request.headers
 
     @pytest.mark.asyncio
-    @pytest.mark.parametrize("value", ["", " ", "Bearer", "Basic", "token", "ApiKey"])
+    @pytest.mark.parametrize("value", [
+        "", " ", "Bearer", "Basic", "token", "ApiKey",
+        "Bearer Bearer", "ApiKey ApiKey", "token token", "bEaReR   BEARER", "aPiKeY\tAPIKEY",
+    ])
     async def test_api_key_rejects_authorization_without_a_credential(self, value: str) -> None:
         server: Final = MCPServer(
             server_id="caller-empty", name="caller-empty", url="https://upstream.example/mcp",

tests/test_litellm/proxy/_experimental/mcp_server/test_openapi_to_mcp_generator.py

@@ -40,6 +40,35 @@ from litellm.proxy._experimental.mcp_server.exceptions import (
 GET_ASYNC_CLIENT_TARGET = "litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator.get_async_httpx_client"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("value,accepted", [
+    ("Bearer Bearer", False), ("ApiKey ApiKey", False), ("token token", False),
+    ("bEaReR   BEARER", False), ("aPiKeY\tAPIKEY", False),
+    ("Bearer fixture-key", True), ("ApiKey fixture-key", True), ("token fixture-key", True),
+])
+async def test_api_key_authorization_validates_payload_before_http(
+    respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch, value: str, accepted: bool,
+) -> None:
+    monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
+    tool: Final = create_tool_function(
+        "/echo", "get", {}, "https://upstream.example", auth_type=MCPAuth.api_key,
+    )
+    destination: Final = respx_mock.get("https://upstream.example/echo").respond(200, text="authenticated")
+    caller_token: Final = _request_auth_header.set(value)
+    try:
+        if accepted:
+            assert await tool() == "authenticated"
+            assert destination.call_count == 1
+            assert destination.calls.last.request.headers["authorization"] == value
+        else:
+            with pytest.raises(HTTPException, match="requires a usable upstream credential") as exc:
+                await tool()
+            assert exc.value.status_code == 500
+            assert destination.call_count == 0
+    finally:
+        _request_auth_header.reset(caller_token)
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("static,forwarded,caller,resolved,expected", [
     ({"Authorization": "Bearer configured"}, {"authorization": "Bearer forwarded"}, None, None, "Bearer configured"),