feat(mcp/v2): map authorization_code + passthrough in _to_server_spec

Adds the interactive-oauth2 split so v2 routes these modes instead of deferring to v1:
- interactive oauth2 (needs_user_oauth_token) + delegate_auth_to_upstream=false -> AuthorizationCodeConfig
  (gateway-stored per-user token; the caller JWT is never forwarded). This is the v2 form of the
  LIT-3795 fix: a non-delegated interactive oauth2 server must not forward the proxy-auth JWT.
- interactive oauth2 + delegate_auth_to_upstream=true -> PassthroughConfig (forward the caller token).
- auth_type=none + is_oauth_passthrough -> PassthroughConfig (forward), else NoneConfig.

aws_sigv4 stays on the aws_auth seam; misconfigured modes still defer to v1. These per-user modes
fire live at the egress cutover (they ride the extra_headers/mcp_auth_header seam, not
resolve_mcp_auth); the token bridges backing authorization_code land next in this step.
This commit is contained in:
Tin Chi Lo 2026-06-19 10:37:30 -07:00
parent 9256b5e620
commit fccd04a494
2 changed files with 123 additions and 1 deletions

View file

@ -47,6 +47,7 @@ from litellm.proxy.gateway.mcp.outbound_credentials.types import (
ClientCredentialsConfig,
CredError,
NoneConfig,
PassthroughConfig,
ServerSpec,
SharedKey,
StaticKeys,
@ -111,6 +112,14 @@ def _provider() -> UpstreamCredentialProvider:
def _to_server_spec(server: MCPServer) -> Optional[ServerSpec]:
resource = server.url or server.server_id
if server.auth_type in (None, MCPAuth.none):
# A none server opted into upstream OAuth passthrough forwards the caller's bearer; the
# passthrough arm sends the inbound token. Otherwise no upstream credential.
if server.is_oauth_passthrough:
return ServerSpec(
server_id=server.server_id,
resource=resource,
config=PassthroughConfig(),
)
return ServerSpec(
server_id=server.server_id, resource=resource, config=NoneConfig()
)
@ -171,7 +180,31 @@ def _to_server_spec(server: MCPServer) -> Optional[ServerSpec]:
scopes=tuple(server.scopes or ()),
),
)
return None # other modes are not grafted yet
if server.needs_user_oauth_token:
# Interactive oauth2 (3LO), not M2M/exchange (handled above). delegate_auth_to_upstream
# forwards the caller token (passthrough); otherwise the gateway holds a per-user token
# (authorization_code). This split is the LIT-3795 fix: a non-delegated interactive oauth2
# server must not forward the caller JWT upstream.
if server.delegate_auth_to_upstream:
return ServerSpec(
server_id=server.server_id,
resource=resource,
config=PassthroughConfig(),
)
return ServerSpec(
server_id=server.server_id,
resource=resource,
config=AuthorizationCodeConfig(
client_id=server.client_id,
client_secret=(
SecretStr(server.client_secret) if server.client_secret else None
),
authorization_url=server.authorization_url,
token_url=server.token_url,
scopes=tuple(server.scopes or ()),
),
)
return None # aws_sigv4 uses the aws_auth seam; misconfigured modes defer to v1
def _added_headers(auth: httpx.Auth) -> Dict[str, str]:

View file

@ -324,3 +324,92 @@ async def test_token_exchange_server_maps_to_config():
assert spec.config.client_secret.get_secret_value() == "csecret"
assert spec.config.scopes == ("a", "b")
assert spec.resource == "https://aud.example" # audience preferred for the binding
async def test_interactive_oauth2_no_delegate_maps_to_authorization_code():
# LIT-3795: a non-delegated interactive oauth2 server resolves to authorization_code
# (gateway-stored per-user token), so the caller JWT is never forwarded upstream.
from litellm.proxy._experimental.mcp_server.v2_resolver_bridge import (
_to_server_spec,
)
from litellm.proxy.gateway.mcp.outbound_credentials.types import (
AuthorizationCodeConfig,
)
server = MCPServer(
server_id="oa1",
name="oa1",
transport=MCPTransport.http,
url="https://up.example/mcp",
auth_type=MCPAuth.oauth2,
oauth2_flow="authorization_code",
delegate_auth_to_upstream=False,
client_id="cid",
client_secret="csecret",
authorization_url="https://idp/authorize",
token_url="https://idp/token",
scopes=["openid"],
)
spec = _to_server_spec(server)
assert spec is not None
assert isinstance(spec.config, AuthorizationCodeConfig)
assert spec.config.token_url == "https://idp/token"
assert spec.config.client_id == "cid"
async def test_interactive_oauth2_delegate_maps_to_passthrough():
from litellm.proxy._experimental.mcp_server.v2_resolver_bridge import (
_to_server_spec,
)
from litellm.proxy.gateway.mcp.outbound_credentials.types import PassthroughConfig
server = MCPServer(
server_id="oa2",
name="oa2",
transport=MCPTransport.http,
url="https://up.example/mcp",
auth_type=MCPAuth.oauth2,
oauth2_flow="authorization_code",
delegate_auth_to_upstream=True,
)
spec = _to_server_spec(server)
assert spec is not None
assert isinstance(spec.config, PassthroughConfig)
async def test_none_oauth_passthrough_maps_to_passthrough():
from litellm.proxy._experimental.mcp_server.v2_resolver_bridge import (
_to_server_spec,
)
from litellm.proxy.gateway.mcp.outbound_credentials.types import PassthroughConfig
server = MCPServer(
server_id="pt1",
name="pt1",
transport=MCPTransport.http,
url="https://up.example/mcp",
auth_type=MCPAuth.none,
oauth_passthrough=True,
extra_headers=["Authorization"],
)
spec = _to_server_spec(server)
assert spec is not None
assert isinstance(spec.config, PassthroughConfig)
async def test_none_without_passthrough_maps_to_none():
from litellm.proxy._experimental.mcp_server.v2_resolver_bridge import (
_to_server_spec,
)
from litellm.proxy.gateway.mcp.outbound_credentials.types import NoneConfig
server = MCPServer(
server_id="n1",
name="n1",
transport=MCPTransport.http,
url="https://up.example/mcp",
auth_type=MCPAuth.none,
)
spec = _to_server_spec(server)
assert spec is not None
assert isinstance(spec.config, NoneConfig)