diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py index 5090aa7d7d5..68a61b85175 100644 --- a/litellm/proxy/_experimental/mcp_server/server.py +++ b/litellm/proxy/_experimental/mcp_server/server.py @@ -3719,8 +3719,15 @@ if MCP_AVAILABLE: headers={"www-authenticate": upstream_www_authenticate}, ) + def _get_authorization_header_from_scope(scope: Scope) -> Optional[str]: + """First ``Authorization`` header value in the ASGI scope, or None.""" + for key, value in scope.get("headers", []): + if key.lower() == b"authorization": + return value.decode("latin-1") + return None + def _scope_has_authorization_header(scope: Scope) -> bool: - return any(key.lower() == b"authorization" for key, _ in scope.get("headers", [])) + return _get_authorization_header_from_scope(scope) is not None def _get_forwarded_auth_from_scope(scope: Scope) -> Optional[str]: """Return the upstream-bound ``Authorization`` header value, or None. @@ -3733,17 +3740,24 @@ if MCP_AVAILABLE: ``MCPRequestHandler.process_mcp_request``), and forwarding it upstream would leak the proxy key to a third-party MCP server. """ - authorization = None - has_litellm_key_header = False - for key, value in scope.get("headers", []): - key_lower = key.lower() - if key_lower == b"authorization": - authorization = value.decode("latin-1") - elif key_lower == b"x-litellm-api-key": - has_litellm_key_header = True + has_litellm_key_header = any(key.lower() == b"x-litellm-api-key" for key, _ in scope.get("headers", [])) if not has_litellm_key_header: return None - return authorization + return _get_authorization_header_from_scope(scope) + + def _is_delegate_upstream_probe_target(server: MCPServer) -> bool: + """Whether ``server`` is an interactive delegate-auth server whose client-supplied + token should be preflighted upstream. + + Mirrors the anonymous-delegate gate in ``get_allowed_mcp_servers``: the flow is + resolved via ``effective_oauth2_flow`` so an unstamped M2M-shape row fails closed + (its stored client credentials drive egress; the caller's bearer is irrelevant). + """ + return ( + server.auth_type == MCPAuth.oauth2 + and server.delegate_auth_to_upstream is True + and MCPServerManager.effective_oauth2_flow(server) != "client_credentials" + ) async def _probe_upstream_auth( url: str, @@ -3805,7 +3819,7 @@ if MCP_AVAILABLE: mcp_servers: Optional[List[str]], client_ip: Optional[str], ) -> None: - """Probe pass-through upstream servers in parallel before the MCP session starts. + """Probe pass-through and delegate-auth upstream servers in parallel before the MCP session starts. Only servers the caller's key is already authorized to reach are probed — the list is derived from _get_allowed_mcp_servers so that a user cannot @@ -3813,11 +3827,42 @@ if MCP_AVAILABLE: The MCP SDK commits HTTP 200 headers before invoking handlers, so a 401 can only be returned before that point. This function raises HTTPException(401) - with a WWW-Authenticate header if any upstream rejects the client token. + with a WWW-Authenticate header if any upstream rejects the client token, or 403 + if the upstream accepts it but forbids the caller. Fails-open: network errors are logged and the request is allowed through. + + Delegate-auth servers (``auth_type=oauth2`` + ``delegate_auth_to_upstream``) + are probed with the caller's bare ``Authorization`` bearer. That bearer is only + an upstream token (never a LiteLLM key) when admission took the delegate bypass, + so the delegate target is resolved through ``get_mcp_server_by_name`` -- the same + resolver admission used -- rather than the wider allowed-server prefix/access-group + matching. A name that only reaches a delegate server via server_id or an access + group would have been admitted as a real LiteLLM key, so probing it would leak that + key upstream; requiring the admission-resolver match closes that gap. Without the + probe a rejected token is absorbed by the tools/list handler and masked as an empty + tool list. Gated to single-server routes so one rejected token cannot 401 a + multi-server aggregate connect, matching the OBO preflight gating; the challenge + echoes the requested name so aliased routes get the same resource_metadata URL as + the tokenless preemptive challenge. """ forwarded_auth = _get_forwarded_auth_from_scope(scope) - if not forwarded_auth: + requested_single_target = mcp_servers[0] if mcp_servers is not None and len(mcp_servers) == 1 else None + # The bare Authorization header (no x-litellm-api-key) is a valid upstream token + # only when admission classified it as one, i.e. the single requested name resolves + # to a delegate server under admission's own resolver. Resolve it the same way here + # so a server_id- or access-group-named delegate (which admission would have treated + # as a LiteLLM key) is never probed with that key. + delegate_server = ( + global_mcp_server_manager.get_mcp_server_by_name(requested_single_target, client_ip=client_ip) + if requested_single_target + else None + ) + delegate_auth = ( + _get_authorization_header_from_scope(scope) + if delegate_server is not None and _is_delegate_upstream_probe_target(delegate_server) + else None + ) + if not forwarded_auth and not delegate_auth: return # Use the authorized server set, not the raw user-supplied names, so that @@ -3827,33 +3872,49 @@ if MCP_AVAILABLE: mcp_servers=mcp_servers, client_ip=client_ip, ) - passthrough_servers = [ - srv - for srv in allowed_servers - # Restrict to genuine OAuth pass-through servers (auth_type none + - # Authorization in extra_headers). Gateway-managed OAuth2 servers - # must not receive the ``resource_metadata=`` challenge emitted - # below — they require ``authorization_uri=`` pointing at the - # gateway AS metadata. ``is_oauth_passthrough`` already requires - # ``auth_type in (None, MCPAuth.none)``, which is mutually - # exclusive with ``has_client_credentials`` (oauth2 + M2M flow), - # so M2M servers are implicitly excluded here. - if srv.is_oauth_passthrough - ] - if not passthrough_servers: + passthrough_targets: Tuple[Tuple[MCPServer, str, str], ...] = ( + tuple( + (srv, forwarded_auth, srv.name) + for srv in allowed_servers + # Restrict to genuine OAuth pass-through servers (auth_type none + + # Authorization in extra_headers). Gateway-managed OAuth2 servers + # must not receive the ``resource_metadata=`` challenge emitted + # below — they require ``authorization_uri=`` pointing at the + # gateway AS metadata. ``is_oauth_passthrough`` already requires + # ``auth_type in (None, MCPAuth.none)``, which is mutually + # exclusive with ``has_client_credentials`` (oauth2 + M2M flow), + # so M2M servers are implicitly excluded here. + if srv.is_oauth_passthrough + ) + if forwarded_auth + else () + ) + # Probe the admission-resolved delegate server only when the caller is actually + # authorized for it (present in the IP-filtered allowed set), keyed by server_id. + delegate_targets: Tuple[Tuple[MCPServer, str, str], ...] = ( + tuple( + (srv, delegate_auth, requested_single_target) + for srv in allowed_servers + if delegate_server is not None and srv.server_id == delegate_server.server_id + ) + if delegate_auth and requested_single_target + else () + ) + probe_targets = passthrough_targets + delegate_targets + if not probe_targets: return probe_results = await asyncio.gather( - *[_probe_upstream_auth(srv.url or "", forwarded_auth) for srv in passthrough_servers] + *[_probe_upstream_auth(srv.url or "", auth_header) for srv, auth_header, _ in probe_targets] ) - for srv, (probe_status, _) in zip(passthrough_servers, probe_results): + for (srv, _, challenge_server_name), (probe_status, _) in zip(probe_targets, probe_results): if probe_status == 401: # Token is missing or expired: keep pass-through clients on the # protected-resource discovery flow so they re-authorize against # the upstream IdP metadata proxied by LiteLLM. www_authenticate = _get_passthrough_www_authenticate( scope=scope, - server_name=srv.name, + server_name=challenge_server_name, invalid_token=True, ) raise HTTPException( diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py index 32996905166..a983ac3ff48 100644 --- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py +++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py @@ -5428,6 +5428,457 @@ def test_get_forwarded_auth_from_scope_skips_when_no_litellm_key_header(): assert _get_forwarded_auth_from_scope(scope) is None +def _delegate_auth_mcp_server(server_id: str = "delegate-1") -> MCPServer: + return MCPServer( + server_id=server_id, + name="delegate_test", + url="http://upstream:9401/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.oauth2, + delegate_auth_to_upstream=True, + oauth2_flow="authorization_code", + ) + + +def _delegate_scope(headers: list) -> dict: + return { + "type": "http", + "method": "POST", + "path": "/mcp/delegate_test", + "scheme": "http", + "server": ("localhost", 4000), + "headers": headers, + } + + +def _patch_delegate_resolver(server: MCPServer, *resolvable_names: str): + """Patch the admission-parity resolver the delegate probe gates on. Returns + ``server`` only for names admission's ``get_mcp_server_by_name`` would match + (alias / server_name / name); every other name (server_id, access group) yields + None, exactly as the real resolver does.""" + + def _resolve(name, client_ip=None): + return server if name in resolvable_names else None + + return patch( + "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager.get_mcp_server_by_name", + side_effect=_resolve, + ) + + +@pytest.mark.asyncio +async def test_delegate_bad_token_gets_connect_time_401(): + """Regression (LIT-4194): a rejected upstream token on a delegate-auth server + must fail the connect with 401 + ``error="invalid_token"``, not be absorbed + into HTTP 200 + an empty tool list by the tools/list handler. + + Delegate-mode clients send only ``Authorization`` (no ``x-litellm-api-key``), + so ``_get_forwarded_auth_from_scope`` returns None and, before the fix, the + preflight returned early without probing. + """ + from litellm.proxy._experimental.mcp_server.server import ( + _check_passthrough_upstream_auth, + ) + from litellm.proxy._types import UserAPIKeyAuth + + server = _delegate_auth_mcp_server() + scope = _delegate_scope([(b"authorization", b"Bearer bogus-token")]) + + with _patch_delegate_resolver(server, "delegate_test"), patch( + "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers", + new=AsyncMock(return_value=[server]), + ), patch( + "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth", + new=AsyncMock(return_value=(401, 'Bearer realm="upstream", error="invalid_token"')), + ) as probe: + with pytest.raises(HTTPException) as exc_info: + await _check_passthrough_upstream_auth( + scope=scope, + user_api_key_auth=UserAPIKeyAuth(), + mcp_servers=["delegate_test"], + client_ip=None, + ) + + assert exc_info.value.status_code == 401 + challenge = exc_info.value.headers["www-authenticate"] + assert 'error="invalid_token"' in challenge + assert 'resource_metadata="http://localhost:4000/.well-known/oauth-protected-resource/mcp/delegate_test"' in challenge + probe.assert_awaited_once() + probe_url, probe_auth = probe.call_args.args + assert probe_url == "http://upstream:9401/mcp" + assert probe_auth == "Bearer bogus-token" + + +@pytest.mark.asyncio +async def test_delegate_valid_token_passes_preflight(): + """An upstream-accepted token must not be blocked by the delegate preflight.""" + from litellm.proxy._experimental.mcp_server.server import ( + _check_passthrough_upstream_auth, + ) + from litellm.proxy._types import UserAPIKeyAuth + + server = _delegate_auth_mcp_server() + scope = _delegate_scope([(b"authorization", b"Bearer good-token")]) + + with _patch_delegate_resolver(server, "delegate_test"), patch( + "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers", + new=AsyncMock(return_value=[server]), + ), patch( + "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth", + new=AsyncMock(return_value=(200, None)), + ) as probe: + await _check_passthrough_upstream_auth( + scope=scope, + user_api_key_auth=UserAPIKeyAuth(), + mcp_servers=["delegate_test"], + client_ip=None, + ) + + probe.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_delegate_valid_token_forbidden_returns_403(): + """An upstream that accepts the token but forbids the caller (403) must surface + as a bare 403 with no ``WWW-Authenticate`` re-auth hint (a fresh token with the + same scopes would loop), not as an invalid_token challenge.""" + from litellm.proxy._experimental.mcp_server.server import ( + _check_passthrough_upstream_auth, + ) + from litellm.proxy._types import UserAPIKeyAuth + + server = _delegate_auth_mcp_server() + scope = _delegate_scope([(b"authorization", b"Bearer scoped-out-token")]) + + with _patch_delegate_resolver(server, "delegate_test"), patch( + "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers", + new=AsyncMock(return_value=[server]), + ), patch( + "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth", + new=AsyncMock(return_value=(403, None)), + ): + with pytest.raises(HTTPException) as exc_info: + await _check_passthrough_upstream_auth( + scope=scope, + user_api_key_auth=UserAPIKeyAuth(), + mcp_servers=["delegate_test"], + client_ip=None, + ) + + assert exc_info.value.status_code == 403 + assert not (exc_info.value.headers or {}) + + +@pytest.mark.asyncio +async def test_delegate_tokenless_request_not_probed(): + """Tokenless delegate requests are the preemptive challenge's job; the + preflight must not probe upstream with an empty credential.""" + from litellm.proxy._experimental.mcp_server.server import ( + _check_passthrough_upstream_auth, + ) + from litellm.proxy._types import UserAPIKeyAuth + + server = _delegate_auth_mcp_server() + scope = _delegate_scope([(b"content-type", b"application/json")]) + + with _patch_delegate_resolver(server, "delegate_test"), patch( + "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers", + new=AsyncMock(return_value=[server]), + ), patch( + "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth", + new=AsyncMock(return_value=(401, None)), + ) as probe: + await _check_passthrough_upstream_auth( + scope=scope, + user_api_key_auth=UserAPIKeyAuth(), + mcp_servers=["delegate_test"], + client_ip=None, + ) + + probe.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_delegate_preflight_skipped_on_multi_server_routes(): + """The delegate probe is gated to single-server routes so one rejected token + cannot 401 a multi-server aggregate connect (matching the OBO preflight).""" + from litellm.proxy._experimental.mcp_server.server import ( + _check_passthrough_upstream_auth, + ) + from litellm.proxy._types import UserAPIKeyAuth + + servers = [_delegate_auth_mcp_server("delegate-1"), _delegate_auth_mcp_server("delegate-2")] + scope = _delegate_scope([(b"authorization", b"Bearer bogus-token")]) + + with _patch_delegate_resolver(servers[0], "delegate_test", "other_server"), patch( + "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers", + new=AsyncMock(return_value=servers), + ), patch( + "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth", + new=AsyncMock(return_value=(401, None)), + ) as probe: + await _check_passthrough_upstream_auth( + scope=scope, + user_api_key_auth=UserAPIKeyAuth(), + mcp_servers=["delegate_test", "other_server"], + client_ip=None, + ) + + probe.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_bare_authorization_never_probes_passthrough_servers(): + """A bare ``Authorization`` header may be a LiteLLM key (backward-compat), so + only delegate servers (where admission classified it as an upstream token) + may be probed with it; ``is_oauth_passthrough`` servers still require the + unambiguous ``x-litellm-api-key`` + ``Authorization`` pair.""" + from litellm.proxy._experimental.mcp_server.server import ( + _check_passthrough_upstream_auth, + ) + from litellm.proxy._types import UserAPIKeyAuth + + passthrough_server = MCPServer( + server_id="pt-1", + name="pt_server", + url="http://upstream:9402/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.none, + oauth_passthrough=True, + extra_headers=["Authorization"], + ) + scope = _delegate_scope([(b"authorization", b"Bearer ambiguous-token")]) + + with _patch_delegate_resolver(passthrough_server, "pt_server"), patch( + "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers", + new=AsyncMock(return_value=[passthrough_server]), + ), patch( + "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth", + new=AsyncMock(return_value=(401, None)), + ) as probe: + await _check_passthrough_upstream_auth( + scope=scope, + user_api_key_auth=UserAPIKeyAuth(), + mcp_servers=["pt_server"], + client_ip=None, + ) + + probe.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_delegate_not_probed_when_named_only_via_server_id(): + """Security regression (LIT-4194): a delegate server reachable by the requested + name only through its server_id (or an access group) is admitted as a real + LiteLLM key by ``process_mcp_request`` (its ``get_mcp_server_by_name`` misses), + so the bare ``Authorization`` header is that LiteLLM key. The probe must resolve + the target through the SAME resolver and therefore skip it, never forwarding the + key upstream, even though the widened allowed-server set still contains it.""" + from litellm.proxy._experimental.mcp_server.server import ( + _check_passthrough_upstream_auth, + ) + from litellm.proxy._types import UserAPIKeyAuth + + server = _delegate_auth_mcp_server(server_id="delegate-secret-id") + # Admission's resolver matches alias/server_name/name only, never server_id: the + # requested server_id resolves to None here, mirroring the real divergence. + scope = { + "type": "http", + "method": "POST", + "path": "/mcp/delegate-secret-id", + "scheme": "http", + "server": ("localhost", 4000), + "headers": [(b"authorization", b"Bearer sk-litellm-proxy-key")], + } + + with _patch_delegate_resolver(server, "delegate_test"), patch( + "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers", + new=AsyncMock(return_value=[server]), + ), patch( + "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth", + new=AsyncMock(return_value=(401, None)), + ) as probe: + await _check_passthrough_upstream_auth( + scope=scope, + user_api_key_auth=UserAPIKeyAuth(user_id="u1", api_key="hashed-sk"), + mcp_servers=["delegate-secret-id"], + client_ip=None, + ) + + probe.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_delegate_preflight_with_unpatched_probe(): + """Integration across the preflight and the unpatched ``_probe_upstream_auth``, + mocked only at the httpx-client boundary (tests/test_litellm is mocked-only; the + real-network proof lives in the PR's live-proxy evidence). The mock honors the + ``AsyncHTTPHandler.post`` contract by raising ``httpx.HTTPStatusError`` on the + upstream 401, so the production ``except httpx.HTTPStatusError`` branch is the one + exercised. A rejected token surfaces as the connect-time 401 challenge; an + accepted token passes untouched, and the caller's bearer reaches the delegate URL.""" + import httpx + + from litellm.proxy._experimental.mcp_server.server import ( + _check_passthrough_upstream_auth, + ) + from litellm.proxy._types import UserAPIKeyAuth + + accepted = MagicMock() + accepted.status_code = 200 + accepted.headers = {} + rejected = MagicMock() + rejected.status_code = 401 + rejected.headers = {"www-authenticate": 'Bearer realm="stub-upstream", error="invalid_token"'} + + async def respond_by_token(url=None, headers=None, json=None, timeout=None, **kwargs): + if headers.get("Authorization") == "Bearer good-token": + return accepted + raise httpx.HTTPStatusError( + "401 Unauthorized", + request=httpx.Request("POST", url), + response=rejected, + ) + + mock_client = MagicMock() + mock_client.post = AsyncMock(side_effect=respond_by_token) + + server = _delegate_auth_mcp_server() + + with _patch_delegate_resolver(server, "delegate_test"), patch( + "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers", + new=AsyncMock(return_value=[server]), + ), patch( + "litellm.proxy._experimental.mcp_server.server.get_async_httpx_client", + return_value=mock_client, + ): + with pytest.raises(HTTPException) as exc_info: + await _check_passthrough_upstream_auth( + scope=_delegate_scope([(b"authorization", b"Bearer bogus-token")]), + user_api_key_auth=UserAPIKeyAuth(), + mcp_servers=["delegate_test"], + client_ip=None, + ) + + await _check_passthrough_upstream_auth( + scope=_delegate_scope([(b"authorization", b"Bearer good-token")]), + user_api_key_auth=UserAPIKeyAuth(), + mcp_servers=["delegate_test"], + client_ip=None, + ) + + assert exc_info.value.status_code == 401 + challenge = exc_info.value.headers["www-authenticate"] + assert 'error="invalid_token"' in challenge + assert 'resource_metadata="http://localhost:4000/.well-known/oauth-protected-resource/mcp/delegate_test"' in challenge + probed_urls = [call.kwargs["url"] for call in mock_client.post.await_args_list] + assert probed_urls == ["http://upstream:9401/mcp", "http://upstream:9401/mcp"] + + +@pytest.mark.asyncio +async def test_delegate_challenge_echoes_requested_alias(): + """An alias-routed delegate request must be probed, and the challenge must echo + the requested alias (not the canonical server name) so the resource_metadata + URL matches what the tokenless preemptive challenge emits for the same route.""" + from litellm.proxy._experimental.mcp_server.server import ( + _check_passthrough_upstream_auth, + ) + from litellm.proxy._types import UserAPIKeyAuth + + server = _delegate_auth_mcp_server().model_copy(update={"alias": "dt-alias"}) + scope = { + "type": "http", + "method": "POST", + "path": "/mcp/dt-alias", + "scheme": "http", + "server": ("localhost", 4000), + "headers": [(b"authorization", b"Bearer bogus-token")], + } + + with _patch_delegate_resolver(server, "dt-alias"), patch( + "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers", + new=AsyncMock(return_value=[server]), + ), patch( + "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth", + new=AsyncMock(return_value=(401, 'Bearer error="invalid_token"')), + ): + with pytest.raises(HTTPException) as exc_info: + await _check_passthrough_upstream_auth( + scope=scope, + user_api_key_auth=UserAPIKeyAuth(), + mcp_servers=["dt-alias"], + client_ip=None, + ) + + challenge = exc_info.value.headers["www-authenticate"] + assert 'error="invalid_token"' in challenge + assert 'resource_metadata="http://localhost:4000/.well-known/oauth-protected-resource/mcp/dt-alias"' in challenge + + +@pytest.mark.asyncio +async def test_delegate_probe_not_fanned_out_to_access_group_members(): + """A single access-group name passes the one-target route gate but must not fan + the delegate probe out to group-expanded member servers; the group name resolves + to no server under admission's resolver, so no probe fires.""" + from litellm.proxy._experimental.mcp_server.server import ( + _check_passthrough_upstream_auth, + ) + from litellm.proxy._types import UserAPIKeyAuth + + group_member = _delegate_auth_mcp_server() + + with _patch_delegate_resolver(group_member, "delegate_test"), patch( + "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers", + new=AsyncMock(return_value=[group_member]), + ), patch( + "litellm.proxy._experimental.mcp_server.server._probe_upstream_auth", + new=AsyncMock(return_value=(401, None)), + ) as probe: + await _check_passthrough_upstream_auth( + scope=_delegate_scope([(b"authorization", b"Bearer bogus-token")]), + user_api_key_auth=UserAPIKeyAuth(), + mcp_servers=["prod_tools_group"], + client_ip=None, + ) + + probe.assert_not_awaited() + + +def test_is_delegate_upstream_probe_target_fails_closed_on_m2m_shape(): + """An unstamped M2M-shape row (null ``oauth2_flow`` + client credentials) + resolves to ``client_credentials`` and must not be probed with the caller's + bearer; its stored client credentials drive egress instead.""" + from litellm.proxy._experimental.mcp_server.server import ( + _is_delegate_upstream_probe_target, + ) + + assert _is_delegate_upstream_probe_target(_delegate_auth_mcp_server()) is True + + m2m_shape = MCPServer( + server_id="delegate-m2m", + name="delegate_m2m", + url="http://upstream:9401/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.oauth2, + delegate_auth_to_upstream=True, + oauth2_flow=None, + token_url="http://idp:9000/token", + client_id="client", + client_secret="secret", + ) + assert _is_delegate_upstream_probe_target(m2m_shape) is False + + non_delegate = MCPServer( + server_id="oauth2-plain", + name="oauth2_plain", + url="http://upstream:9401/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.oauth2, + oauth2_flow="authorization_code", + ) + assert _is_delegate_upstream_probe_target(non_delegate) is False + + @pytest.mark.asyncio async def test_create_mcp_client_sampling_disabled_by_default(): """Sampling callback must be None when allow_sampling is not set (default False)."""