scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-11 22:51:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 154

Merge pull request #32741 from BerriAI/litellm_lit4194_delegate_invalid_token

Browse source 
fix(mcp): surface rejected delegate-auth upstream tokens as connect-time 401
This commit is contained in:
tin-berri 2026-07-13 10:39:16 -07:00 • committed by GitHub
commit 48fd1240a9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 542 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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(

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

@ -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)."""