feat(mcp/v2): graft the remaining static Authorization modes (basic, authorization, token)

Fold basic, authorization, and token into to_server_spec alongside bearer_token: all four are the same
ApiKeyConfig on the Authorization header, differing only by scheme prefix (Bearer / token / verbatim /
Basic), so they share one block with a prefix lookup. basic base64-encodes user:pass to match v1's
to_basic_auth. They reuse the api_key resolve arm; each stops deferring and produces v1-identical headers.

Every MCPAuth value is now mapped. _should_defer's 'to_server_spec is None' condition now only fires for
misconfigured servers (e.g. a static mode with no token), not for any whole mode.

Validated: parametrized parity test (v2 headers byte-identical to v1 for all four schemes); 31 bridge tests pass.
This commit is contained in:
Tin Chi Lo 2026-06-20 17:56:15 -07:00
parent 6ccb5aab76
commit bed1d6a857
2 changed files with 47 additions and 16 deletions

View file

@ -12,6 +12,7 @@ This lives on the v1 side so the v2 core keeps its no-v1-imports invariant.
from __future__ import annotations
import base64
import functools
import os
from typing import TYPE_CHECKING, Dict, Optional
@ -146,17 +147,31 @@ def to_server_spec(server: MCPServer) -> Optional[ServerSpec]:
key_source=SharedKey(value=SecretStr(token)),
),
)
if server.auth_type == MCPAuth.bearer_token:
# Static credential on the Authorization header; these differ only by scheme prefix (basic is
# base64(user:pass)). All reuse the api_key arm. authorization uses an empty prefix (verbatim),
# so test the prefix with `is not None`, not truthiness.
authorization_prefix = {
MCPAuth.bearer_token: "Bearer",
MCPAuth.token: "token",
MCPAuth.authorization: "",
MCPAuth.basic: "Basic",
}.get(server.auth_type)
if authorization_prefix is not None:
token = server.authentication_token
if not token:
return None # bearer_token with no token: let v1 handle it (parity-safe)
return None # static Authorization mode with no token: let v1 handle it (parity-safe)
value = (
base64.b64encode(token.encode("utf-8")).decode()
if server.auth_type == MCPAuth.basic
else token
)
return ServerSpec(
server_id=server.server_id,
resource=resource,
config=ApiKeyConfig(
header_name="Authorization",
value_prefix="Bearer",
key_source=SharedKey(value=SecretStr(token)),
value_prefix=authorization_prefix,
key_source=SharedKey(value=SecretStr(value)),
),
)
if server.auth_type == MCPAuth.aws_sigv4:

View file

@ -5,6 +5,8 @@ headers must be byte-identical to what v1 produces, and every other mode (or any
fall back to v1 unchanged.
"""
import base64
import httpx
import pytest
@ -71,20 +73,34 @@ async def test_none_attaches_no_auth(v2_on):
assert _v1_headers(MCPAuth.none, None) == _v1_headers(MCPAuth.none, v2_value) == {}
async def test_non_grafted_mode_defers_to_v1(v2_on):
# basic is not grafted yet -> v2 returns None so v1 handles it
assert await resolve_v2_auth_value(_server(MCPAuth.basic, "k")) is None
async def test_bearer_token_parity(v2_on):
token = "up-secret"
server = _server(MCPAuth.bearer_token, token)
@pytest.mark.parametrize(
"auth_type,token,expected",
[
(MCPAuth.bearer_token, "up-secret", {"Authorization": "Bearer up-secret"}),
(MCPAuth.token, "up-secret", {"Authorization": "token up-secret"}),
(
MCPAuth.authorization,
"Custom raw-value",
{"Authorization": "Custom raw-value"},
),
(
MCPAuth.basic,
"user:pass",
{"Authorization": f"Basic {base64.b64encode(b'user:pass').decode()}"},
),
],
)
async def test_static_authorization_modes_parity(v2_on, auth_type, token, expected):
server = _server(auth_type, token)
v2_value = await resolve_v2_auth_value(server)
assert v2_value == {"Authorization": f"Bearer {token}"}
assert v2_value == expected
# byte-identical to v1's final upstream headers
assert _v1_headers(MCPAuth.bearer_token, token) == _v1_headers(
MCPAuth.bearer_token, v2_value
)
assert _v1_headers(auth_type, token) == _v1_headers(auth_type, v2_value) == expected
async def test_static_auth_without_token_defers_to_v1(v2_on):
# A static Authorization mode with no configured token defers to v1 (parity-safe).
assert await resolve_v2_auth_value(_server(MCPAuth.bearer_token, None)) is None
async def test_api_key_without_token_defers_to_v1(v2_on):