scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Merge pull request #31989 from BerriAI/litellm_mcp_passthrough_delegate_modes

Browse source 
feat(mcp): add true_passthrough and oauth_delegate auth modes
This commit is contained in:
tin-berri 2026-07-08 17:16:37 -07:00 • committed by GitHub
commit 4e6ec995e7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 1804 additions and 888 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

64
litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py
View file

@ -220,6 +220,12 @@ class MCPRequestHandler:
# when EVERY target is auth_type=oauth2 with delegate_auth_to_upstream
# set; fails closed otherwise.
validated_user_api_key_auth = UserAPIKeyAuth()
elif MCPRequestHandler._target_servers_are_true_passthrough(
path=request_route,
mcp_servers=mcp_servers,
client_ip=IPAddressUtils.get_mcp_client_ip(request),
):
validated_user_api_key_auth = UserAPIKeyAuth()
elif oauth2_headers:
# Authorization on a non-delegated server: the bearer must be a real
# LiteLLM credential, so a failed validation is a genuine 401/403 and
@ -399,6 +405,33 @@ class MCPRequestHandler:
return False
return True
@staticmethod
def _target_servers_are_true_passthrough(
path: str, mcp_servers: Optional[list[str]], client_ip: Optional[str]
) -> bool:
"""
True only when EVERY MCP server the request targets is ``auth_type == true_passthrough``.
Fails closed when any target does not opt in or cannot be resolved.
Used by :meth:`process_mcp_request` to skip LiteLLM admission auth entirely: the gateway is a
transparent proxy and the caller's ``Authorization`` is an upstream token, never a LiteLLM key.
Mirrors :meth:`_target_servers_delegate_auth_to_upstream`; a mixed-target request keeps normal auth.
"""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
from litellm.types.mcp import MCPAuth
target_names = MCPRequestHandler._resolve_target_server_names(path=path, mcp_servers_header=mcp_servers)
if not target_names:
return False
for name in target_names:
server = global_mcp_server_manager.get_mcp_server_by_name(name, client_ip=client_ip)
if server is None or server.auth_type != MCPAuth.true_passthrough:
return False
return True
@staticmethod
def _resolve_target_server_names(path: str, mcp_servers_header: Optional[List[str]]) -> List[str]:
"""
@ -558,10 +591,19 @@ class MCPRequestHandler:
ASGI headers are in format: List[List[bytes, bytes]]
We need to convert them to the format Headers expects.
Collapsing the ASGI list into a dict keeps the last value for a duplicated
header name, so a request carrying more than one ``Authorization`` is
rejected first: for the client-forwarded token modes the gateway relays the
caller's ``Authorization`` upstream, so a duplicate would make which token is
forwarded ambiguous (and diverge from what admission inspected). Multiple
``Authorization`` headers is malformed for bearer auth anyway (RFC 9110: not
a comma-combinable field), so fail closed with a 400.
"""
raw_headers = scope.get("headers", [])
MCPRequestHandler._reject_duplicate_authorization(raw_headers)
try:
# ASGI headers are list of [name: bytes, value: bytes] pairs
raw_headers = scope.get("headers", [])
# Convert bytes to strings and create dict for Headers constructor
headers_dict = {name.decode("latin-1"): value.decode("latin-1") for name, value in raw_headers}
return Headers(headers_dict)
@ -570,6 +612,26 @@ class MCPRequestHandler:
# Return empty Headers object with empty dict
return Headers({})
@staticmethod
def _reject_duplicate_authorization(raw_headers: object) -> None:
"""Raise 400 when the raw ASGI headers carry more than one ``Authorization`` header."""
if not isinstance(raw_headers, (list, tuple)):
return
count = 0
for entry in raw_headers:
if not isinstance(entry, (list, tuple)) or len(entry) < 1:
continue
name = entry[0]
if isinstance(name, (bytes, bytearray)) and bytes(name).lower() == b"authorization":
count += 1
elif isinstance(name, str) and name.lower() == "authorization":
count += 1
if count > 1:
raise HTTPException(
status_code=400,
detail="Multiple Authorization headers are not allowed",
)
@staticmethod
async def get_allowed_mcp_servers(
user_api_key_auth: Optional[UserAPIKeyAuth] = None,

23
litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py
View file

@ -1302,11 +1302,15 @@ async def _build_oauth_protected_resource_response(
"""
Build OAuth protected resource response with the appropriate URL pattern.
For pass-through MCP servers (``MCPServer.is_oauth_passthrough``), the
gateway proxies the upstream's own ``oauth-protected-resource`` metadata
so that standards-compliant MCP clients discover the **upstream** IdP
instead of the gateway. The ``resource`` field is rewritten to the
gateway's own URL so clients present the bearer token back to the gateway.
For pass-through MCP servers, the gateway proxies the upstream's own
``oauth-protected-resource`` metadata so standards-compliant MCP clients
discover the **upstream** IdP instead of the gateway. For ``true_passthrough``
and ``oauth_delegate`` the metadata is returned verbatim (``resource`` stays
the upstream): the caller's token is forwarded to and validated by the
upstream, so its audience must be the upstream — rewriting it to the gateway
would make a strict IdP (e.g. Entra) refuse to mint it or the upstream reject
it. Only the legacy ``is_oauth_passthrough`` opt-in rewrites ``resource`` to
the gateway's own URL so clients present the bearer token back to the gateway.
Args:
request: FastAPI Request object
@ -1347,7 +1351,9 @@ async def _build_oauth_protected_resource_response(
# Pass-through branch: proxy the upstream's own metadata so discovery
# directs the client at the real IdP (Okta, Keycloak, …) instead of us.
if mcp_server is not None and mcp_server.is_oauth_passthrough:
if mcp_server is not None and (
mcp_server.is_oauth_passthrough or mcp_server.is_oauth_delegate or mcp_server.is_true_passthrough
):
try:
upstream_metadata = await fetch_upstream_oauth_protected_resource(mcp_server)
except Exception as exc:
@ -1363,8 +1369,9 @@ async def _build_oauth_protected_resource_response(
)
if upstream_metadata is not None:
response = {**upstream_metadata, "resource": resource_url}
return response
if mcp_server.is_true_passthrough or mcp_server.is_oauth_delegate:
return upstream_metadata
return {**upstream_metadata, "resource": resource_url}
# Upstream responded but with non-200 or non-dict payload. For
# pass-through servers the gateway is NOT the authorization server,

146
litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
View file

@ -81,6 +81,7 @@ from litellm.proxy._experimental.mcp_server.outbound_credentials.token_exchange_
)
from litellm.proxy._experimental.mcp_server.outbound_credentials.types import (
AuthorizationCodeConfig,
PassthroughConfig,
ServerSpec,
TokenExchangeConfig,
)
@ -216,6 +217,13 @@ def _should_strip_caller_authorization(
pass-through cold-start case (RFC 9728) the bearer in
``Authorization`` is the upstream OAuth token and must be
forwarded, so we keep it.
- **oauth_delegate servers**: admission always runs and there is no
anonymous path, so the caller's separate ``Authorization`` is
forwarded only when a distinct ``x-litellm-api-key`` carried
admission. Without that header the ``Authorization`` *was* the
admission credential — a virtual key, an IdP JWT, or an SSO / OIDC /
session token whose ``api_key`` is ``None`` — and must never reach
the upstream, so it is stripped regardless of the ``api_key`` value.
"""
if mcp_server.auth_type == MCPAuth.oauth2_token_exchange:
# OBO: the inbound Authorization is the subject token. It is exchanged at the IdP and only the
@ -229,11 +237,13 @@ def _should_strip_caller_authorization(
# upstream — it would override another user's stored credential. Delegate and
# pass-through return None from to_server_spec and keep forwarding the bearer.
return True
if not mcp_server.is_oauth_passthrough:
if not (mcp_server.is_oauth_passthrough or mcp_server.is_oauth_delegate):
return False
normalized_raw_headers = {str(k).lower(): v for k, v in (raw_headers or {}).items() if isinstance(k, str)}
has_explicit_litellm_admission_header = normalized_raw_headers.get("x-litellm-api-key") is not None
if mcp_server.is_oauth_delegate:
return not has_explicit_litellm_admission_header
admission_consumed_authorization_as_litellm_key = (
user_api_key_auth is not None
and bool(getattr(user_api_key_auth, "api_key", None))
@ -326,6 +336,89 @@ async def _resolve_byok_mcp_auth_header(
return mcp_auth_header
def _client_forwarded_authorization_headers(
mcp_server: MCPServer,
oauth2_headers: Optional[dict[str, str]],
raw_headers: Optional[dict[str, str]],
user_api_key_auth: Optional[UserAPIKeyAuth],
) -> Optional[dict[str, str]]:
"""Egress headers for the client-forwarded-token modes (``true_passthrough`` / ``oauth_delegate``).
Forwards the caller's ``Authorization`` to the upstream, stripped when
``_should_strip_caller_authorization`` says it was consumed as the LiteLLM admission key. Shared by
``_call_regular_mcp_tool`` and ``server.py``'s ``_prepare_mcp_server_headers`` so the two egress
paths cannot drift, mirroring the ``_should_strip_caller_authorization`` split.
"""
extra_headers = oauth2_headers.copy() if oauth2_headers else None
if extra_headers and _should_strip_caller_authorization(
mcp_server=mcp_server,
raw_headers=raw_headers,
user_api_key_auth=user_api_key_auth,
):
return _without_authorization(extra_headers)
return extra_headers
def _take_forwarded_authorization(
headers: Optional[dict[str, str]],
) -> tuple[Optional[str], Optional[dict[str, str]]]:
"""Pop the ``Authorization`` value out of ``headers`` (case-insensitive), returning it with the
remaining headers, so the passthrough resolver arm is the single Authorization source rather than
the header also riding in ``extra_headers`` (which the resolved auth would then defer to)."""
if not headers:
return None, headers
value = next((v for k, v in headers.items() if k.lower() == "authorization"), None)
return value, _without_authorization(headers)
def _passthrough_token_from_mcp_auth_header(
mcp_auth_header: Optional[Union[str, dict[str, str]]],
) -> Optional[str]:
"""The caller's per-server upstream credential for a passthrough-mode server, or None.
Sourced from ``x-mcp-{alias}-authorization`` (string or per-header dict form) or the deprecated
global ``x-mcp-auth`` fallback. Per-server headers are the multi-server shape: they bind one
token to one server, so an aggregate scope with several passthrough-mode servers never replays
a single credential across upstreams. The value is forwarded verbatim, so it must be the full
header value (e.g. ``Bearer <upstream-token>``)."""
if isinstance(mcp_auth_header, str):
return mcp_auth_header or None
if isinstance(mcp_auth_header, dict):
return next((v for k, v in mcp_auth_header.items() if k.lower() == "authorization"), None)
return None
def _consumes_caller_authorization(server: MCPServer) -> bool:
"""True when this server's egress forwards the caller's request-wide ``Authorization`` upstream:
the client-forwarded token modes, legacy OAuth pass-through, and legacy upstream-delegated
interactive oauth2. An unstamped oauth2 row (flow column not yet backfilled) reads as a consumer,
which errs toward suppression — the fail-safe direction."""
if server.is_true_passthrough or server.is_oauth_delegate or server.is_oauth_passthrough:
return True
return (
server.auth_type == MCPAuth.oauth2
and getattr(server, "delegate_auth_to_upstream", False) is True
and not server.has_client_credentials
)
def _caller_authorization_fans_out(
server: MCPServer,
scope_servers: Optional[list[MCPServer]],
) -> bool:
"""True when forwarding the caller's request-wide ``Authorization`` to ``server`` inside a
listing fan-out would replay one credential against multiple upstreams: another server in the
scope also consumes it (RFC 9700 cross-resource replay). ``scope_servers`` is None for
explicitly-addressed operations (tool call, get_prompt, read_resource, single-server routes),
where the client named the one target and the gateway is not choosing recipients."""
if scope_servers is None:
return False
return any(
other is not None and other.server_id != server.server_id and _consumes_caller_authorization(other)
for other in scope_servers
)
def _extract_upstream_auth_failure(
exc: BaseException,
) -> Optional[tuple[int, Optional[str]]]:
@ -1635,15 +1728,18 @@ class MCPServerManager:
delegate_server_ids = [
server.server_id
for server in self.get_registry().values()
if getattr(server, "auth_type", None) == MCPAuth.oauth2
and getattr(server, "delegate_auth_to_upstream", False) is True
# M2M servers must not be exposed anonymously: an
# unauthenticated caller would get LiteLLM to proxy tool
# calls using its stored client_credentials. Resolve the flow
# rather than reading has_client_credentials so an unstamped
# M2M-shape row (null column, verbatim-read as non-M2M) still
# fails closed here, matching the anonymous-delegate auth gate.
and MCPServerManager.effective_oauth2_flow(server) != "client_credentials"
if (
getattr(server, "auth_type", None) == MCPAuth.oauth2
and getattr(server, "delegate_auth_to_upstream", False) is True
# M2M servers must not be exposed anonymously: an
# unauthenticated caller would get LiteLLM to proxy tool
# calls using its stored client_credentials. Resolve the flow
# rather than reading has_client_credentials so an unstamped
# M2M-shape row (null column, verbatim-read as non-M2M) still
# fails closed here, matching the anonymous-delegate auth gate.
and MCPServerManager.effective_oauth2_flow(server) != "client_credentials"
)
or getattr(server, "auth_type", None) == MCPAuth.true_passthrough
]
combined_servers.update(delegate_server_ids)
@ -2241,16 +2337,17 @@ class MCPServerManager:
spec = None if transport == MCPTransport.stdio else to_server_spec(server)
provider = cred_provider or self._cred_provider
# A caller-supplied per-request override (mcp_auth_header / x-mcp-*) defers to the v1 path
# so it wins - except for the per-user modes the v2 resolver owns (authorization_code's
# stored token and token_exchange's RFC 8693 minted token). A caller must not be able to
# substitute another user's stored credential, nor silently disable the OBO exchange and
# forward an arbitrary bearer upstream, so we keep the v2 spec and ignore the override for
# both; the REST tools preview supplies its not-yet-persisted token through the resolver
# (cred_provider), never this path.
# so it wins - except for the modes the v2 resolver owns per-caller (authorization_code's
# stored token, token_exchange's RFC 8693 minted token, and the passthrough modes'
# forwarded caller token). A caller must not be able to substitute another user's stored
# credential, nor silently disable the OBO exchange and forward an arbitrary bearer
# upstream, so we keep the v2 spec and ignore the override for these; the REST tools
# preview supplies its not-yet-persisted token through the resolver (cred_provider),
# never this path.
if (
spec is not None
and mcp_auth_header
and not isinstance(spec.config, (AuthorizationCodeConfig, TokenExchangeConfig))
and not isinstance(spec.config, (AuthorizationCodeConfig, PassthroughConfig, TokenExchangeConfig))
):
spec = None
auth_value = (
@ -2317,11 +2414,17 @@ class MCPServerManager:
server_url = server.url or ""
if spec is not None:
inbound_token = subject_token
if isinstance(spec.config, PassthroughConfig):
inbound_token, extra_headers = _take_forwarded_authorization(extra_headers)
per_server_token = _passthrough_token_from_mcp_auth_header(mcp_auth_header)
if per_server_token is not None:
inbound_token = per_server_token
resolved_auth, extra_headers = await self._resolve_v2_auth(
server=server,
spec=spec,
provider=provider,
subject_token=subject_token,
subject_token=inbound_token,
user_api_key_auth=user_api_key_auth,
extra_headers=extra_headers,
)
@ -3743,6 +3846,13 @@ class MCPServerManager:
user_api_key_auth=user_api_key_auth,
):
extra_headers = _without_authorization(extra_headers)
elif mcp_server.is_true_passthrough or mcp_server.is_oauth_delegate:
extra_headers = _client_forwarded_authorization_headers(
mcp_server=mcp_server,
oauth2_headers=oauth2_headers,
raw_headers=raw_headers,
user_api_key_auth=user_api_key_auth,
)
if mcp_server.extra_headers and raw_headers:
if extra_headers is None:

10
litellm/proxy/_experimental/mcp_server/outbound_credentials/adapter.py
View file

@ -23,6 +23,7 @@ from litellm.proxy._experimental.mcp_server.outbound_credentials.types import (
AuthorizationCodeConfig,
CredError,
NoneConfig,
PassthroughConfig,
ServerSpec,
SharedKey,
Subject,
@ -62,9 +63,10 @@ def to_server_spec(server: MCPServer) -> Optional[ServerSpec]:
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``, the static-header family (``api_key`` plus the Authorization schemes,
all shared-key), ``oauth2`` per-user tokens (``authorization_code``), and
``oauth2_token_exchange`` (OBO); client_credentials (M2M), delegated/passthrough
oauth2, and SigV4 return None and stay on v1.
all shared-key), ``oauth2`` per-user tokens (``authorization_code``), ``oauth2_token_exchange``
(OBO), and the client-forwarded token modes ``true_passthrough`` / ``oauth_delegate``
(``PassthroughConfig``); client_credentials (M2M), delegated/passthrough oauth2, and SigV4
return None and stay on v1.
"""
if server.is_byok:
return None # per-user BYOK source not migrated yet -> defer to v1 (any auth_type)
@ -94,6 +96,8 @@ def to_server_spec(server: MCPServer) -> Optional[ServerSpec]:
)
# client_credentials (M2M) and delegate/passthrough oauth2 stay on v1
return None
case MCPAuth.true_passthrough | MCPAuth.oauth_delegate:
return ServerSpec(server_id=server.server_id, resource=resource, config=PassthroughConfig())
case MCPAuth.oauth2_token_exchange:
return _token_exchange_spec(server, resource)
case MCPAuth.aws_sigv4:

23
litellm/proxy/_experimental/mcp_server/outbound_credentials/resolver.py
View file

@ -7,10 +7,11 @@ no precedence cascade. It is wildcard-free with an `assert_never` tail, so addin
an arm fails the type gate (basedpyright `reportMatchNotExhaustive`); a bypassed gate fails loudly
at runtime instead of returning `None`.
`none` and `api_key` (shared-key source) are live, as is `authorization_code`, which reads the
user's token from the injected `OAuthTokenStore`, and `token_exchange`, which swaps the caller's
inbound token through the injected `TokenExchanger`. The remaining arms are `not_implemented` stubs
that each land in a follow-up PR with their seam. Pure v2: no imports from v1.
`none`, `api_key` (shared-key source), and `passthrough` (forwards the caller's own inbound token)
are live, as is `authorization_code`, which reads the user's token from the injected
`OAuthTokenStore`, and `token_exchange`, which swaps the caller's inbound token through the
injected `TokenExchanger`. The remaining arms are `not_implemented` stubs that each land in a
follow-up PR with their seam. Pure v2: no imports from v1.
"""
from __future__ import annotations
@ -97,7 +98,7 @@ class UpstreamCredentialProvider:
case ApiKeyConfig() as config:
return self._api_key(config)
case PassthroughConfig():
return _not_implemented(AuthSpecKind.passthrough)
return self._passthrough(subject)
case ClientCredentialsConfig():
return _not_implemented(AuthSpecKind.client_credentials)
case TokenExchangeConfig() as config:
@ -118,6 +119,18 @@ class UpstreamCredentialProvider:
"""
return await self._authz_token(subject, server) is not None
def _passthrough(self, subject: Subject) -> Result[httpx.Auth, CredError]:
"""Forward the caller's own upstream credential verbatim; the gateway mints nothing.
The inbound token is the caller's already-disambiguated ``Authorization`` (never the LiteLLM
admission credential; the edge adapter drops that before building the ``Subject``). When it is
absent the request is sent unauthenticated so the upstream's own 401 surfaces, rather than the
gateway challenging on the upstream's behalf.
"""
if subject.inbound_token is None:
return Ok(NoOpAuth())
return Ok(StaticHeaderAuth(subject.inbound_token.get_secret_value(), header_name="Authorization"))
def _api_key(self, config: ApiKeyConfig) -> Result[httpx.Auth, CredError]:
match config.key_source:
case SharedKey() as source:

125
litellm/proxy/_experimental/mcp_server/server.py
View file

@ -336,6 +336,8 @@ if MCP_AVAILABLE:
)
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
MCPServerManager,
_caller_authorization_fans_out,
_client_forwarded_authorization_headers,
_should_strip_caller_authorization,
_without_authorization,
global_mcp_server_manager,
@ -1444,6 +1446,35 @@ if MCP_AVAILABLE:
return allowed_mcp_servers
def _client_has_per_server_auth_header(
server: MCPServer,
mcp_server_auth_headers: Optional[Dict[str, Dict[str, str]]],
) -> bool:
"""True if the request carries a per-server ``x-mcp-{alias}-authorization``
header for this server. This is the multi-server binding: it names one
upstream, so it is unambiguously the caller's upstream token regardless of
auth mode (never the LiteLLM admission credential).
Resolves through the same ``lookup_mcp_server_auth_in_headers`` egress uses, so
the connect gate and egress agree on which per-server header names match: a
dashboard client sends ``x-mcp-{sanitize_mcp_alias_for_header(alias)}-authorization``,
and matching only the raw alias here would 401 a token egress would forward.
"""
if not mcp_server_auth_headers:
return False
from litellm.proxy._experimental.mcp_server.utils import (
lookup_mcp_server_auth_in_headers,
)
server_headers = lookup_mcp_server_auth_in_headers(
mcp_server_auth_headers, alias=server.alias, server_name=server.server_name
)
if isinstance(server_headers, str):
return bool(server_headers.strip())
if isinstance(server_headers, dict):
return any(isinstance(hk, str) and hk.lower() == "authorization" for hk in server_headers)
return False
def _client_has_passthrough_authorization(
server: MCPServer,
oauth2_headers: Optional[Dict[str, str]],
@ -1461,24 +1492,7 @@ if MCP_AVAILABLE:
for k in oauth2_headers.keys():
if k.lower() == "authorization":
return True
if mcp_server_auth_headers:
for key in (server.alias, server.server_name, server.name):
if not key:
continue
server_headers = None
for k, v in mcp_server_auth_headers.items():
if k.lower() == key.lower():
server_headers = v
break
if server_headers is None:
continue
if isinstance(server_headers, str) and server_headers.strip():
return True
if isinstance(server_headers, dict):
for hk in server_headers.keys():
if hk.lower() == "authorization":
return True
return False
return _client_has_per_server_auth_header(server, mcp_server_auth_headers)
async def _get_user_oauth_extra_headers_from_db(
server: MCPServer,
@ -1533,8 +1547,16 @@ if MCP_AVAILABLE:
oauth2_headers: Optional[Dict[str, str]],
raw_headers: Optional[Dict[str, str]],
user_api_key_auth: Optional[UserAPIKeyAuth] = None,
scope_servers: Optional[list[MCPServer]] = None,
) -> Tuple[Optional[Union[Dict[str, str], str]], Optional[Dict[str, str]]]:
"""Build auth and extra headers for a server."""
"""Build auth and extra headers for a server.
``scope_servers`` is the full server list a fan-out handler iterates. Passing it lets the
client-forwarded token modes withhold the caller's request-wide ``Authorization`` when
another server in the scope would also receive it (``_caller_authorization_fans_out``);
explicitly-addressed operations leave it None. Per-server ``x-mcp-{alias}-authorization``
headers are unaffected — they bind one token to one server and are the multi-server shape.
"""
server_auth_header: Optional[Union[Dict[str, str], str]] = None
if mcp_server_auth_headers:
from litellm.proxy._experimental.mcp_server.utils import (
@ -1548,6 +1570,16 @@ if MCP_AVAILABLE:
)
extra_headers: Optional[Dict[str, str]] = None
is_client_forwarded_mode = server.is_true_passthrough or server.is_oauth_delegate
# In a multi-server listing scope the request-wide Authorization can only carry one token,
# so it is withheld from a client-forwarded server when another server in scope also consumes
# it (RFC 9700 cross-resource replay); such scopes must bind per-server via
# x-mcp-{alias}-authorization. The decision is computed once so BOTH the forwarding branch and
# the extra_headers copy loop below honor it — otherwise a server that lists Authorization in
# extra_headers would re-copy the withheld bearer from raw_headers and replay it anyway.
withhold_forwarded_authorization = is_client_forwarded_mode and _caller_authorization_fans_out(
server, scope_servers
)
if server.auth_type == MCPAuth.oauth2:
# For OAuth2 M2M servers, upstream Authorization must come from
# client_credentials token fetch, never from caller headers.
@ -1566,6 +1598,14 @@ if MCP_AVAILABLE:
user_api_key_auth=user_api_key_auth,
):
extra_headers = _without_authorization(extra_headers)
elif is_client_forwarded_mode:
if not withhold_forwarded_authorization:
extra_headers = _client_forwarded_authorization_headers(
mcp_server=server,
oauth2_headers=oauth2_headers,
raw_headers=raw_headers,
user_api_key_auth=user_api_key_auth,
)
if server.extra_headers and raw_headers:
if extra_headers is None:
@ -1586,7 +1626,9 @@ if MCP_AVAILABLE:
for header in server.extra_headers:
if not isinstance(header, str):
continue
if header.lower() == "authorization" and strip_caller_authorization:
if header.lower() == "authorization" and (
strip_caller_authorization or withhold_forwarded_authorization
):
continue
header_value = normalized_raw_headers.get(header.lower())
if header_value is None:
@ -1790,6 +1832,7 @@ if MCP_AVAILABLE:
oauth2_headers=oauth2_headers,
raw_headers=raw_headers,
user_api_key_auth=user_api_key_auth,
scope_servers=allowed_mcp_servers,
)
# Prefer server-stored per-user OAuth when configured, so a stale
@ -1976,6 +2019,7 @@ if MCP_AVAILABLE:
oauth2_headers=oauth2_headers,
raw_headers=raw_headers,
user_api_key_auth=user_api_key_auth,
scope_servers=allowed_mcp_servers,
)
try:
@ -2028,6 +2072,7 @@ if MCP_AVAILABLE:
oauth2_headers=oauth2_headers,
raw_headers=raw_headers,
user_api_key_auth=user_api_key_auth,
scope_servers=allowed_mcp_servers,
)
try:
@ -2078,6 +2123,7 @@ if MCP_AVAILABLE:
oauth2_headers=oauth2_headers,
raw_headers=raw_headers,
user_api_key_auth=user_api_key_auth,
scope_servers=allowed_mcp_servers,
)
try:
@ -3582,6 +3628,41 @@ if MCP_AVAILABLE:
headers={"www-authenticate": www_authenticate},
)
if (
server
and server.is_oauth_delegate
and len(mcp_servers or []) == 1
and _get_forwarded_auth_from_scope(scope) is None
and not _client_has_per_server_auth_header(server, mcp_server_auth_headers)
):
www_authenticate = _get_passthrough_www_authenticate(
scope=scope,
server_name=server_name,
)
raise HTTPException(
status_code=401,
detail="Unauthorized",
headers={"www-authenticate": www_authenticate},
)
if (
server
and server.is_true_passthrough
and len(mcp_servers or []) == 1
and not _scope_has_authorization_header(scope)
and not _client_has_per_server_auth_header(server, mcp_server_auth_headers)
):
upstream_status, upstream_www_authenticate = await _probe_upstream_auth(server.url or "", "")
if upstream_status == 401 and upstream_www_authenticate:
raise HTTPException(
status_code=401,
detail="Unauthorized",
headers={"www-authenticate": upstream_www_authenticate},
)
def _scope_has_authorization_header(scope: Scope) -> bool:
return any(key.lower() == b"authorization" for key, _ in scope.get("headers", []))
def _get_forwarded_auth_from_scope(scope: Scope) -> Optional[str]:
"""Return the upstream-bound ``Authorization`` header value, or None.
@ -3609,7 +3690,7 @@ if MCP_AVAILABLE:
url: str,
auth_header: str,
timeout: float = 5.0,
) -> tuple:
) -> tuple[int, Optional[str]]:
"""JSON-RPC initialize-probe the upstream URL to check whether the token is accepted.
Uses POST so StreamableHTTP MCP servers run the same auth path as a
@ -3639,8 +3720,8 @@ if MCP_AVAILABLE:
},
}
probe_headers = {
"Authorization": auth_header,
"Accept": "application/json, text/event-stream",
**({"Authorization": auth_header} if auth_header else {}),
}
try:
resp = await client.post(

4
litellm/types/mcp.py
View file

@ -38,6 +38,8 @@ class MCPAuth(str, enum.Enum):
aws_sigv4 = "aws_sigv4"
token = "token"
oauth2_token_exchange = "oauth2_token_exchange"
true_passthrough = "true_passthrough"
oauth_delegate = "oauth_delegate"
# RFC 8693 default subject_token_type. A NULL column / omitted config key means
@ -60,6 +62,8 @@ MCPAuthType = Optional[
MCPAuth.aws_sigv4,
MCPAuth.token,
MCPAuth.oauth2_token_exchange,
MCPAuth.true_passthrough,
MCPAuth.oauth_delegate,
]
]

15
litellm/types/mcp_server/mcp_server_manager.py
View file

@ -152,6 +152,18 @@ class MCPServer(BaseModel):
"""True if this is an OAuth2 server that relies on per-user tokens (no client_credentials)."""
return self.auth_type == MCPAuth.oauth2 and not self.has_client_credentials
@property
def is_true_passthrough(self) -> bool:
"""True for the transparent-proxy mode: LiteLLM performs no admission auth and forwards the
client's ``Authorization`` to the upstream unchanged."""
return self.auth_type == MCPAuth.true_passthrough
@property
def is_oauth_delegate(self) -> bool:
"""True for the delegated-upstream-OAuth mode: LiteLLM still admits the caller (API key / SSO /
JWT) but forwards the caller's separate upstream ``Authorization`` unchanged, minting nothing."""
return self.auth_type == MCPAuth.oauth_delegate
@property
def requires_per_user_auth(self) -> bool:
"""
@ -167,6 +179,9 @@ class MCPServer(BaseModel):
if self.needs_user_oauth_token:
return True
if self.is_true_passthrough or self.is_oauth_delegate:
return True
# PAT passthrough: auth_type is none but extra_headers includes auth headers
if self.auth_type == MCPAuth.none and self.extra_headers:
auth_header_names = {"authorization", "x-api-key", "api-key", "apikey"}

712
tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py
View file

File diff suppressed because it is too large Load diff

7
tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_adapter.py
View file

@ -23,6 +23,7 @@ from litellm.proxy._experimental.mcp_server.outbound_credentials.types import (
AuthorizationCodeConfig,
CredError,
NoneConfig,
PassthroughConfig,
SharedKey,
TokenExchangeConfig,
)
@ -234,6 +235,12 @@ def test_token_exchange_empty_subject_token_type_normalizes_to_default():
assert spec.config.subject_token_type == "urn:ietf:params:oauth:token-type:access_token"
@pytest.mark.parametrize("auth_type", [MCPAuth.true_passthrough, MCPAuth.oauth_delegate])
def test_client_forwarded_modes_map_to_passthrough_config(auth_type):
spec = to_server_spec(_server(auth_type=auth_type))
assert spec is not None and isinstance(spec.config, PassthroughConfig)
@pytest.mark.parametrize(
"server",
[

25
tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_resolver.py
View file

@ -1,9 +1,9 @@
"""Tests for the resolver dispatch: live arms produce auth, stubbed arms fail closed.
`none`, `api_key` (shared-key source), `authorization_code`, and `token_exchange` are implemented;
every other arm, plus the `api_key` BYOK source, returns a typed `not_implemented` error until its
mode lands. Parametrizing the stubs over one config each also guards reachability: a dropped `case`
would hit `assert_never` and raise instead of returning the stub.
`none`, `api_key` (shared-key source), `passthrough`, `authorization_code`, and `token_exchange` are
implemented; every other arm, plus the `api_key` BYOK source, returns a typed `not_implemented` error
until its mode lands. Parametrizing the stubs over one config each also guards reachability: a dropped
`case` would hit `assert_never` and raise instead of returning the stub.
"""
import httpx
@ -253,9 +253,24 @@ async def test_token_exchange_without_an_exchanger_fails_closed():
assert result.error.tag == "misconfigured"
@pytest.mark.asyncio
async def test_passthrough_forwards_the_inbound_token_verbatim():
subject = Subject(tenant_id="", subject_id="", inbound_token=SecretStr("Bearer upstream-xyz"))
result = await UpstreamCredentialProvider().resolve_credentials(subject, _spec(PassthroughConfig()))
assert isinstance(result, Ok)
assert isinstance(result.ok, StaticHeaderAuth)
assert _emitted(result.ok)["Authorization"] == "Bearer upstream-xyz"
@pytest.mark.asyncio
async def test_passthrough_without_inbound_token_is_a_no_op():
result = await UpstreamCredentialProvider().resolve_credentials(_SUBJECT, _spec(PassthroughConfig()))
assert isinstance(result, Ok)
assert isinstance(result.ok, NoOpAuth)
_STUBBED = [
("api_key_byok", ApiKeyConfig(key_source=Byok())),
("passthrough", PassthroughConfig()),
("client_credentials", ClientCredentialsConfig()),
("aws_sigv4", AwsSigV4Config(region="us-east-1")),
]

151
tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_oauth_passthrough.py
View file

@ -34,8 +34,7 @@ from litellm.types.mcp_server.mcp_server_manager import MCPServer
def _mock_mcp_client_ip():
"""Bypass IP-based access control in tests."""
with patch(
"litellm.proxy._experimental.mcp_server.discoverable_endpoints"
".IPAddressUtils.get_mcp_client_ip",
"litellm.proxy._experimental.mcp_server.discoverable_endpoints.IPAddressUtils.get_mcp_client_ip",
return_value=None,
):
yield
@ -191,9 +190,7 @@ async def test_oauth_protected_resource_passthrough_proxies_upstream_metadata():
extra_headers=["Authorization"],
oauth_passthrough=True,
)
global_mcp_server_manager.registry[passthrough_server.server_id] = (
passthrough_server
)
global_mcp_server_manager.registry[passthrough_server.server_id] = passthrough_server
upstream_payload = {
"resource": "https://upstream.example.com/mcp",
@ -207,18 +204,14 @@ async def test_oauth_protected_resource_passthrough_proxies_upstream_metadata():
mock_client = MagicMock()
mock_client.get = AsyncMock(return_value=mock_response)
with patch.object(
discoverable_endpoints, "get_async_httpx_client", return_value=mock_client
):
with patch.object(discoverable_endpoints, "get_async_httpx_client", return_value=mock_client):
result = await _build_oauth_protected_resource_response(
request=_make_request(),
mcp_server_name="sample_docs",
use_standard_pattern=True,
)
assert result["authorization_servers"] == [
"https://okta.example.com/oauth2/default"
]
assert result["authorization_servers"] == ["https://okta.example.com/oauth2/default"]
# resource is normalized to the gateway URL so bearers are sent back to us
assert result["resource"].endswith("/mcp/sample_docs")
assert result["scopes_supported"] == ["openid", "profile"]
@ -242,9 +235,7 @@ async def test_oauth_protected_resource_passthrough_cache_hit():
extra_headers=["Authorization"],
oauth_passthrough=True,
)
global_mcp_server_manager.registry[passthrough_server.server_id] = (
passthrough_server
)
global_mcp_server_manager.registry[passthrough_server.server_id] = passthrough_server
mock_response = MagicMock()
mock_response.status_code = 200
@ -254,9 +245,7 @@ async def test_oauth_protected_resource_passthrough_cache_hit():
mock_client = MagicMock()
mock_client.get = AsyncMock(return_value=mock_response)
with patch.object(
discoverable_endpoints, "get_async_httpx_client", return_value=mock_client
):
with patch.object(discoverable_endpoints, "get_async_httpx_client", return_value=mock_client):
await _build_oauth_protected_resource_response(
request=_make_request(),
mcp_server_name="sample_docs",
@ -348,12 +337,8 @@ async def test_oauth_metadata_cache_expired_entry_is_refetched():
mock_client = MagicMock()
mock_client.get = AsyncMock(return_value=mock_response)
with patch.object(
discoverable_endpoints, "get_async_httpx_client", return_value=mock_client
):
result = await discoverable_endpoints.fetch_upstream_oauth_protected_resource(
passthrough_server
)
with patch.object(discoverable_endpoints, "get_async_httpx_client", return_value=mock_client):
result = await discoverable_endpoints.fetch_upstream_oauth_protected_resource(passthrough_server)
assert result == {"authorization_servers": ["https://fresh.example.com"]}
assert mock_client.get.await_count == 1
@ -377,16 +362,12 @@ async def test_oauth_protected_resource_passthrough_network_error_returns_502():
extra_headers=["Authorization"],
oauth_passthrough=True,
)
global_mcp_server_manager.registry[passthrough_server.server_id] = (
passthrough_server
)
global_mcp_server_manager.registry[passthrough_server.server_id] = passthrough_server
mock_client = MagicMock()
mock_client.get = AsyncMock(side_effect=httpx.ConnectError("boom"))
with patch.object(
discoverable_endpoints, "get_async_httpx_client", return_value=mock_client
):
with patch.object(discoverable_endpoints, "get_async_httpx_client", return_value=mock_client):
with pytest.raises(HTTPException) as exc_info:
await _build_oauth_protected_resource_response(
request=_make_request(),
@ -414,16 +395,10 @@ async def test_fetch_upstream_metadata_returns_none_when_not_all_candidates_netw
not_found_response = MagicMock()
not_found_response.status_code = 404
mock_client = MagicMock()
mock_client.get = AsyncMock(
side_effect=[not_found_response, httpx.ConnectError("path fallback failed")]
)
mock_client.get = AsyncMock(side_effect=[not_found_response, httpx.ConnectError("path fallback failed")])
with patch.object(
discoverable_endpoints, "get_async_httpx_client", return_value=mock_client
):
result = await discoverable_endpoints.fetch_upstream_oauth_protected_resource(
passthrough_server
)
with patch.object(discoverable_endpoints, "get_async_httpx_client", return_value=mock_client):
result = await discoverable_endpoints.fetch_upstream_oauth_protected_resource(passthrough_server)
assert result is None
assert mock_client.get.await_count == 2
@ -458,9 +433,7 @@ async def test_oauth_protected_resource_gateway_managed_unchanged():
mock_client = MagicMock()
mock_client.get = AsyncMock()
with patch.object(
discoverable_endpoints, "get_async_httpx_client", return_value=mock_client
):
with patch.object(discoverable_endpoints, "get_async_httpx_client", return_value=mock_client):
result = await _build_oauth_protected_resource_response(
request=_make_request(),
mcp_server_name="keycloak_whoami",
@ -468,7 +441,97 @@ async def test_oauth_protected_resource_gateway_managed_unchanged():
)
mock_client.get.assert_not_awaited()
assert result["authorization_servers"] == [
"https://gateway.example.com/keycloak_whoami"
]
assert result["authorization_servers"] == ["https://gateway.example.com/keycloak_whoami"]
assert result["scopes_supported"] == ["read"]
def _make_upstream_metadata_client() -> tuple[dict, MagicMock]:
upstream_payload = {
"resource": "https://upstream.example.com/mcp",
"authorization_servers": ["https://okta.example.com/oauth2/default"],
"scopes_supported": ["openid", "profile"],
"bearer_methods_supported": ["header"],
}
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = upstream_payload
mock_client = MagicMock()
mock_client.get = AsyncMock(return_value=mock_response)
return upstream_payload, mock_client
@pytest.mark.asyncio
async def test_oauth_protected_resource_oauth_delegate_returns_upstream_metadata_verbatim():
"""oauth_delegate discovery must return the upstream metadata verbatim,
resource included. The caller's token is forwarded to and validated by the
upstream, so its audience must be the upstream; rewriting resource to the
gateway would make a strict IdP refuse to mint it or the upstream reject it.
A regression that dropped oauth_delegate from the pass-through predicate would
fall through to the gateway-AS branch and advertise LiteLLM as the AS."""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
global_mcp_server_manager.registry.clear()
delegate_server = MCPServer(
server_id="delegate-1",
name="sample_docs",
server_name="sample_docs",
alias="sample_docs",
url="https://upstream.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth_delegate,
)
global_mcp_server_manager.registry[delegate_server.server_id] = delegate_server
upstream_payload, mock_client = _make_upstream_metadata_client()
try:
with patch.object(discoverable_endpoints, "get_async_httpx_client", return_value=mock_client):
result = await _build_oauth_protected_resource_response(
request=_make_request(),
mcp_server_name="sample_docs",
use_standard_pattern=True,
)
assert result == upstream_payload
assert result["authorization_servers"] == ["https://okta.example.com/oauth2/default"]
assert result["resource"] == "https://upstream.example.com/mcp"
finally:
global_mcp_server_manager.registry.clear()
@pytest.mark.asyncio
async def test_oauth_protected_resource_true_passthrough_returns_upstream_metadata_verbatim():
"""true_passthrough discovery must return the upstream metadata verbatim,
resource included, so the client treats the upstream as the resource and
authorizes directly against it. A regression that rewrote resource (the
gateway-proxied behavior) would break the transparent-proxy contract."""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
global_mcp_server_manager.registry.clear()
true_passthrough_server = MCPServer(
server_id="tp-1",
name="sample_docs",
server_name="sample_docs",
alias="sample_docs",
url="https://upstream.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.true_passthrough,
)
global_mcp_server_manager.registry[true_passthrough_server.server_id] = true_passthrough_server
upstream_payload, mock_client = _make_upstream_metadata_client()
try:
with patch.object(discoverable_endpoints, "get_async_httpx_client", return_value=mock_client):
result = await _build_oauth_protected_resource_response(
request=_make_request(),
mcp_server_name="sample_docs",
use_standard_pattern=True,
)
assert result == upstream_payload
assert result["resource"] == "https://upstream.example.com/mcp"
finally:
global_mcp_server_manager.registry.clear()

626
tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
View file

File diff suppressed because it is too large Load diff

335
tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
View file

@ -1267,6 +1267,341 @@ class TestMCPServerManager:
assert captured_extra_headers == {"Authorization": "Bearer upstream-oauth-bearer"}
async def _capture_call_extra_headers(self, server, oauth2_headers, raw_headers, user_api_key_auth):
manager = MCPServerManager()
mock_client = AsyncMock()
mock_client.call_tool = AsyncMock(return_value=CallToolResult(content=[], isError=False))
captured = {"extra_headers": "unset"}
async def capture_create_mcp_client(
server, mcp_auth_header, extra_headers, stdio_env, subject_token=None, **kwargs
): # pragma: no cover - helper
captured["extra_headers"] = extra_headers
return mock_client
manager._create_mcp_client = AsyncMock(side_effect=capture_create_mcp_client)
await manager._call_regular_mcp_tool(
mcp_server=server,
original_tool_name="tool",
arguments={},
tasks=[],
mcp_auth_header=None,
mcp_server_auth_headers=None,
oauth2_headers=oauth2_headers,
raw_headers=raw_headers,
proxy_logging_obj=None,
user_api_key_auth=user_api_key_auth,
)
return captured["extra_headers"]
@pytest.mark.asyncio
async def test_call_regular_mcp_tool_true_passthrough_forwards_authorization(self):
from litellm.proxy._types import UserAPIKeyAuth
server = MCPServer(
server_id="server-true-passthrough",
name="tp-server",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.true_passthrough,
)
extra_headers = await self._capture_call_extra_headers(
server,
oauth2_headers={"Authorization": "Bearer upstream-token"},
raw_headers={"authorization": "Bearer upstream-token"},
user_api_key_auth=UserAPIKeyAuth(api_key=None),
)
assert extra_headers == {"Authorization": "Bearer upstream-token"}
@pytest.mark.asyncio
async def test_call_regular_mcp_tool_oauth_delegate_forwards_separate_authorization(self):
from litellm.proxy._types import UserAPIKeyAuth
server = MCPServer(
server_id="server-oauth-delegate",
name="od-server",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth_delegate,
)
extra_headers = await self._capture_call_extra_headers(
server,
oauth2_headers={"Authorization": "Bearer upstream-token"},
raw_headers={
"x-litellm-api-key": "Bearer sk-litellm-key",
"authorization": "Bearer upstream-token",
},
user_api_key_auth=UserAPIKeyAuth(api_key="sk-litellm-key"),
)
assert extra_headers == {"Authorization": "Bearer upstream-token"}
@pytest.mark.asyncio
async def test_call_regular_mcp_tool_oauth_delegate_never_forwards_admission_key(self):
from litellm.proxy._types import UserAPIKeyAuth
server = MCPServer(
server_id="server-oauth-delegate-leak",
name="od-server",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth_delegate,
)
extra_headers = await self._capture_call_extra_headers(
server,
oauth2_headers={"Authorization": "Bearer sk-litellm-key"},
raw_headers={"authorization": "Bearer sk-litellm-key"},
user_api_key_auth=UserAPIKeyAuth(api_key="sk-litellm-key"),
)
assert not extra_headers or "authorization" not in {k.lower() for k in extra_headers}
def test_should_strip_caller_authorization_new_modes(self):
from litellm.proxy._types import UserAPIKeyAuth
true_passthrough = MCPServer(
server_id="tp",
name="tp",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.true_passthrough,
)
assert (
_should_strip_caller_authorization(
mcp_server=true_passthrough,
raw_headers={"authorization": "Bearer upstream"},
user_api_key_auth=UserAPIKeyAuth(api_key=None),
)
is False
)
oauth_delegate = MCPServer(
server_id="od",
name="od",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth_delegate,
)
assert (
_should_strip_caller_authorization(
mcp_server=oauth_delegate,
raw_headers={
"x-litellm-api-key": "Bearer sk-litellm-key",
"authorization": "Bearer upstream",
},
user_api_key_auth=UserAPIKeyAuth(api_key="sk-litellm-key"),
)
is False
)
assert (
_should_strip_caller_authorization(
mcp_server=oauth_delegate,
raw_headers={"authorization": "Bearer sk-litellm-key"},
user_api_key_auth=UserAPIKeyAuth(api_key="sk-litellm-key"),
)
is True
)
def test_should_strip_authorization_for_oauth_delegate_admitted_via_jwt_without_api_key(self):
"""JWT / SSO / OIDC / session admission yields a UserAPIKeyAuth with a user_id but
api_key=None; the caller's Authorization was that credential and must be stripped for
oauth_delegate when no separate x-litellm-api-key carried admission (LIT-3794-class leak)."""
from litellm.proxy._types import UserAPIKeyAuth
oauth_delegate = MCPServer(
server_id="od-jwt",
name="od",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth_delegate,
)
assert (
_should_strip_caller_authorization(
mcp_server=oauth_delegate,
raw_headers={"authorization": "Bearer eyJ-idp-jwt"},
user_api_key_auth=UserAPIKeyAuth(user_id="alice", api_key=None),
)
is True
)
assert (
_should_strip_caller_authorization(
mcp_server=oauth_delegate,
raw_headers={
"x-litellm-api-key": "Bearer sk-1234",
"authorization": "Bearer upstream",
},
user_api_key_auth=UserAPIKeyAuth(user_id="alice", api_key=None),
)
is False
)
@pytest.mark.asyncio
async def test_call_regular_mcp_tool_oauth_delegate_never_forwards_jwt_admission(self):
from litellm.proxy._types import UserAPIKeyAuth
server = MCPServer(
server_id="od-jwt-e2e",
name="od",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth_delegate,
)
extra_headers = await self._capture_call_extra_headers(
server,
oauth2_headers={"Authorization": "Bearer eyJ-idp-jwt"},
raw_headers={"authorization": "Bearer eyJ-idp-jwt"},
user_api_key_auth=UserAPIKeyAuth(user_id="alice", api_key=None),
)
assert not extra_headers or "authorization" not in {k.lower() for k in extra_headers}
def test_new_passthrough_modes_require_per_user_auth(self):
for auth_type in (MCPAuth.true_passthrough, MCPAuth.oauth_delegate):
server = MCPServer(
server_id="s",
name="s",
url="https://example.com",
transport=MCPTransport.http,
auth_type=auth_type,
)
assert server.requires_per_user_auth is True
@pytest.mark.asyncio
async def test_create_mcp_client_forwarded_modes_use_the_passthrough_arm(self):
manager = MCPServerManager()
server = MCPServer(
server_id="tp-egress",
name="tp",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.true_passthrough,
)
with (
patch(
"litellm.proxy._experimental.mcp_server.mcp_server_manager.resolve_mcp_auth",
new_callable=AsyncMock,
) as mock_resolve,
patch("litellm.proxy._experimental.mcp_server.mcp_server_manager.MCPClient") as mock_client_cls,
):
await manager._create_mcp_client(server=server, extra_headers={"Authorization": "Bearer upstream-token"})
mock_resolve.assert_not_awaited()
kwargs = mock_client_cls.call_args.kwargs
emitted = httpx.Request("GET", "https://example.com/mcp")
flow = kwargs["resolved_auth"].auth_flow(emitted)
next(flow)
flow.close()
assert emitted.headers["Authorization"] == "Bearer upstream-token"
assert not kwargs["extra_headers"] or "authorization" not in {k.lower() for k in kwargs["extra_headers"]}
@staticmethod
def _emitted_authorization(mock_client_cls) -> str:
kwargs = mock_client_cls.call_args.kwargs
emitted = httpx.Request("GET", "https://example.com/mcp")
flow = kwargs["resolved_auth"].auth_flow(emitted)
next(flow)
flow.close()
return emitted.headers["Authorization"]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"per_server_header",
["Bearer per-server-token", {"Authorization": "Bearer per-server-token"}],
)
async def test_create_mcp_client_passthrough_prefers_per_server_token(self, per_server_header):
"""A per-server x-mcp-{alias}-authorization value is the explicit one-token-one-server
binding, so it must win over the request-wide Authorization and reach the upstream
verbatim through the passthrough arm."""
manager = MCPServerManager()
server = MCPServer(
server_id="tp-per-server",
name="tp",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.true_passthrough,
)
with (
patch(
"litellm.proxy._experimental.mcp_server.mcp_server_manager.resolve_mcp_auth",
new_callable=AsyncMock,
) as mock_resolve,
patch("litellm.proxy._experimental.mcp_server.mcp_server_manager.MCPClient") as mock_client_cls,
):
await manager._create_mcp_client(
server=server,
mcp_auth_header=per_server_header,
extra_headers={"Authorization": "Bearer global-token"},
)
mock_resolve.assert_not_awaited()
assert self._emitted_authorization(mock_client_cls) == "Bearer per-server-token"
kwargs = mock_client_cls.call_args.kwargs
assert not kwargs["extra_headers"] or "authorization" not in {k.lower() for k in kwargs["extra_headers"]}
def test_consumes_caller_authorization_per_mode(self):
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
_consumes_caller_authorization,
)
def build(**kwargs) -> MCPServer:
return MCPServer(
server_id="s",
name="s",
url="https://example.com",
transport=MCPTransport.http,
**kwargs,
)
assert _consumes_caller_authorization(build(auth_type=MCPAuth.true_passthrough)) is True
assert _consumes_caller_authorization(build(auth_type=MCPAuth.oauth_delegate)) is True
assert (
_consumes_caller_authorization(
build(auth_type=MCPAuth.none, extra_headers=["Authorization"], oauth_passthrough=True)
)
is True
)
assert _consumes_caller_authorization(build(auth_type=MCPAuth.oauth2, delegate_auth_to_upstream=True)) is True
assert _consumes_caller_authorization(build(auth_type=MCPAuth.api_key, authentication_token="x")) is False
assert (
_consumes_caller_authorization(
build(
auth_type=MCPAuth.oauth2,
delegate_auth_to_upstream=True,
oauth2_flow="client_credentials",
token_url="https://idp/token",
)
)
is False
)
def test_caller_authorization_fans_out_only_with_second_consumer(self):
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
_caller_authorization_fans_out,
)
delegate = MCPServer(
server_id="od",
name="od",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth_delegate,
)
second = MCPServer(
server_id="tp",
name="tp",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.true_passthrough,
)
static_server = MCPServer(
server_id="static",
name="static",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.api_key,
authentication_token="x",
)
assert _caller_authorization_fans_out(delegate, None) is False
assert _caller_authorization_fans_out(delegate, [delegate]) is False
assert _caller_authorization_fans_out(delegate, [delegate, static_server]) is False
assert _caller_authorization_fans_out(delegate, [delegate, second]) is True
@pytest.mark.asyncio
async def test_get_prompts_from_server_success(self):
"""Ensure prompts are fetched and prefixed when requested."""

424
tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_stale_session.py
View file

@ -713,6 +713,9 @@ async def test_handle_streamable_http_mcp_delegated_server_surfaces_upstream_cha
delegated_server.auth_type = MCPAuth.oauth2
delegated_server.delegate_auth_to_upstream = True
delegated_server.needs_user_oauth_token = True
delegated_server.is_oauth_passthrough = False
delegated_server.is_oauth_delegate = False
delegated_server.is_true_passthrough = False
delegated_server.server_id = "delegated-oauth-server"
upstream_challenge = 'Bearer resource_metadata="https://upstream.example.com/.well-known/oauth-protected-resource"'
@ -1048,3 +1051,424 @@ async def test_handle_streamable_http_mcp_token_exchange_without_subject_returns
assert "/.well-known/oauth-protected-resource" in challenge
assert challenge.split('resource_metadata="', 1)[1].split('"', 1)[0].endswith("/mcp/obo_server")
assert 'error="invalid_token"' in challenge
def _passthrough_mode_scope(server_name: str, extra_headers=None):
headers = [
(b"content-type", b"application/json"),
(b"host", b"litellm.example.com"),
] + list(extra_headers or [])
return {
"type": "http",
"method": "POST",
"path": f"/mcp/{server_name}",
"_original_path": f"/{server_name}/mcp",
"scheme": "https",
"query_string": b"",
"root_path": "",
"server": ("litellm.example.com", 443),
"headers": headers,
}
def _build_passthrough_mode_server(server_name: str, auth_type):
from litellm.proxy._types import MCPTransport
from litellm.types.mcp_server.mcp_server_manager import MCPServer
return MCPServer(
server_id=f"{server_name}-id",
name=server_name,
server_name=server_name,
alias=server_name,
url="https://upstream.example.com/mcp",
transport=MCPTransport.http,
auth_type=auth_type,
)
@pytest.mark.asyncio
async def test_handle_streamable_http_mcp_oauth_delegate_without_token_returns_gateway_proxied_401():
"""oauth_delegate is admitted with the LiteLLM key but still owns upstream
OAuth. With no forwarded upstream token the gateway must challenge with the
proxied resource_metadata (which advertises the upstream IdP), never the
gateway authorization_uri and never a silent 200."""
from fastapi import HTTPException
try:
from litellm.proxy._experimental.mcp_server.server import (
handle_streamable_http_mcp,
session_manager_stateful,
)
except ImportError:
pytest.skip("MCP server not available")
scope = _passthrough_mode_scope("od_server")
receive = AsyncMock(
return_value={
"type": "http.request",
"body": b'{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}',
"more_body": False,
}
)
send = AsyncMock()
user_auth = MagicMock()
user_auth.user_id = "u1"
od_server = _build_passthrough_mode_server("od_server", MCPAuth.oauth_delegate)
with (
patch(
"litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
new_callable=AsyncMock,
return_value=(user_auth, None, ["od_server"], None, None, None),
),
patch("litellm.proxy._experimental.mcp_server.server.set_auth_context"),
patch(
"litellm.proxy._experimental.mcp_server.server._SESSION_MANAGERS_INITIALIZED",
True,
),
patch(
"litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
return_value=od_server,
),
patch.object(
session_manager_stateful,
"handle_request",
new_callable=AsyncMock,
) as mock_handle_request,
):
with pytest.raises(HTTPException) as exc_info:
await handle_streamable_http_mcp(scope, receive, send)
assert mock_handle_request.await_count == 0
assert exc_info.value.status_code == 401
challenge = exc_info.value.headers["www-authenticate"]
assert "resource_metadata=" in challenge
assert "authorization_uri=" not in challenge
assert "/.well-known/oauth-protected-resource/od_server/mcp" in challenge
@pytest.mark.asyncio
async def test_handle_streamable_http_mcp_oauth_delegate_with_forwarded_token_skips_challenge():
"""When the oauth_delegate caller carries both the LiteLLM key and a separate
upstream Authorization, the gateway must forward to the session manager, not
re-challenge. Guards the ``_get_forwarded_auth_from_scope(...) is None``
condition: dropping it would 401 even a fully-authenticated request."""
try:
from litellm.proxy._experimental.mcp_server.server import (
handle_streamable_http_mcp,
session_manager_stateless,
)
except ImportError:
pytest.skip("MCP server not available")
scope = _passthrough_mode_scope(
"od_server",
extra_headers=[
(b"x-litellm-api-key", b"Bearer sk-1234"),
(b"authorization", b"Bearer upstream-token"),
],
)
receive = AsyncMock(
return_value={
"type": "http.request",
"body": b'{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}',
"more_body": False,
}
)
send = AsyncMock()
user_auth = MagicMock()
user_auth.user_id = "u1"
od_server = _build_passthrough_mode_server("od_server", MCPAuth.oauth_delegate)
with (
patch(
"litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
new_callable=AsyncMock,
return_value=(user_auth, None, ["od_server"], None, None, None),
),
patch("litellm.proxy._experimental.mcp_server.server.set_auth_context"),
patch(
"litellm.proxy._experimental.mcp_server.server._SESSION_MANAGERS_INITIALIZED",
True,
),
patch(
"litellm.proxy._experimental.mcp_server.server._handle_stale_mcp_session",
new_callable=AsyncMock,
return_value=False,
),
patch(
"litellm.proxy._experimental.mcp_server.server._check_passthrough_upstream_auth",
new_callable=AsyncMock,
),
patch(
"litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
return_value=od_server,
),
patch.object(
session_manager_stateless,
"handle_request",
new_callable=AsyncMock,
) as mock_handle_request,
patch.object(session_manager_stateless, "_server_instances", {}),
):
await handle_streamable_http_mcp(scope, receive, send)
assert mock_handle_request.await_count == 1
async def _run_passthrough_connect(
*,
auth_type,
server_names,
mcp_server_auth_headers,
scope_extra_headers=None,
):
"""Drive handle_streamable_http_mcp through the preemptive-401 gate and report whether it
challenged (raised) or forwarded to the session manager. Returns (challenged, www_authenticate)."""
from litellm.proxy._experimental.mcp_server.server import (
handle_streamable_http_mcp,
session_manager_stateless,
)
scope = _passthrough_mode_scope(server_names[0], extra_headers=scope_extra_headers)
receive = AsyncMock(
return_value={
"type": "http.request",
"body": b'{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}',
"more_body": False,
}
)
send = AsyncMock()
user_auth = MagicMock()
user_auth.user_id = "u1"
server = _build_passthrough_mode_server(server_names[0], auth_type)
with (
patch(
"litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
new_callable=AsyncMock,
return_value=(user_auth, None, server_names, mcp_server_auth_headers, None, None),
),
patch("litellm.proxy._experimental.mcp_server.server.set_auth_context"),
patch("litellm.proxy._experimental.mcp_server.server._SESSION_MANAGERS_INITIALIZED", True),
patch(
"litellm.proxy._experimental.mcp_server.server._handle_stale_mcp_session",
new_callable=AsyncMock,
return_value=False,
),
patch(
"litellm.proxy._experimental.mcp_server.server._check_passthrough_upstream_auth",
new_callable=AsyncMock,
),
patch(
"litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
return_value=server,
),
patch.object(session_manager_stateless, "handle_request", new_callable=AsyncMock) as mock_handle_request,
patch.object(session_manager_stateless, "_server_instances", {}),
):
try:
await handle_streamable_http_mcp(scope, receive, send)
except HTTPException as exc:
return True, (exc.headers or {}).get("www-authenticate")
return mock_handle_request.await_count == 0, None
@pytest.mark.asyncio
@pytest.mark.parametrize("auth_type", [MCPAuth.oauth_delegate, MCPAuth.true_passthrough])
async def test_handle_streamable_http_mcp_per_server_header_skips_preemptive_challenge(auth_type):
"""A per-server x-mcp-{alias}-authorization header binds the upstream token to one server; the
connect gate must recognize it and forward instead of spuriously 401-ing, since egress already
honors it. Without this, the mandatory multi-server binding is unusable at connect."""
try:
from litellm.proxy._experimental.mcp_server.server import handle_streamable_http_mcp # noqa: F401
except ImportError:
pytest.skip("MCP server not available")
challenged, _ = await _run_passthrough_connect(
auth_type=auth_type,
server_names=["pt_server"],
mcp_server_auth_headers={"pt_server": {"Authorization": "Bearer upstream-token"}},
)
assert challenged is False
@pytest.mark.asyncio
@pytest.mark.parametrize("auth_type", [MCPAuth.oauth_delegate, MCPAuth.true_passthrough])
async def test_handle_streamable_http_mcp_sanitized_per_server_header_skips_preemptive_challenge(auth_type):
"""A dashboard client sends x-mcp-{sanitize_mcp_alias_for_header(alias)}-authorization, so the
alias 'pt-server' arrives as the header key 'pt_server'. Egress resolves that via the sanitized
alias, so the connect gate must too, or it 401s a token egress would forward."""
try:
from litellm.proxy._experimental.mcp_server.server import handle_streamable_http_mcp # noqa: F401
except ImportError:
pytest.skip("MCP server not available")
challenged, _ = await _run_passthrough_connect(
auth_type=auth_type,
server_names=["pt-server"],
mcp_server_auth_headers={"pt_server": {"Authorization": "Bearer upstream-token"}},
)
assert challenged is False
@pytest.mark.asyncio
@pytest.mark.parametrize("auth_type", [MCPAuth.oauth_delegate, MCPAuth.true_passthrough])
async def test_handle_streamable_http_mcp_aggregate_does_not_preemptively_challenge(auth_type):
"""A multi-server aggregate must degrade gracefully: the preemptive 401 is single-server only, so
one server missing a token cannot 401 the whole connect (the listing absorbs per-server failures)."""
try:
from litellm.proxy._experimental.mcp_server.server import handle_streamable_http_mcp # noqa: F401
except ImportError:
pytest.skip("MCP server not available")
challenged, _ = await _run_passthrough_connect(
auth_type=auth_type,
server_names=["pt_server", "pt_server_2"],
mcp_server_auth_headers=None,
)
assert challenged is False
@pytest.mark.asyncio
async def test_handle_streamable_http_mcp_true_passthrough_without_token_surfaces_verbatim_upstream_challenge():
"""true_passthrough is a transparent proxy: with no client Authorization the
gateway probes the upstream and surfaces its own WWW-Authenticate verbatim,
so the client discovers and authorizes against the upstream directly. Guards
against answering initialize locally with a silent 200."""
from fastapi import HTTPException
try:
from litellm.proxy._experimental.mcp_server.server import (
handle_streamable_http_mcp,
session_manager_stateful,
)
except ImportError:
pytest.skip("MCP server not available")
upstream_challenge = 'Bearer resource_metadata="https://upstream.example.com/.well-known/oauth-protected-resource"'
probe_response = MagicMock()
probe_response.status_code = 401
probe_response.headers = {"www-authenticate": upstream_challenge}
probe_client = MagicMock()
probe_client.post = AsyncMock(return_value=probe_response)
scope = _passthrough_mode_scope("tp_server")
receive = AsyncMock(
return_value={
"type": "http.request",
"body": b'{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}',
"more_body": False,
}
)
send = AsyncMock()
user_auth = MagicMock()
user_auth.user_id = None
tp_server = _build_passthrough_mode_server("tp_server", MCPAuth.true_passthrough)
with (
patch(
"litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
new_callable=AsyncMock,
return_value=(user_auth, None, ["tp_server"], None, None, None),
),
patch("litellm.proxy._experimental.mcp_server.server.set_auth_context"),
patch(
"litellm.proxy._experimental.mcp_server.server._SESSION_MANAGERS_INITIALIZED",
True,
),
patch(
"litellm.proxy._experimental.mcp_server.server.get_async_httpx_client",
return_value=probe_client,
),
patch(
"litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
return_value=tp_server,
),
patch.object(
session_manager_stateful,
"handle_request",
new_callable=AsyncMock,
) as mock_handle_request,
):
with pytest.raises(HTTPException) as exc_info:
await handle_streamable_http_mcp(scope, receive, send)
assert mock_handle_request.await_count == 0
assert exc_info.value.status_code == 401
assert exc_info.value.headers["www-authenticate"] == upstream_challenge
probe_client.post.assert_awaited_once()
@pytest.mark.asyncio
async def test_handle_streamable_http_mcp_true_passthrough_with_token_skips_probe_and_challenge():
"""When the true_passthrough caller already carries an Authorization the
gateway must forward without probing or challenging. Guards the
``not _scope_has_authorization_header(scope)`` condition and the no-probe
fast path."""
try:
from litellm.proxy._experimental.mcp_server.server import (
handle_streamable_http_mcp,
session_manager_stateless,
)
except ImportError:
pytest.skip("MCP server not available")
probe_client = MagicMock()
probe_client.post = AsyncMock()
scope = _passthrough_mode_scope(
"tp_server",
extra_headers=[(b"authorization", b"Bearer upstream-token")],
)
receive = AsyncMock(
return_value={
"type": "http.request",
"body": b'{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}',
"more_body": False,
}
)
send = AsyncMock()
user_auth = MagicMock()
user_auth.user_id = None
tp_server = _build_passthrough_mode_server("tp_server", MCPAuth.true_passthrough)
with (
patch(
"litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
new_callable=AsyncMock,
return_value=(user_auth, None, ["tp_server"], None, None, None),
),
patch("litellm.proxy._experimental.mcp_server.server.set_auth_context"),
patch(
"litellm.proxy._experimental.mcp_server.server._SESSION_MANAGERS_INITIALIZED",
True,
),
patch(
"litellm.proxy._experimental.mcp_server.server._handle_stale_mcp_session",
new_callable=AsyncMock,
return_value=False,
),
patch(
"litellm.proxy._experimental.mcp_server.server._check_passthrough_upstream_auth",
new_callable=AsyncMock,
),
patch(
"litellm.proxy._experimental.mcp_server.server.get_async_httpx_client",
return_value=probe_client,
),
patch(
"litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name",
return_value=tp_server,
),
patch.object(
session_manager_stateless,
"handle_request",
new_callable=AsyncMock,
) as mock_handle_request,
patch.object(session_manager_stateless, "_server_instances", {}),
):
await handle_streamable_http_mcp(scope, receive, send)
assert mock_handle_request.await_count == 1
probe_client.post.assert_not_awaited()

2
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -27020,7 +27020,7 @@ export interface components {
/** Alias */
alias?: string | null;
/** Auth Type */
auth_type?: ("none" | "api_key" | "bearer_token" | "basic" | "authorization" | "oauth2" | "aws_sigv4" | "token" | "oauth2_token_exchange") | null;
auth_type?: ("none" | "api_key" | "bearer_token" | "basic" | "authorization" | "oauth2" | "aws_sigv4" | "token" | "oauth2_token_exchange" | "true_passthrough" | "oauth_delegate") | null;
/** Mcp Info */
mcp_info?: {
[key: string]: unknown;