feat(mcp): graft v2 resolver onto _create_mcp_client (none + api_key static family)

basedpyright-code-budget.json

@@ -69,7 +69,7 @@
   },
   "reportMatchNotExhaustive": {
     "baseline": 1,
-    "slack": 3
+    "slack": 0
   },
   "reportMissingParameterType": {
     "baseline": 3933,

litellm/experimental_mcp_client/client.py

@@ -224,6 +224,7 @@ def __init__(
         extra_headers: Optional[Dict[str, str]] = None,
         ssl_verify: Optional[VerifyTypes] = None,
         aws_auth: Optional[httpx.Auth] = None,
+        resolved_auth: Optional[httpx.Auth] = None,
         sampling_callback: Optional[Callable] = None,
         elicitation_callback: Optional[Callable] = None,
         logging_callback: Optional[Callable] = None,
@@ -237,6 +238,9 @@ def __init__(
         self.extra_headers: Optional[Dict[str, str]] = extra_headers
         self.ssl_verify: Optional[VerifyTypes] = ssl_verify
         self._aws_auth: Optional[httpx.Auth] = aws_auth
+        # A pre-resolved httpx.Auth (e.g. from the v2 credential resolver) attached to the
+        # upstream client's auth= slot, taking precedence over the SigV4 aws_auth.
+        self._resolved_auth: Optional[httpx.Auth] = resolved_auth
         self._last_initialize_instructions: Optional[str] = None
         self._sampling_callback: Optional[Callable] = sampling_callback
         self._elicitation_callback: Optional[Callable] = elicitation_callback
@@ -482,11 +486,15 @@ def factory(
             verbose_logger.debug(
                 f"MCP client using SSL configuration: {type(ssl_config).__name__}"
             )
-            # Use SigV4 auth if configured and no explicit auth provided.
-            # The MCP SDK's sse_client and streamable_http_client call this
-            # factory without passing auth=, so self._aws_auth is used.
-            # For non-SigV4 clients, self._aws_auth is None — no behavior change.
-            effective_auth = auth if auth is not None else self._aws_auth
+            # The MCP SDK's sse_client and streamable_http_client call this factory without
+            # passing auth=, so the fallback is used: a v2-resolved auth if present, else the
+            # SigV4 aws_auth. Both are None for the common case — no behavior change.
+            fallback_auth = (
+                self._resolved_auth
+                if self._resolved_auth is not None
+                else self._aws_auth
+            )
+            effective_auth = auth if auth is not None else fallback_auth
             return httpx.AsyncClient(
                 headers=headers,
                 timeout=timeout,

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -56,6 +56,16 @@
     MCP_SAMPLING_AVAILABLE,
 )
 from litellm.proxy._experimental.mcp_server.oauth2_token_cache import resolve_mcp_auth
+from litellm.proxy._experimental.mcp_server.outbound_credentials import (
+    Error,
+    Ok,
+    UpstreamCredentialProvider,
+)
+from litellm.proxy._experimental.mcp_server.outbound_credentials.adapter import (
+    raise_public,
+    to_server_spec,
+    to_subject,
+)
 from litellm.proxy._experimental.mcp_server.utils import (
     MCP_TOOL_PREFIX_SEPARATOR,
     MCPMissingUserEnvVarsError,
@@ -511,7 +521,8 @@ def _resolve_oauth2_flow(
             return "client_credentials"
         return None
 
-    def __init__(self):
+    def __init__(self, cred_provider: Optional[UpstreamCredentialProvider] = None):
+        self._cred_provider = cred_provider or UpstreamCredentialProvider()
         self.registry: Dict[str, MCPServer] = {}
         self.config_mcp_servers: Dict[str, MCPServer] = {}
         """
@@ -1942,11 +1953,19 @@ async def _create_mcp_client(
         Returns:
             Configured MCP client instance.
         """
-        auth_value = await resolve_mcp_auth(
-            server, mcp_auth_header, subject_token=subject_token
-        )
-
         transport = server.transport or MCPTransport.sse
+        spec = None if transport == MCPTransport.stdio else to_server_spec(server)
+        # A per-request override is the caller-supplied credential v1 turns into the upstream
+        # auth, so it must win; defer those to v1 (this defer falls away once the per-user modes
+        # stop writing mcp_auth_header). An inbound header already in extra_headers is handled on
+        # the v2 path below, not here.
+        if spec is not None and mcp_auth_header:
+            spec = None
+        auth_value = (
+            await resolve_mcp_auth(server, mcp_auth_header, subject_token=subject_token)
+            if spec is None
+            else None
+        )
 
         # Create sampling and elicitation callbacks for this client
         sampling_cb = (
@@ -2017,6 +2036,43 @@ async def _create_mcp_client(
             # For HTTP/SSE transports
             server_url = server.url or ""
 
+            if spec is not None:
+                match await self._cred_provider.resolve_credentials(
+                    to_subject(user_api_key_auth, subject_token), spec
+                ):
+                    case Ok(auth):
+                        resolved_auth = auth
+                        # Do not override an Authorization already supplied via extra_headers
+                        # (a guardrail hook such as the JWT signer, static_headers, or a
+                        # forwarded caller header): v1 applies those last, so they win. NoOpAuth
+                        # has no header_name and so never skips.
+                        header_name = getattr(resolved_auth, "header_name", None)
+                        if (
+                            header_name
+                            and extra_headers
+                            and any(
+                                key.lower() == header_name.lower()
+                                for key in extra_headers
+                            )
+                        ):
+                            resolved_auth = None
+                    case Error(err):
+                        raise_public(err)
+                return MCPClient(
+                    server_url=server_url,
+                    transport_type=transport,
+                    auth_type=server.auth_type,
+                    timeout=(
+                        server.timeout
+                        if server.timeout is not None
+                        else MCP_CLIENT_TIMEOUT
+                    ),
+                    extra_headers=extra_headers,
+                    resolved_auth=resolved_auth,
+                    sampling_callback=sampling_cb,
+                    elicitation_callback=elicitation_cb,
+                )
+
             # Create SigV4 auth if configured
             aws_auth = None
             if server.auth_type == MCPAuth.aws_sigv4:

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

@@ -0,0 +1,140 @@
+"""The v1 <-> v2 bridge for the credential resolver.
+
+These edge functions translate v1's request objects into the resolver's typed inputs and map
+its typed errors onto the proxy's public exception contract. They import v1 and live outside the
+package's public surface so the resolver core (``resolver.py`` / ``types.py``) stays v1-free.
+Nothing wires them into ``_create_mcp_client`` yet.
+
+``to_server_spec`` maps only the modes the resolver has gone live for, returning ``None`` for
+every other mode so the caller defers to v1 (parity-safe); it grows one branch per migrated mode.
+"""
+
+from __future__ import annotations
+
+import base64
+from typing import TYPE_CHECKING, NoReturn, Optional
+
+from fastapi import HTTPException
+from pydantic import SecretStr
+from typing_extensions import assert_never
+
+from litellm.proxy._experimental.mcp_server.outbound_credentials.types import (
+    ApiKeyConfig,
+    CredError,
+    NoneConfig,
+    ServerSpec,
+    SharedKey,
+    Subject,
+)
+from litellm.types.mcp import MCPAuth
+
+if TYPE_CHECKING:
+    from litellm.proxy._types import UserAPIKeyAuth
+    from litellm.types.mcp_server.mcp_server_manager import MCPServer
+
+
+def to_subject(
+    user_api_key_auth: Optional[UserAPIKeyAuth], subject_token: Optional[str]
+) -> Subject:
+    """Map v1's authenticated principal onto the resolver's Subject.
+
+    tenant_id / subject_id are empty for an unauthenticated caller; the per-user arms must reject
+    an empty subject rather than share one credential slot across callers.
+    """
+    inbound = SecretStr(subject_token) if subject_token else None
+    if user_api_key_auth is None:
+        return Subject(tenant_id="", subject_id="", inbound_token=inbound)
+    return Subject(
+        tenant_id=user_api_key_auth.org_id or user_api_key_auth.team_id or "",
+        subject_id=user_api_key_auth.user_id or "",
+        inbound_token=inbound,
+    )
+
+
+def to_server_spec(server: MCPServer) -> Optional[ServerSpec]:
+    """Map a v1 server onto a ServerSpec for a migrated mode, or None to defer to v1.
+
+    BYOK is the per-user source of the ``api_key`` mode; its scheme rides on ``auth_type`` just
+    like a shared key, but the value is per-user and not migrated yet, so a BYOK server defers
+    to v1 regardless of ``auth_type`` (this guard is the seam the BYOK arm replaces later).
+
+    Dispatches on the declared ``auth_type``. The match is exhaustive over ``MCPAuthType`` with
+    an ``assert_never`` tail, so a newly added auth mode fails the type gate here until it is
+    explicitly mapped or explicitly deferred, rather than silently falling through to v1. Live
+    modes: ``none`` and the static-header family (``api_key`` plus the Authorization schemes),
+    all shared-key; every other mode returns None and stays on v1.
+    """
+    if server.is_byok:
+        return (
+            None  # per-user BYOK source not migrated yet -> defer to v1 (any auth_type)
+        )
+    resource = server.url or server.server_id
+    auth_type = server.auth_type
+    match auth_type:
+        case None | MCPAuth.none:
+            if server.is_oauth_passthrough:
+                return None  # passthrough is not migrated yet -> defer to v1
+            return ServerSpec(
+                server_id=server.server_id, resource=resource, config=NoneConfig()
+            )
+        case MCPAuth.api_key:
+            return _shared_key_spec(server, resource, "X-API-Key", "")
+        case MCPAuth.bearer_token:
+            return _shared_key_spec(server, resource, "Authorization", "Bearer")
+        case MCPAuth.token:
+            return _shared_key_spec(server, resource, "Authorization", "token")
+        case MCPAuth.authorization:
+            return _shared_key_spec(server, resource, "Authorization", "")
+        case MCPAuth.basic:
+            return _shared_key_spec(
+                server, resource, "Authorization", "Basic", encode=True
+            )
+        case MCPAuth.oauth2 | MCPAuth.oauth2_token_exchange | MCPAuth.aws_sigv4:
+            return None  # OAuth grants and SigV4 are not migrated yet -> defer to v1
+    assert_never(auth_type)
+
+
+def _shared_key_spec(
+    server: MCPServer,
+    resource: str,
+    header_name: str,
+    value_prefix: str,
+    *,
+    encode: bool = False,
+) -> Optional[ServerSpec]:
+    """Build an api_key spec from the server's static token, or defer (None) if it is absent.
+
+    Covers the whole shared-key static-header family: ``api_key`` on ``X-API-Key`` and the
+    Authorization schemes (bearer / token / authorization sent verbatim, basic base64-encoded).
+    """
+    token = server.authentication_token
+    if not token:
+        return None  # no key configured -> defer to v1 (parity-safe)
+    value = base64.b64encode(token.encode("utf-8")).decode() if encode else token
+    return ServerSpec(
+        server_id=server.server_id,
+        resource=resource,
+        config=ApiKeyConfig(
+            header_name=header_name,
+            value_prefix=value_prefix,
+            key_source=SharedKey(value=SecretStr(value)),
+        ),
+    )
+
+
+def raise_public(error: CredError) -> NoReturn:
+    """Map a resolver CredError onto the proxy's public HTTP contract. The one edge that raises."""
+    match error.tag:
+        case "unauthorized":
+            raise HTTPException(status_code=401, detail=error.summary)
+        case "misconfigured":
+            raise HTTPException(status_code=500, detail=error.summary)
+        case "upstream_unavailable":
+            raise HTTPException(status_code=503, detail=error.summary)
+        case "unsupported_mode":
+            raise HTTPException(status_code=500, detail=error.summary)
+        case "precondition_required":
+            raise HTTPException(status_code=412, detail=error.summary)
+        case "not_implemented":
+            raise HTTPException(status_code=501, detail=error.summary)
+    assert_never(error.tag)

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

@@ -7,30 +7,37 @@
 an arm fails the type gate (basedpyright `reportMatchNotExhaustive`); a bypassed gate fails loudly
 at runtime instead of returning `None`.
 
-This skeleton ships every arm as a `not_implemented` stub. Each mode's real body, with its
-injected seam, lands in its own follow-up PR; until then the arm returns a typed error rather
-than silently producing no credential. Pure v2: no imports from v1.
+`none` and `api_key` (shared-key source) are live; the remaining arms are `not_implemented`
+stubs that each land in a follow-up PR with their injected seam. The self-contained arms read
+straight from the config and need no collaborator. Pure v2: no imports from v1.
 """
 
 from __future__ import annotations
 
 import httpx
 from typing_extensions import assert_never
 
+from litellm.proxy._experimental.mcp_server.outbound_credentials.httpx_auth import (
+    NoOpAuth,
+    StaticHeaderAuth,
+)
 from litellm.proxy._experimental.mcp_server.outbound_credentials.result import (
     Error,
+    Ok,
     Result,
 )
 from litellm.proxy._experimental.mcp_server.outbound_credentials.types import (
     ApiKeyConfig,
     AuthorizationCodeConfig,
     AuthSpecKind,
     AwsSigV4Config,
+    Byok,
     ClientCredentialsConfig,
     CredError,
     NoneConfig,
     PassthroughConfig,
     ServerSpec,
+    SharedKey,
     Subject,
     TokenExchangeConfig,
 )
@@ -40,17 +47,17 @@ class UpstreamCredentialProvider:
     """Produces the one `httpx.Auth` for a `(subject, upstream)` pair, per declared mode.
 
     Collaborators (the per-mode credential stores and token fetchers) are injected as each arm
-    is built; the skeleton needs none, since every arm is a stub.
+    is built; the live `none` and `api_key`-shared arms read from the config and need none.
     """
 
     async def resolve_credentials(
         self, subject: Subject, server: ServerSpec
     ) -> Result[httpx.Auth, CredError]:
         match server.config:
             case NoneConfig():
-                return _not_implemented(AuthSpecKind.none)
-            case ApiKeyConfig():
-                return _not_implemented(AuthSpecKind.api_key)
+                return Ok(NoOpAuth())
+            case ApiKeyConfig() as config:
+                return self._api_key(config)
             case PassthroughConfig():
                 return _not_implemented(AuthSpecKind.passthrough)
             case ClientCredentialsConfig():
@@ -63,6 +70,22 @@ async def resolve_credentials(
                 return _not_implemented(AuthSpecKind.aws_sigv4)
         assert_never(server.config)
 
+    def _api_key(self, config: ApiKeyConfig) -> Result[httpx.Auth, CredError]:
+        match config.key_source:
+            case SharedKey() as source:
+                header_name, header_value = config.header(
+                    source.value.get_secret_value()
+                )
+                return Ok(StaticHeaderAuth(header_value, header_name=header_name))
+            case Byok():
+                # Per-user key pulled from the credential store; lands with that seam.
+                return Error(
+                    CredError.of_not_implemented(
+                        "api_key BYOK source not implemented yet"
+                    )
+                )
+        assert_never(config.key_source)
+
 
 def _not_implemented(kind: AuthSpecKind) -> Result[httpx.Auth, CredError]:
     return Error(