diff --git a/tests/integration/mcp/test_mcp_oauth_flows.py b/tests/integration/mcp/test_mcp_oauth_flows.py
index 0bd299f3bbd..4fe2821d7aa 100644
--- a/tests/integration/mcp/test_mcp_oauth_flows.py
+++ b/tests/integration/mcp/test_mcp_oauth_flows.py
@@ -1,21 +1,24 @@
 import base64
 import hashlib
 import json
+import os
 import re
 import secrets
 import textwrap
 import time
 import uuid
-from collections.abc import Iterator
+from collections.abc import Iterator, Mapping
 from contextlib import contextmanager
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Final
-from urllib.parse import parse_qs, urlsplit
+from urllib.parse import parse_qs, urljoin, urlsplit
 
 import httpx
 import jwt
 import pytest
+import yaml
+from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey, generate_private_key
 from integration._support.client import Gateway, eventually, gateway_from_environment
 from integration._support.database import read_rows
 from integration._support.mcp import (
@@ -38,6 +41,7 @@ from integration._support.mcp_grants import create_toolset
 from integration._support.oauth_server import AuthorizationServer, oauth_server
 from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, Wire, wire_server
+from pydantic import TypeAdapter
 
 ADD: Final = {"a": 2, "b": 3}
 CLIENT_REDIRECT: Final = "http://127.0.0.1:9/cb"
@@ -48,6 +52,33 @@ def _base(gateway: Gateway) -> str:
     return str(gateway.client.base_url).rstrip("/")
 
 
+def _response_object(response: httpx.Response) -> dict[str, object]:
+    return TypeAdapter(dict[str, object]).validate_python(response.json())
+
+
+def _open_aliases() -> frozenset[str]:
+    rows: Final = read_rows('SELECT alias FROM "LiteLLM_MCPServerTable" WHERE allow_all_keys', ())
+    return frozenset(str(row["alias"]) for row in rows)
+
+
+def _without_foreign_open_servers(tools: tuple[str, ...], open_aliases: frozenset[str]) -> set[str]:
+    prefixes: Final = tuple(f"{alias}-" for alias in open_aliases)
+    return {tool for tool in tools if not tool.startswith(prefixes)}
+
+
+def _assert_upstream_call(call: Mapping[str, object], name: str, arguments: Mapping[str, object]) -> None:
+    body: Final = call["body"]
+    assert isinstance(body, Mapping), call
+    params: Final = body["params"]
+    assert isinstance(params, Mapping), call
+    assert set(params) == {"name", "arguments", "_meta"}, call
+    assert params["name"] == name, call
+    assert params["arguments"] == arguments, call
+    metadata: Final = params["_meta"]
+    assert isinstance(metadata, Mapping), call
+    assert set(metadata) == {"progressToken"}, call
+
+
 def _authorizations(peer: McpPeer) -> tuple[bytes | None, ...]:
     return tuple(
         value if isinstance(value := call["headers"].get(b"authorization"), bytes) else None
@@ -246,9 +277,14 @@ class _Pkce:
 
 
 def _authorize_through_gateway(
-    gateway: Gateway, auth: AuthorizationServer, alias: str, key: str, client_id: str, pkce: _Pkce,
+    gateway: Gateway,
+    auth: AuthorizationServer,
+    alias: str,
+    key: str,
+    client_id: str,
+    pkce: _Pkce,
     scope: str = "tools.call",
-) -> str:
+) -> tuple[str, str]:
     started: Final = gateway.client.get(
         f"/{alias}/authorize",
         params={
@@ -279,7 +315,8 @@ def _authorize_through_gateway(
     final: Final = parse_qs(urlsplit(returned.headers["location"]).query)
     assert returned.headers["location"].startswith(CLIENT_REDIRECT)
     assert final["state"] == ["client-state"], final
-    return final["code"][0]
+    upstream_code: Final = parse_qs(urlsplit(callback).query)["code"][0]
+    return final["code"][0], upstream_code
 
 
 def _redeem(gateway: Gateway, alias: str, key: str, client_id: str, code: str, pkce: _Pkce) -> httpx.Response:
@@ -296,6 +333,187 @@ def _redeem(gateway: Gateway, alias: str, key: str, client_id: str, code: str, p
     )
 
 
+def _expected_authorization_header(auth_method: str, secret: str) -> str | None:
+    if auth_method == "client_secret_post":
+        return None
+    return "Basic " + base64.b64encode(f"ac-client:{secret}".encode()).decode()
+
+
+def _assert_upstream_token_request(
+    request: Request, auth_method: str, secret: str, expected_form: Mapping[str, str]
+) -> None:
+    """One scripted-IdP token request must carry the gateway's exact form and, for
+    client_secret_basic, the stored credentials in a Basic header, and never the
+    caller's placeholder secret."""
+    form: Final = {name: values[0] for name, values in parse_qs(request.body.decode()).items()}
+    assert form == dict(expected_form), form
+    assert request.headers.get("authorization") == _expected_authorization_header(auth_method, secret), request.headers
+    assert "dummy" not in request.body.decode() and "dummy" not in repr(request.headers), request
+
+
+def _upstream_token_requests(auth: AuthorizationServer) -> tuple[Request, ...]:
+    return tuple(item for item in auth.drain() if item.target.startswith("/token"))
+
+
+@pytest.mark.parametrize("auth_method", ("client_secret_post", "client_secret_basic"))
+def test_upstream_token_requests_carry_the_gateway_callback_and_server_credentials(
+    gateway: Gateway, auth_method: str
+) -> None:
+    with mcp_peer() as peer, oauth_server() as auth, gateway.scenario() as scenario:
+        alias: Final = "rt3" + uuid.uuid4().hex[:10]
+        secret: Final = "rt3-secret-" + uuid.uuid4().hex
+        identity: Final = _register_oauth(
+            scenario,
+            peer,
+            auth,
+            alias,
+            auth_type="oauth2",
+            oauth2_flow="authorization_code",
+            credentials={
+                "client_id": "ac-client",
+                "client_secret": secret,
+                "scopes": ["tools.call"],
+                "token_endpoint_auth_method": auth_method,
+            },
+        )
+        # A second oauth2 server keeps the root /token path honest: with two registered
+        # oauth2 servers the single-server root fallback cannot resolve, so the root
+        # spelling works only when client_id itself names the server.
+        _register_oauth(
+            scenario,
+            peer,
+            auth,
+            "rt3o" + uuid.uuid4().hex[:9],
+            auth_type="oauth2",
+            oauth2_flow="authorization_code",
+            credentials={"client_id": "ac-other", "client_secret": "other-secret", "scopes": ["tools.call"]},
+        )
+        persisted: Final = gateway.request("GET", f"/v1/mcp/server/{identity}")
+        assert persisted.status_code == 200, persisted.text
+        persisted_body: Final = _response_object(persisted)
+        assert persisted_body["credentials"] == {"scopes": ["tools.call"]}, persisted.text
+        owner: Final = scenario.key(object_permission={"mcp_servers": [identity]})
+        registered: Final = gateway.client.post(
+            f"/{alias}/register", json={"redirect_uris": [CLIENT_REDIRECT], "client_name": "integration"}
+        )
+        assert registered.status_code in (200, 201), registered.text
+        client_id: Final = str(_response_object(registered)["client_id"])
+
+        def expected_code_form(code: str, verifier: str) -> dict[str, str]:
+            return {
+                "grant_type": "authorization_code",
+                "code": code,
+                "redirect_uri": f"{_base(gateway)}/callback",
+                "code_verifier": verifier,
+                **({"client_id": "ac-client", "client_secret": secret} if auth_method == "client_secret_post" else {}),
+            }
+
+        def expected_refresh_form(refresh_token: str) -> dict[str, str]:
+            return {
+                "grant_type": "refresh_token",
+                "refresh_token": refresh_token,
+                "scope": "tools.call",
+                **({"client_id": "ac-client", "client_secret": secret} if auth_method == "client_secret_post" else {}),
+            }
+
+        pkce: Final = _Pkce(secrets.token_urlsafe(32))
+        code, upstream_code = _authorize_through_gateway(gateway, auth, alias, owner, client_id, pkce)
+        authorize_requests: Final = tuple(item for item in auth.drain() if item.target.startswith("/authorize"))
+        assert len(authorize_requests) == 1, authorize_requests
+        authorize_query: Final = parse_qs(urlsplit(authorize_requests[0].target).query)
+        assert authorize_query["redirect_uri"] == [f"{_base(gateway)}/callback"], authorize_query
+
+        redeemed: Final = gateway.client.post(
+            f"/{alias}/token",
+            headers={"x-litellm-api-key": owner},
+            data={
+                "grant_type": "authorization_code",
+                "code": code,
+                "code_verifier": pkce.verifier,
+                "client_id": client_id,
+                "client_secret": "dummy",
+                "redirect_uri": CLIENT_REDIRECT,
+            },
+        )
+        assert redeemed.status_code == 200, redeemed.text
+        issued: Final = _response_object(redeemed)
+        initial_token_requests: Final = _upstream_token_requests(auth)
+        assert len(initial_token_requests) == 1, initial_token_requests
+        _assert_upstream_token_request(
+            initial_token_requests[0], auth_method, secret, expected_code_form(upstream_code, pkce.verifier)
+        )
+
+        upstream_refresh_token: Final = str(issued["refresh_token"])
+        refreshed: Final = gateway.client.post(
+            f"/{alias}/token",
+            headers={"x-litellm-api-key": owner},
+            data={
+                "grant_type": "refresh_token",
+                "refresh_token": upstream_refresh_token,
+                "client_id": client_id,
+                "client_secret": "dummy",
+                "scope": "tools.call",
+            },
+        )
+        assert refreshed.status_code == 200, refreshed.text
+        assert _response_object(refreshed)["access_token"] != issued["access_token"], refreshed.text
+        refresh_requests: Final = _upstream_token_requests(auth)
+        assert len(refresh_requests) == 1, refresh_requests
+        _assert_upstream_token_request(
+            refresh_requests[0], auth_method, secret, expected_refresh_form(upstream_refresh_token)
+        )
+
+        pkce_root: Final = _Pkce(secrets.token_urlsafe(32))
+        root_code, root_upstream_code = _authorize_through_gateway(gateway, auth, alias, owner, client_id, pkce_root)
+        root_authorize_requests: Final = tuple(item for item in auth.drain() if item.target.startswith("/authorize"))
+        assert len(root_authorize_requests) == 1, root_authorize_requests
+        root_authorize_query: Final = parse_qs(urlsplit(root_authorize_requests[0].target).query)
+        assert root_authorize_query["redirect_uri"] == [f"{_base(gateway)}/callback"], root_authorize_query
+
+        root_redeemed: Final = gateway.client.post(
+            "/token",
+            headers={"x-litellm-api-key": owner},
+            data={
+                "grant_type": "authorization_code",
+                "code": root_code,
+                "code_verifier": pkce_root.verifier,
+                "client_id": alias,
+                "client_secret": "dummy",
+                "redirect_uri": CLIENT_REDIRECT,
+            },
+        )
+        assert root_redeemed.status_code == 200, root_redeemed.text
+        root_issued: Final = _response_object(root_redeemed)
+        root_token_requests: Final = _upstream_token_requests(auth)
+        assert len(root_token_requests) == 1, root_token_requests
+        _assert_upstream_token_request(
+            root_token_requests[0],
+            auth_method,
+            secret,
+            expected_code_form(root_upstream_code, pkce_root.verifier),
+        )
+
+        root_refresh_token: Final = str(root_issued["refresh_token"])
+        root_refreshed: Final = gateway.client.post(
+            "/token",
+            headers={"x-litellm-api-key": owner},
+            data={
+                "grant_type": "refresh_token",
+                "refresh_token": root_refresh_token,
+                "client_id": alias,
+                "client_secret": "dummy",
+                "scope": "tools.call",
+            },
+        )
+        assert root_refreshed.status_code == 200, root_refreshed.text
+        assert _response_object(root_refreshed)["access_token"] != root_issued["access_token"], root_refreshed.text
+        root_refresh_requests: Final = _upstream_token_requests(auth)
+        assert len(root_refresh_requests) == 1, root_refresh_requests
+        _assert_upstream_token_request(
+            root_refresh_requests[0], auth_method, secret, expected_refresh_form(root_refresh_token)
+        )
+
+
 def test_per_user_authorization_code_with_pkce_binds_the_token_to_the_authorizing_user(gateway: Gateway) -> None:
     with mcp_peer() as peer, oauth_server() as auth, gateway.scenario() as scenario:
         alias: Final = "ac" + uuid.uuid4().hex[:8]
@@ -321,13 +539,13 @@ def test_per_user_authorization_code_with_pkce_binds_the_token_to_the_authorizin
             f"/{alias}/register", json={"redirect_uris": [CLIENT_REDIRECT], "client_name": "integration"}
         )
         assert registered.status_code in (200, 201), registered.text
-        client_id: Final = registered.json()["client_id"]
+        client_id: Final = str(_response_object(registered)["client_id"])
         pkce: Final = _Pkce(secrets.token_urlsafe(32))
-        code: Final = _authorize_through_gateway(gateway, auth, alias, owner, client_id, pkce)
+        code, _ = _authorize_through_gateway(gateway, auth, alias, owner, client_id, pkce)
         wrong_verifier: Final = _redeem(gateway, alias, owner, client_id, code, _Pkce("wrong-" + pkce.verifier))
         assert wrong_verifier.status_code == 400, wrong_verifier.text
         assert tool_calls(peer.drain()) == ()
-        code2: Final = _authorize_through_gateway(gateway, auth, alias, owner, client_id, pkce)
+        code2, _ = _authorize_through_gateway(gateway, auth, alias, owner, client_id, pkce)
         redeemed: Final = _redeem(gateway, alias, owner, client_id, code2, pkce)
         assert redeemed.status_code == 200, redeemed.text
         issued: Final = redeemed.json()
@@ -408,12 +626,14 @@ def _ui_session_cookie(gateway: Gateway, user_id: str) -> dict[str, str]:
     return {"token": jwt.encode(claims, gateway.key, algorithm="HS256")}
 
 
-def _gateway_session_bearer(gateway: Gateway, user_id: str, resource: str | None = None) -> str:
+def _gateway_session_sign_in(
+    gateway: Gateway, user_id: str, resource: str | None = None
+) -> tuple[dict[str, object], str]:
     registered: Final = gateway.client.post(
         "/register", json={"redirect_uris": [CLIENT_REDIRECT], "client_name": "integration"}
     )
     assert registered.status_code in (200, 201), registered.text
-    client_id: Final = registered.json()["client_id"]
+    client_id: Final = str(_response_object(registered)["client_id"])
     pkce: Final = _Pkce(secrets.token_urlsafe(48))
     cookies: Final = _ui_session_cookie(gateway, user_id)
     started: Final = gateway.client.get(
@@ -448,7 +668,461 @@ def _gateway_session_bearer(gateway: Gateway, user_id: str, resource: str | None
         },
     )
     assert issued.status_code == 200, issued.text
-    return _issued_token(issued.json())
+    return _response_object(issued), client_id
+
+
+def _gateway_session_bearer(gateway: Gateway, user_id: str, resource: str | None = None) -> str:
+    issued, _ = _gateway_session_sign_in(gateway, user_id, resource)
+    return _issued_token(issued)
+
+
+def test_spec_client_discovers_the_aggregate_authorization_server_and_signs_in_to_a_session_bearer(
+    gateway: Gateway,
+) -> None:
+    base: Final = _base(gateway)
+    protected: Final = gateway.client.get("/.well-known/oauth-protected-resource/mcp")
+    assert protected.status_code == 200, protected.text
+    assert _response_object(protected) == {
+        "authorization_servers": [f"{base}/mcp"],
+        "resource": f"{base}/mcp",
+        "scopes_supported": [],
+    }, protected.text
+    authorization_server: Final = gateway.client.get("/.well-known/oauth-authorization-server/mcp")
+    assert authorization_server.status_code == 200, authorization_server.text
+    authorization_metadata: Final = _response_object(authorization_server)
+    assert authorization_metadata == {
+        "issuer": f"{base}/mcp",
+        "authorization_endpoint": f"{base}/authorize/mcp-session",
+        "token_endpoint": f"{base}/token",
+        "introspection_endpoint": f"{base}/introspect",
+        "registration_endpoint": f"{base}/register",
+        "response_types_supported": ["code"],
+        "scopes_supported": [],
+        "grant_types_supported": ["authorization_code", "refresh_token"],
+        "code_challenge_methods_supported": ["S256"],
+        "token_endpoint_auth_methods_supported": ["none", "client_secret_post"],
+    }, authorization_server.text
+    registration_endpoint: Final = urlsplit(str(authorization_metadata["registration_endpoint"])).path
+    authorization_endpoint: Final = urlsplit(str(authorization_metadata["authorization_endpoint"])).path
+    token_endpoint: Final = urlsplit(str(authorization_metadata["token_endpoint"])).path
+    with mcp_peer() as peer, gateway.scenario() as scenario:
+        alias: Final = "rt5" + uuid.uuid4().hex[:10]
+        identity: Final = register_mcp(scenario, peer, alias)
+        member: Final = scenario.member(scenario.team(object_permission={"mcp_servers": [identity]}))
+        registered: Final = gateway.client.post(
+            registration_endpoint, json={"redirect_uris": [CLIENT_REDIRECT], "client_name": "integration"}
+        )
+        assert registered.status_code in (200, 201), registered.text
+        client_id: Final = str(_response_object(registered)["client_id"])
+        pkce: Final = _Pkce(secrets.token_urlsafe(48))
+        cookies: Final = _ui_session_cookie(gateway, member)
+        started: Final = gateway.client.get(
+            authorization_endpoint,
+            params={
+                "client_id": client_id,
+                "redirect_uri": CLIENT_REDIRECT,
+                "response_type": "code",
+                "state": "rt5-state",
+                "code_challenge": pkce.challenge,
+                "code_challenge_method": "S256",
+                "resource": f"{base}/mcp",
+            },
+            cookies=cookies,
+        )
+        assert started.status_code == 303, started.text
+        connect_location: Final = urlsplit(started.headers["location"])
+        assert (
+            connect_location.scheme + "://" + connect_location.netloc + connect_location.path == f"{base}/ui/connect"
+        ), started.headers["location"]
+        connect_flow: Final = parse_qs(connect_location.query)["connect_flow"][0]
+        flow_cookie: Final = f"mcp_connect_flow_{connect_flow}"
+        assert flow_cookie in started.cookies, started.headers
+        flow_cookies: Final = {**cookies, flow_cookie: started.cookies[flow_cookie]}
+        described: Final = gateway.client.get("/authorize/flow", params={"flow": connect_flow}, cookies=flow_cookies)
+        assert described.status_code == 200, described.text
+        assert _response_object(described) == {
+            "state": "unscoped",
+            "client_origin": "http://127.0.0.1:9",
+            "server_id": None,
+            "server_name": None,
+            "connected": None,
+        }, described.text
+        completed: Final = gateway.client.post("/authorize/complete", data={"flow": connect_flow}, cookies=flow_cookies)
+        assert completed.status_code == 303, completed.text
+        callback: Final = parse_qs(urlsplit(completed.headers["location"]).query)
+        assert callback["state"] == ["rt5-state"], completed.headers["location"]
+        assert set(callback) == {"code", "state"}, callback
+        token: Final = gateway.client.post(
+            token_endpoint,
+            data={
+                "grant_type": "authorization_code",
+                "code": callback["code"][0],
+                "redirect_uri": CLIENT_REDIRECT,
+                "client_id": client_id,
+                "code_verifier": pkce.verifier,
+            },
+        )
+        assert token.status_code == 200, token.text
+        token_body: Final = _response_object(token)
+        assert set(token_body) == {"access_token", "token_type", "expires_in", "refresh_token"}, token.text
+        assert token_body["token_type"] == "Bearer", token.text
+        assert isinstance(token_body["expires_in"], int) and token_body["expires_in"] > 0, token.text
+        assert str(token_body["access_token"]).startswith("llm_session_"), token.text
+        assert str(token_body["refresh_token"]).startswith("llm_srefresh_"), token.text
+        bearer: Final = str(token_body["access_token"])
+        initialized: Final = gateway.client.post(
+            "/mcp",
+            headers={"Authorization": f"Bearer {bearer}", **ACCEPT},
+            json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": dict(INITIALIZE)},
+        )
+        assert initialized.status_code == 200, initialized.text
+        open_before: Final = _open_aliases()
+        listed_response: Final = gateway.client.post(
+            "/mcp",
+            headers={"Authorization": f"Bearer {bearer}", **ACCEPT},
+            json={"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}},
+        )
+        assert listed_response.status_code == 200, listed_response.text
+        listed: Final = _outcome_from_rpc(listed_response)
+        assert listed.ok, listed.raw
+        open_after: Final = open_before | _open_aliases()
+        assert _without_foreign_open_servers(listed.tools, open_after) == {
+            f"{alias}-add",
+            f"{alias}-multiply",
+            f"{alias}-fail",
+        }, listed.tools
+        peer.drain()
+        called_response: Final = gateway.client.post(
+            "/mcp",
+            headers={"Authorization": f"Bearer {bearer}", **ACCEPT},
+            json={
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "tools/call",
+                "params": {"name": f"{alias}-add", "arguments": {"a": 20, "b": 22}},
+            },
+        )
+        assert called_response.status_code == 200, called_response.text
+        called: Final = _outcome_from_rpc(called_response)
+        assert called.ok and called.text == "42", called.raw
+        calls: Final = tool_calls(peer.drain())
+        assert len(calls) == 1, calls
+        _assert_upstream_call(calls[0], "add", {"a": 20, "b": 22})
+
+
+def test_per_server_resource_scopes_the_session_bearer_to_the_named_server(gateway: Gateway) -> None:
+    base: Final = _base(gateway)
+    with mcp_peer() as peer_a, mcp_peer() as peer_b, gateway.scenario() as scenario:
+        alias_a: Final = "sca" + uuid.uuid4().hex[:10]
+        alias_b: Final = "scb" + uuid.uuid4().hex[:10]
+        server_a: Final = register_mcp(scenario, peer_a, alias_a)
+        server_b: Final = register_mcp(scenario, peer_b, alias_b)
+        member: Final = scenario.member(scenario.team(object_permission={"mcp_servers": [server_a, server_b]}))
+        registered: Final = gateway.client.post(
+            "/register", json={"redirect_uris": [CLIENT_REDIRECT], "client_name": "integration"}
+        )
+        assert registered.status_code in (200, 201), registered.text
+        client_id: Final = str(_response_object(registered)["client_id"])
+        pkce: Final = _Pkce(secrets.token_urlsafe(48))
+        cookies: Final = _ui_session_cookie(gateway, member)
+        started: Final = gateway.client.get(
+            "/authorize/mcp-session",
+            params={
+                "client_id": client_id,
+                "redirect_uri": CLIENT_REDIRECT,
+                "response_type": "code",
+                "state": "sca-state",
+                "code_challenge": pkce.challenge,
+                "code_challenge_method": "S256",
+                "resource": f"{base}/mcp/{alias_a}",
+            },
+            cookies=cookies,
+        )
+        assert started.status_code == 303, started.text
+        connect_location: Final = urlsplit(started.headers["location"])
+        assert (
+            connect_location.scheme + "://" + connect_location.netloc + connect_location.path == f"{base}/ui/connect"
+        ), started.headers["location"]
+        connect_flow: Final = parse_qs(connect_location.query)["connect_flow"][0]
+        flow_cookie: Final = f"mcp_connect_flow_{connect_flow}"
+        assert flow_cookie in started.cookies, started.headers
+        flow_cookies: Final = {**cookies, flow_cookie: started.cookies[flow_cookie]}
+        described: Final = gateway.client.get("/authorize/flow", params={"flow": connect_flow}, cookies=flow_cookies)
+        assert described.status_code == 200, described.text
+        assert _response_object(described) == {
+            "state": "m2m",
+            "client_origin": "http://127.0.0.1:9",
+            "server_id": server_a,
+            "server_name": alias_a,
+            "connected": True,
+        }, described.text
+        completed: Final = gateway.client.post("/authorize/complete", data={"flow": connect_flow}, cookies=flow_cookies)
+        assert completed.status_code == 303, completed.text
+        callback: Final = parse_qs(urlsplit(completed.headers["location"]).query)
+        assert callback["state"] == ["sca-state"], completed.headers["location"]
+        assert set(callback) == {"code", "state"}, callback
+        token: Final = gateway.client.post(
+            "/token",
+            data={
+                "grant_type": "authorization_code",
+                "code": callback["code"][0],
+                "redirect_uri": CLIENT_REDIRECT,
+                "client_id": client_id,
+                "code_verifier": pkce.verifier,
+            },
+        )
+        assert token.status_code == 200, token.text
+        token_body: Final = _response_object(token)
+        assert set(token_body) == {"access_token", "token_type", "expires_in", "refresh_token"}, token.text
+        bearer: Final = str(token_body["access_token"])
+        assert bearer.startswith("llm_session_"), token.text
+        initialized: Final = gateway.client.post(
+            "/mcp",
+            headers={"Authorization": f"Bearer {bearer}", **ACCEPT},
+            json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": dict(INITIALIZE)},
+        )
+        assert initialized.status_code == 200, initialized.text
+        open_before: Final = _open_aliases()
+        listed_response: Final = gateway.client.post(
+            "/mcp",
+            headers={"Authorization": f"Bearer {bearer}", **ACCEPT},
+            json={"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}},
+        )
+        assert listed_response.status_code == 200, listed_response.text
+        listed: Final = _outcome_from_rpc(listed_response)
+        assert listed.ok, listed.raw
+        open_after: Final = open_before | _open_aliases()
+        assert _without_foreign_open_servers(listed.tools, open_after) == {
+            f"{alias_a}-add",
+            f"{alias_a}-multiply",
+            f"{alias_a}-fail",
+        }, listed.tools
+        peer_a.drain()
+        peer_b.drain()
+        called_response: Final = gateway.client.post(
+            "/mcp",
+            headers={"Authorization": f"Bearer {bearer}", **ACCEPT},
+            json={
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "tools/call",
+                "params": {"name": f"{alias_a}-add", "arguments": {"a": 20, "b": 22}},
+            },
+        )
+        assert called_response.status_code == 200, called_response.text
+        called: Final = _outcome_from_rpc(called_response)
+        assert called.ok and called.text == "42", called.raw
+        calls: Final = tool_calls(peer_a.drain())
+        assert len(calls) == 1, calls
+        _assert_upstream_call(calls[0], "add", {"a": 20, "b": 22})
+        assert tool_calls(peer_b.drain()) == ()
+        refused_response: Final = gateway.client.post(
+            "/mcp",
+            headers={"Authorization": f"Bearer {bearer}", **ACCEPT},
+            json={
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "tools/call",
+                "params": {"name": f"{alias_b}-add", "arguments": ADD},
+            },
+        )
+        refused: Final = _outcome_from_rpc(refused_response)
+        assert not refused.ok, refused.raw
+        assert tool_calls(peer_b.drain()) == ()
+
+
+def test_rotated_session_refresh_token_cannot_be_replayed_after_revocation(gateway: Gateway) -> None:
+    with mcp_peer() as peer, gateway.scenario() as scenario:
+        alias: Final = "rt6" + uuid.uuid4().hex[:10]
+        identity: Final = register_mcp(scenario, peer, alias)
+        member: Final = scenario.member(scenario.team(object_permission={"mcp_servers": [identity]}))
+        issued, client_id = _gateway_session_sign_in(gateway, member)
+        r1: Final = str(issued["refresh_token"])
+        a1: Final = str(issued["access_token"])
+        assert r1.startswith("llm_srefresh_"), r1
+        assert a1.startswith("llm_session_"), a1
+        rotated: Final = gateway.client.post(
+            "/token",
+            data={"grant_type": "refresh_token", "refresh_token": r1, "client_id": client_id},
+        )
+        assert rotated.status_code == 200, rotated.text
+        rotated_body: Final = _response_object(rotated)
+        assert set(rotated_body) == {"access_token", "token_type", "expires_in", "refresh_token"}, rotated.text
+        r2: Final = str(rotated_body["refresh_token"])
+        a2: Final = str(rotated_body["access_token"])
+        assert r2 != r1 and r2.startswith("llm_srefresh_"), rotated.text
+        assert a2 != a1 and a2.startswith("llm_session_"), rotated.text
+        replayed: Final = gateway.client.post(
+            "/token",
+            data={"grant_type": "refresh_token", "refresh_token": r1, "client_id": client_id},
+        )
+        assert replayed.status_code == 400, replayed.text
+        assert _response_object(replayed) == {
+            "error": "invalid_grant",
+            "error_description": "the refresh token was already used",
+        }, replayed.text
+        revoked: Final = gateway.client.post("/revoke", data={"token": r2, "client_id": client_id})
+        assert revoked.status_code == 200, revoked.text
+        assert _response_object(revoked) == {}, revoked.text
+        revoked_again: Final = gateway.client.post("/revoke", data={"token": r2, "client_id": client_id})
+        assert revoked_again.status_code == 200, revoked_again.text
+        assert _response_object(revoked_again) == {}, revoked_again.text
+        after_revoke: Final = gateway.client.post(
+            "/token",
+            data={"grant_type": "refresh_token", "refresh_token": r2, "client_id": client_id},
+        )
+        assert after_revoke.status_code == 400, after_revoke.text
+        assert _response_object(after_revoke) == {
+            "error": "invalid_grant",
+            "error_description": "the refresh token was revoked",
+        }, after_revoke.text
+        unknown_client: Final = gateway.client.post(
+            "/revoke", data={"token": r2, "client_id": "dcr-unknown-" + uuid.uuid4().hex}
+        )
+        assert unknown_client.status_code == 401, unknown_client.text
+        assert _response_object(unknown_client) == {
+            "error": "invalid_client",
+            "error_description": "unknown or malformed client_id",
+        }, unknown_client.text
+        open_before: Final = _open_aliases()
+        listed_response: Final = gateway.client.post(
+            "/mcp",
+            headers={"Authorization": f"Bearer {a2}", **ACCEPT},
+            json={"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}},
+        )
+        assert listed_response.status_code == 200, listed_response.text
+        listed: Final = _outcome_from_rpc(listed_response)
+        assert listed.ok, listed.raw
+        open_after: Final = open_before | _open_aliases()
+        assert _without_foreign_open_servers(listed.tools, open_after) == {
+            f"{alias}-add",
+            f"{alias}-multiply",
+            f"{alias}-fail",
+        }, listed.tools
+
+
+def _consent_form_fields(page: str) -> dict[str, str]:
+    return dict(re.findall(r'<input type="hidden" name="([^"]+)" value="([^"]*)">', page))
+
+
+def test_lite_logout_revokes_the_cli_refresh_token_sent_the_way_the_cli_sends_it(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        team_id: Final = scenario.team()
+        member: Final = scenario.member(team_id)
+        discovered: Final = gateway.client.get("/.well-known/litellm-cli-auth")
+        assert discovered.status_code == 200, discovered.text
+        contract: Final = _response_object(discovered)
+        base: Final = _base(gateway)
+        assert (contract["resource"], contract["token_endpoint"], contract["revocation_endpoint"]) == (
+            base,
+            f"{base}/token",
+            f"{base}/revoke",
+        ), discovered.text
+        resource: Final = str(contract["resource"])
+        registered: Final = gateway.client.post(
+            str(contract["registration_endpoint"]),
+            json={
+                "client_name": "LiteLLM CLI",
+                "redirect_uris": [CLIENT_REDIRECT],
+                "grant_types": ["authorization_code", "refresh_token"],
+                "response_types": ["code"],
+                "token_endpoint_auth_method": "none",
+            },
+        )
+        assert registered.status_code in (200, 201), registered.text
+        client_id: Final = str(_response_object(registered)["client_id"])
+        pkce: Final = _Pkce(secrets.token_urlsafe(64))
+        cookies: Final = _ui_session_cookie(gateway, member)
+        consent: Final = gateway.client.get(
+            str(contract["authorization_endpoint"]),
+            params={
+                "response_type": "code",
+                "client_id": client_id,
+                "redirect_uri": CLIENT_REDIRECT,
+                "state": "lite-login-state",
+                "code_challenge": pkce.challenge,
+                "code_challenge_method": "S256",
+                "resource": resource,
+            },
+            cookies=cookies,
+        )
+        assert consent.status_code == 200, consent.text
+        consent_fields: Final = _consent_form_fields(consent.text)
+        assert set(consent_fields) == {"flow", "team_id"} and consent_fields["team_id"] == team_id, consent.text
+        approved: Final = gateway.client.post(
+            "/authorize/complete",
+            data={**consent_fields, "decision": "approve"},
+            cookies={**cookies, **dict(consent.cookies)},
+        )
+        assert approved.status_code == 303, approved.text
+        callback: Final = parse_qs(urlsplit(approved.headers["location"]).query)
+        assert callback["state"] == ["lite-login-state"], approved.headers["location"]
+        issued: Final = gateway.client.post(
+            str(contract["token_endpoint"]),
+            data={
+                "grant_type": "authorization_code",
+                "code": callback["code"][0],
+                "redirect_uri": CLIENT_REDIRECT,
+                "client_id": client_id,
+                "code_verifier": pkce.verifier,
+                "resource": resource,
+            },
+        )
+        assert issued.status_code == 200, issued.text
+        issued_body: Final = _response_object(issued)
+        assert (issued_body["user_id"], issued_body["team_id"]) == (member, team_id), issued.text
+        r1: Final = str(issued_body["refresh_token"])
+        assert r1.startswith("llm_srefresh_"), issued.text
+
+        def refresh(refresh_token: str) -> httpx.Response:
+            return gateway.client.post(
+                str(contract["token_endpoint"]),
+                data={
+                    "grant_type": "refresh_token",
+                    "refresh_token": refresh_token,
+                    "client_id": client_id,
+                    "resource": resource,
+                },
+            )
+
+        def revoke(refresh_token: str) -> httpx.Response:
+            return gateway.client.post(
+                str(contract["revocation_endpoint"]),
+                data={"token": refresh_token, "token_type_hint": "refresh_token", "client_id": client_id},
+            )
+
+        rotated: Final = refresh(r1)
+        assert rotated.status_code == 200, rotated.text
+        rotated_body: Final = _response_object(rotated)
+        r2: Final = str(rotated_body["refresh_token"])
+        assert r2 != r1 and r2.startswith("llm_srefresh_"), rotated.text
+        assert (rotated_body["user_id"], rotated_body["team_id"]) == (member, team_id), rotated.text
+        replayed: Final = refresh(r1)
+        assert replayed.status_code == 400, replayed.text
+        assert _response_object(replayed) == {
+            "error": "invalid_grant",
+            "error_description": "the refresh token was already used",
+        }, replayed.text
+        logged_out: Final = revoke(r2)
+        assert logged_out.status_code == 200, logged_out.text
+        assert _response_object(logged_out) == {}, logged_out.text
+        logged_out_again: Final = revoke(r2)
+        assert logged_out_again.status_code == 200, logged_out_again.text
+        assert _response_object(logged_out_again) == {}, logged_out_again.text
+        after_logout: Final = refresh(r2)
+        assert after_logout.status_code == 400, after_logout.text
+        assert _response_object(after_logout) == {
+            "error": "invalid_grant",
+            "error_description": "the refresh token was revoked",
+        }, after_logout.text
+        completion: Final = gateway.request(
+            "POST",
+            "/v1/chat/completions",
+            {"model": model, "messages": [{"role": "user", "content": "lite logout access token"}]},
+            key=str(rotated_body["access_token"]),
+        )
+        assert completion.status_code == 200, completion.text
 
 
 def _toolset_rpc(gateway: Gateway, bearer: str, name: str, method: str, params: dict[str, object]) -> Outcome:
@@ -584,6 +1258,623 @@ def test_token_exchange_callers_with_different_subject_tokens_own_separate_listi
         assert tool_calls(peer.drain()) == (), "a blocked probe reached the peer"
 
 
+@dataclass(frozen=True, slots=True)
+class _IdpRig:
+    gateway: Gateway
+    private_key: RSAPrivateKey
+    other_private_key: RSAPrivateKey
+    kid: str
+    dead_jwks: Wire
+
+
+def _proxy_config_with_general_settings(directory: Path, name: str, overrides: Mapping[str, object]) -> Path:
+    raw_config: Final[object] = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+    config: Final = TypeAdapter(dict[str, object]).validate_python(raw_config)
+    general_settings: Final = TypeAdapter(dict[str, object]).validate_python(config["general_settings"])
+    merged: Final = {**config, "general_settings": {**general_settings, **overrides}}
+    config_path: Final = directory / name
+    config_path.write_text(yaml.safe_dump(merged))
+    return config_path
+
+
+def _idp_proxy_config(directory: Path, team_id_jwt_field: str | None = None) -> Path:
+    return _proxy_config_with_general_settings(
+        directory,
+        "mcp_idp.yaml",
+        {
+            "enable_jwt_auth": True,
+            "litellm_jwtauth": {
+                "user_id_jwt_field": "sub",
+                "user_id_upsert": True,
+                **({} if team_id_jwt_field is None else {"team_id_jwt_field": team_id_jwt_field}),
+            },
+            "use_x_forwarded_for": True,
+            "mcp_trusted_proxy_ranges": ["127.0.0.1/32"],
+        },
+    )
+
+
+def _signed_subject_token(private_key: RSAPrivateKey, claims: Mapping[str, str], kid: str) -> str:
+    now: Final = int(time.time())
+    return jwt.encode({**claims, "iat": now, "exp": now + 300}, private_key, algorithm="RS256", headers={"kid": kid})
+
+
+def _token_exchange_form(client_id: str, subject_token: str) -> dict[str, str]:
+    return {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "client_id": client_id,
+        "subject_token": subject_token,
+        "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
+        "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
+    }
+
+
+def _registered_dcr_client(gateway: Gateway) -> str:
+    registered: Final = gateway.client.post(
+        "/register", json={"redirect_uris": [CLIENT_REDIRECT], "client_name": "integration"}
+    )
+    assert registered.status_code in (200, 201), registered.text
+    return str(_response_object(registered)["client_id"])
+
+
+@contextmanager
+def _idp_gateway(directory: Path, team_id_jwt_field: str | None, workers: int) -> Iterator[_IdpRig]:
+    assert os.environ.get("LITELLM_LICENSE"), (
+        "enable_jwt_auth is enterprise-only: these tests need LITELLM_LICENSE (CI forwards it); "
+        "without it /token answers 400 'JWT auth is an enterprise only feature; no license is set' "
+        "and the failure reads as a regression instead of a missing license"
+    )
+    private_key: Final = generate_private_key(public_exponent=65537, key_size=2048)
+    other_private_key: Final = generate_private_key(public_exponent=65537, key_size=2048)
+    kid: Final = "kid-a"
+    public_jwk: Final = TypeAdapter(dict[str, object]).validate_python(
+        json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(private_key.public_key()))
+    )
+
+    def respond_good(request: Request) -> Reply:
+        assert request.method == "GET", request
+        return Reply(body=json.dumps({"keys": [{**public_jwk, "kid": kid}]}).encode())
+
+    def respond_dead(request: Request) -> Reply:
+        assert request.method == "GET", request
+        return Reply(drop_connection=True)
+
+    config_path: Final = _idp_proxy_config(directory, team_id_jwt_field)
+    with (
+        wire_server(respond_good) as good_jwks,
+        wire_server(respond_dead) as dead_jwks,
+        gateway_from_environment() as gateway,
+        owned_proxy(
+            gateway,
+            directory,
+            {"JWT_PUBLIC_KEY_URL": f"{good_jwks.url},{dead_jwks.url}"},
+            config=config_path,
+            remove_environment=("PROXY_BASE_URL",),
+            workers=workers,
+        ) as candidate,
+    ):
+        yield _IdpRig(candidate, private_key, other_private_key, kid, dead_jwks)
+
+
+@pytest.fixture(scope="module")
+def idp_rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_IdpRig]:
+    with _idp_gateway(tmp_path_factory.mktemp("mcp-idp"), None, workers=2) as rig:
+        yield rig
+
+
+@pytest.fixture(scope="module")
+def idp_team_rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_IdpRig]:
+    with _idp_gateway(tmp_path_factory.mktemp("mcp-idp-team"), "team_id", workers=1) as rig:
+        yield rig
+
+
+@pytest.fixture(scope="module")
+def forwarded_gateway(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Gateway]:
+    directory: Final = tmp_path_factory.mktemp("mcp-forwarded")
+    config: Final = _proxy_config_with_general_settings(
+        directory,
+        "mcp_forwarded.yaml",
+        {"use_x_forwarded_for": True, "mcp_trusted_proxy_ranges": ["127.0.0.1/32"]},
+    )
+    with (
+        gateway_from_environment() as gateway,
+        owned_proxy(
+            gateway,
+            directory,
+            {"FORWARDED_ALLOW_IPS": "192.0.2.1"},
+            config=config,
+            remove_environment=("PROXY_BASE_URL",),
+            workers=1,
+        ) as candidate,
+    ):
+        yield candidate
+
+
+@pytest.fixture(
+    scope="module",
+    params=(
+        {},
+        {"use_x_forwarded_for": True, "mcp_trusted_proxy_ranges": ["192.0.2.0/24"]},
+    ),
+    ids=("xff-off", "outside-trusted-range"),
+)
+def untrusted_gateway(request: pytest.FixtureRequest, tmp_path_factory: pytest.TempPathFactory) -> Iterator[Gateway]:
+    directory: Final = tmp_path_factory.mktemp("mcp-untrusted-forwarded")
+    config: Final = _proxy_config_with_general_settings(
+        directory,
+        "mcp_untrusted.yaml",
+        TypeAdapter(dict[str, object]).validate_python(request.param),
+    )
+    with (
+        gateway_from_environment() as gateway,
+        owned_proxy(
+            gateway,
+            directory,
+            {"FORWARDED_ALLOW_IPS": "192.0.2.1"},
+            config=config,
+            remove_environment=("PROXY_BASE_URL",),
+            workers=1,
+        ) as candidate,
+    ):
+        yield candidate
+
+
+def test_idp_subject_token_exchange_mints_a_credential_for_the_mapped_user_and_refuses_bad_subjects(
+    idp_rig: _IdpRig,
+) -> None:
+    registered: Final = idp_rig.gateway.client.post(
+        "/register", json={"redirect_uris": [CLIENT_REDIRECT], "client_name": "integration"}
+    )
+    assert registered.status_code in (200, 201), registered.text
+    client_id: Final = str(_response_object(registered)["client_id"])
+    subject: Final = "mcp-idp-user-" + uuid.uuid4().hex
+
+    def token(private_key: RSAPrivateKey, user_id: str, kid: str) -> str:
+        now: Final = int(time.time())
+        return jwt.encode(
+            {"sub": user_id, "iat": now, "exp": now + 300},
+            private_key,
+            algorithm="RS256",
+            headers={"kid": kid},
+        )
+
+    valid_subject: Final = token(idp_rig.private_key, subject, idp_rig.kid)
+    exchange_form: Final = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "client_id": client_id,
+        "subject_token": valid_subject,
+        "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
+        "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
+    }
+    id_token_exchange: Final = idp_rig.gateway.client.post(
+        "/token",
+        data={
+            **exchange_form,
+            "subject_token_type": "urn:ietf:params:oauth:token-type:id_token",
+        },
+    )
+    assert id_token_exchange.status_code == 200, id_token_exchange.text
+    id_token_exchange_body: Final = _response_object(id_token_exchange)
+    assert set(id_token_exchange_body) == {
+        "access_token",
+        "token_type",
+        "expires_in",
+        "refresh_token",
+        "user_id",
+        "team_id",
+        "issued_token_type",
+    }, id_token_exchange.text
+    assert id_token_exchange_body["token_type"] == "Bearer", id_token_exchange.text
+    assert id_token_exchange_body["issued_token_type"] == "urn:ietf:params:oauth:token-type:access_token", (
+        id_token_exchange.text
+    )
+    assert id_token_exchange_body["user_id"] == subject, id_token_exchange.text
+    assert id_token_exchange_body["team_id"] is None, id_token_exchange.text
+    assert isinstance(id_token_exchange_body["expires_in"], int) and id_token_exchange_body["expires_in"] > 0, (
+        id_token_exchange.text
+    )
+    assert str(id_token_exchange_body["refresh_token"]).startswith("llm_srefresh_"), id_token_exchange.text
+    exchanged: Final = idp_rig.gateway.client.post("/token", data=exchange_form)
+    assert exchanged.status_code == 200, exchanged.text
+    outage_form: Final = {
+        **exchange_form,
+        "subject_token": token(idp_rig.other_private_key, subject, "kid-b"),
+    }
+    unavailable: Final = idp_rig.gateway.client.post("/token", data=outage_form)
+    dead_requests: Final = idp_rig.dead_jwks.drain()
+    assert unavailable.status_code == 503, unavailable.text
+    assert _response_object(unavailable) == {
+        "error": "temporarily_unavailable",
+        "error_description": (
+            "the gateway could not verify subject_token because its identity provider or database is unavailable; retry"
+        ),
+    }, unavailable.text
+    assert tuple(request.method for request in dead_requests) == ("GET", "GET", "GET"), dead_requests
+    exchanged_body: Final = _response_object(exchanged)
+    assert set(exchanged_body) == {
+        "access_token",
+        "token_type",
+        "expires_in",
+        "refresh_token",
+        "user_id",
+        "team_id",
+        "issued_token_type",
+    }, exchanged.text
+    assert exchanged_body["token_type"] == "Bearer", exchanged.text
+    assert exchanged_body["issued_token_type"] == "urn:ietf:params:oauth:token-type:access_token", exchanged.text
+    assert exchanged_body["user_id"] == subject, exchanged.text
+    assert exchanged_body["team_id"] is None, exchanged.text
+    assert isinstance(exchanged_body["expires_in"], int) and exchanged_body["expires_in"] > 0, exchanged.text
+    assert str(exchanged_body["refresh_token"]).startswith("llm_srefresh_"), exchanged.text
+    access_token: Final = str(exchanged_body["access_token"])
+    id_token_access_token: Final = str(id_token_exchange_body["access_token"])
+    with idp_rig.gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        completion: Final = idp_rig.gateway.request(
+            "POST",
+            "/v1/chat/completions",
+            {"model": model, "messages": [{"role": "user", "content": "mcp idp spend"}]},
+            key=access_token,
+        )
+        assert completion.status_code == 200, completion.text
+        request_id: Final = str(_response_object(completion)["id"])
+        id_token_completion: Final = idp_rig.gateway.request(
+            "POST",
+            "/v1/chat/completions",
+            {"model": model, "messages": [{"role": "user", "content": "mcp idp id token spend"}]},
+            key=id_token_access_token,
+        )
+        assert id_token_completion.status_code == 200, id_token_completion.text
+        id_token_request_id: Final = str(_response_object(id_token_completion)["id"])
+        request_ids: Final = tuple(sorted((request_id, id_token_request_id)))
+        spend: Final = eventually(
+            lambda: read_rows(
+                "SELECT request_id, metadata->>'user_api_key_user_id' AS user_api_key_user_id "
+                'FROM "LiteLLM_SpendLogs" WHERE request_id IN (%s, %s) ORDER BY request_id',
+                request_ids,
+            ),
+            lambda rows: len(rows) == 2,
+            seconds=70,
+        )
+        assert spend == [
+            {"request_id": request_ids[0], "user_api_key_user_id": subject},
+            {"request_id": request_ids[1], "user_api_key_user_id": subject},
+        ], spend
+    user: Final = eventually(
+        lambda: read_rows('SELECT user_id, user_role, teams FROM "LiteLLM_UserTable" WHERE user_id = %s', (subject,)),
+        lambda rows: len(rows) == 1,
+        seconds=70,
+    )
+    assert user == [{"user_id": subject, "user_role": None, "teams": []}], user
+    access_token_hash: Final = hashlib.sha256(access_token.encode()).hexdigest()
+    verification_tokens: Final = read_rows(
+        'SELECT token, user_id FROM "LiteLLM_VerificationToken" WHERE token = %s',
+        (access_token_hash,),
+    )
+    assert verification_tokens == [], verification_tokens
+    missing_subject: Final = idp_rig.gateway.client.post(
+        "/token",
+        data={key: value for key, value in exchange_form.items() if key != "subject_token"},
+    )
+    assert missing_subject.status_code == 400, missing_subject.text
+    assert _response_object(missing_subject) == {
+        "error": "invalid_request",
+        "error_description": "subject_token and subject_token_type are required",
+    }, missing_subject.text
+    rejected_subject: Final = idp_rig.gateway.client.post(
+        "/token", data={**exchange_form, "subject_token": token(idp_rig.other_private_key, subject, idp_rig.kid)}
+    )
+    assert rejected_subject.status_code == 400, rejected_subject.text
+    assert _response_object(rejected_subject) == {
+        "error": "invalid_request",
+        "error_description": "subject_token was rejected by the gateway's JWT auth",
+    }, rejected_subject.text
+    non_jwt: Final = idp_rig.gateway.client.post("/token", data={**exchange_form, "subject_token": "not-a-jwt"})
+    assert non_jwt.status_code == 400, non_jwt.text
+    assert _response_object(non_jwt) == {
+        "error": "invalid_request",
+        "error_description": "subject_token is not a JWT",
+    }, non_jwt.text
+
+
+def test_idp_subject_token_exchange_binds_the_team_claim_and_refuses_a_signed_subject_with_no_user(
+    idp_team_rig: _IdpRig,
+) -> None:
+    client_id: Final = _registered_dcr_client(idp_team_rig.gateway)
+    subject: Final = "mcp-idp-team-user-" + uuid.uuid4().hex
+    with idp_team_rig.gateway.scenario() as scenario:
+        model: Final = scenario.model()
+        team_id: Final = scenario.team()
+        exchanged: Final = idp_team_rig.gateway.client.post(
+            "/token",
+            data=_token_exchange_form(
+                client_id,
+                _signed_subject_token(idp_team_rig.private_key, {"sub": subject, "team_id": team_id}, idp_team_rig.kid),
+            ),
+        )
+        assert exchanged.status_code == 200, exchanged.text
+        body: Final = _response_object(exchanged)
+        assert (body["user_id"], body["team_id"], body["issued_token_type"]) == (
+            subject,
+            team_id,
+            "urn:ietf:params:oauth:token-type:access_token",
+        ), exchanged.text
+        completion: Final = idp_team_rig.gateway.request(
+            "POST",
+            "/v1/chat/completions",
+            {"model": model, "messages": [{"role": "user", "content": "mcp idp team spend"}]},
+            key=str(body["access_token"]),
+        )
+        assert completion.status_code == 200, completion.text
+        request_id: Final = str(_response_object(completion)["id"])
+        spend: Final = eventually(
+            lambda: read_rows(
+                "SELECT team_id, metadata->>'user_api_key_user_id' AS user_api_key_user_id, "
+                "metadata->>'user_api_key_team_id' AS user_api_key_team_id "
+                'FROM "LiteLLM_SpendLogs" WHERE request_id = %s',
+                (request_id,),
+            ),
+            lambda rows: len(rows) == 1,
+            seconds=70,
+        )
+        assert spend == [{"team_id": team_id, "user_api_key_user_id": subject, "user_api_key_team_id": team_id}], spend
+        no_user: Final = idp_team_rig.gateway.client.post(
+            "/token",
+            data=_token_exchange_form(
+                client_id, _signed_subject_token(idp_team_rig.private_key, {"team_id": team_id}, idp_team_rig.kid)
+            ),
+        )
+        assert no_user.status_code == 400, no_user.text
+        assert _response_object(no_user) == {
+            "error": "invalid_request",
+            "error_description": "subject_token names no user the gateway knows",
+        }, no_user.text
+
+
+def test_idp_jwks_server_error_answers_temporarily_unavailable(tmp_path: Path) -> None:
+    pytest.skip(
+        "BUG: an IdP JWKS answering HTTP 503 makes the gateway /token token-exchange answer 400 invalid_request "
+        "(subject_token rejected) instead of 503 temporarily_unavailable"
+    )
+    assert os.environ.get("LITELLM_LICENSE"), (
+        "enable_jwt_auth is enterprise-only: these tests need LITELLM_LICENSE (CI forwards it); "
+        "without it /token answers 400 'JWT auth is an enterprise only feature; no license is set' "
+        "and the failure reads as a regression instead of a missing license"
+    )
+    private_key: Final = generate_private_key(public_exponent=65537, key_size=2048)
+    kid: Final = "mcp-runtime-503"
+
+    def respond(request: Request) -> Reply:
+        assert request.method == "GET", request
+        return Reply(status=503, body=b'{"error":"jwks_down"}')
+
+    config: Final = _idp_proxy_config(tmp_path)
+    with (
+        wire_server(respond) as jwks,
+        gateway_from_environment() as gateway,
+        owned_proxy(
+            gateway,
+            tmp_path,
+            {"JWT_PUBLIC_KEY_URL": jwks.url},
+            config=config,
+            remove_environment=("PROXY_BASE_URL",),
+            workers=2,
+        ) as idp_gateway,
+    ):
+        registered: Final = idp_gateway.client.post(
+            "/register", json={"redirect_uris": [CLIENT_REDIRECT], "client_name": "integration"}
+        )
+        assert registered.status_code in (200, 201), registered.text
+        client_id: Final = str(_response_object(registered)["client_id"])
+        now: Final = int(time.time())
+        subject_token: Final = jwt.encode(
+            {"sub": "mcp-idp-http-503-" + uuid.uuid4().hex, "iat": now, "exp": now + 300},
+            private_key,
+            algorithm="RS256",
+            headers={"kid": kid},
+        )
+        exchange_form: Final = {
+            "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+            "client_id": client_id,
+            "subject_token": subject_token,
+            "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
+            "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        }
+        response: Final = idp_gateway.client.post("/token", data=exchange_form)
+        jwks_requests: Final = jwks.drain()
+        assert response.status_code == 503, (
+            f"request={exchange_form!r}, jwks_requests={jwks_requests!r}, "
+            f"response_status={response.status_code}, response={response.text}"
+        )
+        assert _response_object(response) == {
+            "error": "temporarily_unavailable",
+            "error_description": (
+                "the gateway could not verify subject_token because its identity provider or database is unavailable; "
+                "retry"
+            ),
+        }, response.text
+
+
+def test_standard_pattern_discovery_names_the_per_server_issuer_and_follows_forwarded_headers(
+    gateway: Gateway, forwarded_gateway: Gateway
+) -> None:
+    with mcp_peer() as peer, gateway.scenario() as scenario:
+        alias: Final = "rt8" + uuid.uuid4().hex[:10]
+        register_mcp(
+            scenario,
+            peer,
+            alias,
+            auth_type="oauth2",
+            oauth2_flow="authorization_code",
+            credentials={
+                "client_id": "rt8-client",
+                "client_secret": "rt8-secret",
+                "scopes": ["tools.call"],
+            },
+        )
+        base: Final = _base(gateway)
+        protected: Final = gateway.client.get(f"/.well-known/oauth-protected-resource/mcp/{alias}")
+        assert protected.status_code == 200, protected.text
+        aggregate_issuer: Final = f"{base}/mcp"
+        expected_protected: Final = {
+            "authorization_servers": [aggregate_issuer],
+            "resource": f"{base}/mcp/{alias}",
+            "scopes_supported": ["tools.call"],
+        }
+        assert _response_object(protected) == expected_protected, protected.text
+        aggregate_authorization: Final = gateway.client.get("/.well-known/oauth-authorization-server/mcp")
+        assert aggregate_authorization.status_code == 200, aggregate_authorization.text
+        assert _response_object(aggregate_authorization)["issuer"] == aggregate_issuer, aggregate_authorization.text
+        authorization: Final = gateway.client.get(f"/.well-known/oauth-authorization-server/mcp/{alias}")
+        assert authorization.status_code == 200, authorization.text
+        expected_authorization: Final = {
+            "issuer": f"{base}/mcp/{alias}",
+            "authorization_endpoint": f"{base}/{alias}/authorize",
+            "token_endpoint": f"{base}/{alias}/token",
+            "response_types_supported": ["code"],
+            "scopes_supported": ["tools.call"],
+            "grant_types_supported": ["authorization_code", "refresh_token"],
+            "code_challenge_methods_supported": ["S256"],
+            "token_endpoint_auth_methods_supported": ["client_secret_post"],
+            "registration_endpoint": f"{base}/{alias}/register",
+        }
+        assert _response_object(authorization) == expected_authorization, authorization.text
+        legacy_authorization: Final = gateway.client.get(f"/.well-known/oauth-authorization-server/{alias}")
+        assert legacy_authorization.status_code == 200, legacy_authorization.text
+        assert _response_object(legacy_authorization) == {**expected_authorization, "issuer": f"{base}/{alias}"}, (
+            legacy_authorization.text
+        )
+        legacy_protected: Final = gateway.client.get(f"/.well-known/oauth-protected-resource/{alias}/mcp")
+        assert legacy_protected.status_code == 200, legacy_protected.text
+        assert _response_object(legacy_protected) == {
+            "authorization_servers": [f"{base}/mcp"],
+            "resource": f"{base}/{alias}/mcp",
+            "scopes_supported": ["tools.call"],
+        }, legacy_protected.text
+        root_protected: Final = gateway.client.get("/.well-known/oauth-protected-resource")
+        assert root_protected.status_code == 200, root_protected.text
+        assert _response_object(root_protected) == {
+            "authorization_servers": [f"{base}/mcp"],
+            "resource": base,
+            "scopes_supported": [],
+        }, root_protected.text
+        anonymous: Final = gateway.client.post(
+            f"/mcp/{alias}",
+            headers=ACCEPT,
+            json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": dict(INITIALIZE)},
+        )
+        assert anonymous.status_code == 401, anonymous.text
+        challenge: Final = anonymous.headers["www-authenticate"]
+        metadata_ref: Final = re.search(r'resource_metadata="([^"]+)"', challenge)
+        assert metadata_ref is not None, challenge
+        metadata: Final = gateway.client.get(urljoin(base + "/", metadata_ref.group(1)))
+        assert metadata.status_code == 200, metadata.text
+        assert _response_object(metadata) == expected_protected, metadata.text
+    with mcp_peer() as peer, forwarded_gateway.scenario() as scenario:
+        alias: Final = "rt8f" + uuid.uuid4().hex[:9]
+        register_mcp(
+            scenario,
+            peer,
+            alias,
+            auth_type="oauth2",
+            oauth2_flow="authorization_code",
+            credentials={"client_id": "rt8f-client", "client_secret": "rt8f-secret"},
+        )
+        forwarded_base: Final = "https://gw.example.test"
+        forwarded_headers: Final = {
+            "X-Forwarded-Proto": "https",
+            "X-Forwarded-Host": "gw.example.test",
+        }
+        forwarded_prm: Final = forwarded_gateway.client.get(
+            f"/.well-known/oauth-protected-resource/mcp/{alias}", headers=forwarded_headers
+        )
+        assert forwarded_prm.status_code == 200, forwarded_prm.text
+        assert _response_object(forwarded_prm) == {
+            "authorization_servers": [f"{forwarded_base}/mcp"],
+            "resource": f"{forwarded_base}/mcp/{alias}",
+            "scopes_supported": [],
+        }, forwarded_prm.text
+        forwarded_as: Final = forwarded_gateway.client.get(
+            f"/.well-known/oauth-authorization-server/mcp/{alias}", headers=forwarded_headers
+        )
+        assert forwarded_as.status_code == 200, forwarded_as.text
+        assert _response_object(forwarded_as) == {
+            "issuer": f"{forwarded_base}/mcp/{alias}",
+            "authorization_endpoint": f"{forwarded_base}/{alias}/authorize",
+            "token_endpoint": f"{forwarded_base}/{alias}/token",
+            "response_types_supported": ["code"],
+            "scopes_supported": [],
+            "grant_types_supported": ["authorization_code", "refresh_token"],
+            "code_challenge_methods_supported": ["S256"],
+            "token_endpoint_auth_methods_supported": ["client_secret_post"],
+            "registration_endpoint": f"{forwarded_base}/{alias}/register",
+        }, forwarded_as.text
+        local_base: Final = _base(forwarded_gateway)
+        local_as: Final = forwarded_gateway.client.get(f"/.well-known/oauth-authorization-server/mcp/{alias}")
+        assert local_as.status_code == 200, local_as.text
+        assert _response_object(local_as) == {
+            "issuer": f"{local_base}/mcp/{alias}",
+            "authorization_endpoint": f"{local_base}/{alias}/authorize",
+            "token_endpoint": f"{local_base}/{alias}/token",
+            "response_types_supported": ["code"],
+            "scopes_supported": [],
+            "grant_types_supported": ["authorization_code", "refresh_token"],
+            "code_challenge_methods_supported": ["S256"],
+            "token_endpoint_auth_methods_supported": ["client_secret_post"],
+            "registration_endpoint": f"{local_base}/{alias}/register",
+        }, local_as.text
+
+
+def test_standard_pattern_discovery_ignores_forwarded_headers_from_an_untrusted_peer(
+    untrusted_gateway: Gateway,
+) -> None:
+    with mcp_peer() as peer, untrusted_gateway.scenario() as scenario:
+        alias: Final = "rt8u" + uuid.uuid4().hex[:9]
+        register_mcp(
+            scenario,
+            peer,
+            alias,
+            auth_type="oauth2",
+            oauth2_flow="authorization_code",
+            credentials={
+                "client_id": "rt8u-client",
+                "client_secret": "rt8u-secret",
+                "scopes": ["tools.call"],
+            },
+        )
+        base: Final = _base(untrusted_gateway)
+        untrusted_headers: Final = {
+            "X-Forwarded-Proto": "https",
+            "X-Forwarded-Host": "evil.example",
+        }
+        authorization: Final = untrusted_gateway.client.get(
+            f"/.well-known/oauth-authorization-server/mcp/{alias}", headers=untrusted_headers
+        )
+        assert authorization.status_code == 200, authorization.text
+        assert _response_object(authorization) == {
+            "issuer": f"{base}/mcp/{alias}",
+            "authorization_endpoint": f"{base}/{alias}/authorize",
+            "token_endpoint": f"{base}/{alias}/token",
+            "response_types_supported": ["code"],
+            "scopes_supported": ["tools.call"],
+            "grant_types_supported": ["authorization_code", "refresh_token"],
+            "code_challenge_methods_supported": ["S256"],
+            "token_endpoint_auth_methods_supported": ["client_secret_post"],
+            "registration_endpoint": f"{base}/{alias}/register",
+        }, authorization.text
+        assert "evil.example" not in authorization.text, authorization.text
+        protected: Final = untrusted_gateway.client.get(
+            f"/.well-known/oauth-protected-resource/mcp/{alias}", headers=untrusted_headers
+        )
+        assert protected.status_code == 200, protected.text
+        assert _response_object(protected) == {
+            "authorization_servers": [f"{base}/mcp"],
+            "resource": f"{base}/mcp/{alias}",
+            "scopes_supported": ["tools.call"],
+        }, protected.text
+        assert "evil.example" not in protected.text, protected.text
+
+
 @contextmanager
 def _scope_enforcing_peer(auth: AuthorizationServer) -> Iterator[tuple[McpPeer, Wire]]:
     with scripted_peer(
@@ -591,6 +1882,7 @@ def _scope_enforcing_peer(auth: AuthorizationServer) -> Iterator[tuple[McpPeer,
         ScriptedTool("write_op", lambda _: text_result("write completed")),
         ScriptedTool("forbid_op", lambda _: text_result("must not execute")),
     ) as inner:
+
         def enforce(request: Request) -> Reply:
             body: Final = json.loads(request.body) if request.body else {}
             if body.get("method") == "tools/call":
@@ -602,9 +1894,18 @@ def _scope_enforcing_peer(auth: AuthorizationServer) -> Iterator[tuple[McpPeer,
                 with auth.lock:
                     granted: Final = auth.access_tokens.get(token, {}).get("scope", "")
                 if required not in granted.split():
-                    return Reply(status=403, headers={"WWW-Authenticate": f'Bearer error="insufficient_scope", scope="{required}"'})
-            forwarded: Final = httpx.request(request.method, inner.url, content=request.body, headers=dict(request.headers))
-            return Reply(status=forwarded.status_code, body=forwarded.content, content_type=forwarded.headers.get("content-type", "application/json"))
+                    return Reply(
+                        status=403,
+                        headers={"WWW-Authenticate": f'Bearer error="insufficient_scope", scope="{required}"'},
+                    )
+            forwarded: Final = httpx.request(
+                request.method, inner.url, content=request.body, headers=dict(request.headers)
+            )
+            return Reply(
+                status=forwarded.status_code,
+                body=forwarded.content,
+                content_type=forwarded.headers.get("content-type", "application/json"),
+            )
 
         with wire_server(enforce) as wire:
             yield McpPeer(wire.url + "/mcp", inner.calls), wire
@@ -612,16 +1913,25 @@ def _scope_enforcing_peer(auth: AuthorizationServer) -> Iterator[tuple[McpPeer,
 
 @pytest.mark.parametrize("entry", ["rest", "server_mcp"])
 def test_managed_scope_step_up_preserves_read_and_write_without_retry(gateway: Gateway, entry: EntryPoint) -> None:
-    with oauth_server(scopes=("tools.read", "tools.write")) as auth, _scope_enforcing_peer(auth) as (peer, wire), gateway.scenario() as scenario:
+    with (
+        oauth_server(scopes=("tools.read", "tools.write")) as auth,
+        _scope_enforcing_peer(auth) as (peer, wire),
+        gateway.scenario() as scenario,
+    ):
         alias: Final = "scope" + uuid.uuid4().hex[:8]
         identity: Final = _register_oauth(
-            scenario, peer, auth, alias, auth_type="oauth2", oauth2_flow="authorization_code",
+            scenario,
+            peer,
+            auth,
+            alias,
+            auth_type="oauth2",
+            oauth2_flow="authorization_code",
             credentials={"client_id": "scope-client", "client_secret": "scope-secret", "scopes": ["tools.read"]},
         )
         owner: Final = scenario.key(user_id=scenario.user(), object_permission={"mcp_servers": [identity]})
         stranger: Final = scenario.key(user_id=scenario.user(), object_permission={"mcp_servers": [identity]})
         pkce: Final = _Pkce(secrets.token_urlsafe(32))
-        code: Final = _authorize_through_gateway(gateway, auth, alias, owner, "scope-client", pkce, "tools.read")
+        code, _ = _authorize_through_gateway(gateway, auth, alias, owner, "scope-client", pkce, "tools.read")
         first: Final = _redeem(gateway, alias, owner, "scope-client", code, pkce)
         assert first.status_code == 200, first.text
         caller: Final = McpCaller(gateway, owner, entry, alias)
@@ -630,16 +1940,21 @@ def test_managed_scope_step_up_preserves_read_and_write_without_retry(gateway: G
         peer.drain()
         wire.drain()
         denied: Final = (
-            call_tool(gateway, owner, identity, f"{alias}-write_op", {}) if entry == "rest"
+            call_tool(gateway, owner, identity, f"{alias}-write_op", {})
+            if entry == "rest"
             else caller.rpc("tools/call", {"name": f"{alias}-write_op", "arguments": {}})
         )
         assert denied.status_code == 403, denied.text
         challenge: Final = denied.headers["www-authenticate"]
         assert 'error="insufficient_scope"' in challenge and 'scope="tools.write"' in challenge
         assert f'resource_metadata="{_base(gateway)}/.well-known/oauth-protected-resource/' in challenge
-        attempts: Final = tuple(request for request in wire.drain() if request.body and json.loads(request.body).get("method") == "tools/call")
+        attempts: Final = tuple(
+            request
+            for request in wire.drain()
+            if request.body and json.loads(request.body).get("method") == "tools/call"
+        )
         assert len(attempts) == 1 and tool_calls(peer.drain()) == (), "denied operation was retried or executed"
-        second_code: Final = _authorize_through_gateway(gateway, auth, alias, owner, "scope-client", pkce, "tools.write")
+        second_code, _ = _authorize_through_gateway(gateway, auth, alias, owner, "scope-client", pkce, "tools.write")
         second: Final = _redeem(gateway, alias, owner, "scope-client", second_code, pkce)
         assert second.status_code == 200, second.text
         with auth.lock:
@@ -681,7 +1996,7 @@ def test_managed_step_up_advertises_sdk_compatible_metadata_and_bound_permission
             user_id=scenario.user(), object_permission={"mcp_servers": [identity], "mcp_toolsets": [toolset_id]}
         )
         pkce: Final = _Pkce(secrets.token_urlsafe(32))
-        code: Final = _authorize_through_gateway(gateway, auth, alias, owner, "scope-client", pkce, "tools.read")
+        code, _ = _authorize_through_gateway(gateway, auth, alias, owner, "scope-client", pkce, "tools.read")
         first: Final = _redeem(gateway, alias, owner, "scope-client", code, pkce)
         assert first.status_code == 200, first.text
         path: Final = route.format(toolset=toolset_name)
diff --git a/tests/integration/mcp/test_mcp_transports.py b/tests/integration/mcp/test_mcp_transports.py
index b448b8b4a64..2b6494914c8 100644
--- a/tests/integration/mcp/test_mcp_transports.py
+++ b/tests/integration/mcp/test_mcp_transports.py
@@ -1,27 +1,122 @@
+import builtins
 import json
 import uuid
+from collections.abc import Mapping
 from typing import Final
 
+import httpx
 import pytest
+from mcp.shared.exceptions import MCPError
 from integration._support.client import Gateway
+from integration._support.database import read_rows
 from integration._support.mcp import (
     ENTRY_POINTS,
+    INITIALIZE,
     PEER_KINDS,
     EntryPoint,
     McpCaller,
+    Outcome,
     PeerKind,
+    _outcome_from_rpc,
+    _parse_rpc_body,
     mcp_peer,
     official_client_outcomes,
     peer_of,
     register_mcp,
     tool_calls,
 )
+from integration._support.mcp_grants import create_toolset
 
 ADD: Final = {"http": "add", "sse": "add", "stdio": "add", "openapi": "getpet"}
 ARGUMENTS: Final = {"add": {"a": 3, "b": 4}, "getpet": {"petId": "7"}}
 EXPECTED: Final = {"add": "7", "getpet": json.dumps({"id": "7", "name": "integration-pet"})}
 
 
+def _streamable_rpc(gateway: Gateway, path: str, key: str, method: str, params: Mapping[str, object]) -> httpx.Response:
+    return gateway.client.post(
+        path,
+        json={"jsonrpc": "2.0", "id": 1, "method": method, "params": params},
+        headers={"Authorization": f"Bearer {key}", "Accept": "application/json, text/event-stream"},
+    )
+
+
+def _streamable_tools(gateway: Gateway, path: str, key: str) -> tuple[httpx.Response, Outcome]:
+    response: Final = _streamable_rpc(gateway, path, key, "tools/list", {})
+    return response, _outcome_from_rpc(response)
+
+
+def _list_and_call(
+    gateway: Gateway, client: str, path: str, key: str, name: str, arguments: Mapping[str, object]
+) -> tuple[Outcome, Outcome]:
+    if client == "official-sdk":
+        return official_client_outcomes(gateway, key, path, name, dict(arguments))
+    initialized: Final = _streamable_rpc(gateway, path, key, "initialize", dict(INITIALIZE))
+    assert initialized.status_code == 200, initialized.text
+    listed_response, listed = _streamable_tools(gateway, path, key)
+    assert listed_response.status_code == 200, listed_response.text
+    called_response: Final = _streamable_rpc(
+        gateway, path, key, "tools/call", {"name": name, "arguments": dict(arguments)}
+    )
+    assert called_response.status_code == 200, called_response.text
+    return listed, _outcome_from_rpc(called_response)
+
+
+def _refused_call(
+    gateway: Gateway,
+    client: str,
+    path: str,
+    key: str,
+    name: str,
+    arguments: Mapping[str, object],
+    text: str = "Error: User not allowed to call this tool.",
+) -> Outcome:
+    expected: Final = text
+    if client == "official-sdk":
+        _, refused = official_client_outcomes(gateway, key, path, name, dict(arguments))
+        assert (refused.status, refused.error, refused.text) == (200, expected, expected), refused
+        return refused
+    response: Final = _streamable_rpc(gateway, path, key, "tools/call", {"name": name, "arguments": dict(arguments)})
+    assert response.status_code == 200, response.text
+    assert _parse_rpc_body(response) == {
+        "jsonrpc": "2.0",
+        "id": 1,
+        "result": {
+            "content": [{"text": expected, "type": "text"}],
+            "isError": True,
+        },
+    }, response.text
+    return _outcome_from_rpc(response)
+
+
+def _leaf_exceptions(error: BaseException) -> tuple[BaseException, ...]:
+    if isinstance(error, builtins.BaseExceptionGroup):
+        return tuple(leaf for child in error.exceptions for leaf in _leaf_exceptions(child))
+    return (error,)
+
+
+def _assert_upstream_call(call: Mapping[str, object], name: str, arguments: Mapping[str, object]) -> None:
+    body: Final = call["body"]
+    assert isinstance(body, Mapping), call
+    params: Final = body["params"]
+    assert isinstance(params, Mapping), call
+    assert set(params) == {"name", "arguments", "_meta"}, call
+    assert params["name"] == name, call
+    assert params["arguments"] == arguments, call
+    metadata: Final = params["_meta"]
+    assert isinstance(metadata, Mapping), call
+    assert set(metadata) == {"progressToken"}, call
+
+
+def _open_aliases() -> frozenset[str]:
+    rows: Final = read_rows('SELECT alias FROM "LiteLLM_MCPServerTable" WHERE allow_all_keys', ())
+    return frozenset(str(row["alias"]) for row in rows)
+
+
+def _without_foreign_open_servers(tools: tuple[str, ...], open_aliases: frozenset[str]) -> set[str]:
+    prefixes: Final = tuple(f"{alias}-" for alias in open_aliases)
+    return {tool for tool in tools if not tool.startswith(prefixes)}
+
+
 def _peer_saw_call(peer_kind: PeerKind, observed: tuple[dict[str, object], ...], tool: str) -> bool:
     if peer_kind == "openapi":
         return any(item.get("path") == "/pets/7" and item.get("method") == "GET" for item in observed)
@@ -95,6 +190,306 @@ def test_official_client_session_lists_and_calls_through_gateway(
         assert _peer_saw_call(peer_kind, peer.drain(), "add")
 
 
+@pytest.mark.parametrize("client", ("raw-jsonrpc", "official-sdk"))
+def test_standard_per_server_path_scopes_the_session_to_the_named_servers(gateway: Gateway, client: str) -> None:
+    open_before: Final = _open_aliases()
+    with mcp_peer() as peer_a, mcp_peer() as peer_b, mcp_peer() as peer_c, gateway.scenario() as scenario:
+        alias_a: Final = "rt1a" + uuid.uuid4().hex[:10]
+        alias_b: Final = "rt1b" + uuid.uuid4().hex[:10]
+        alias_c: Final = "rt1c" + uuid.uuid4().hex[:10]
+        server_a: Final = register_mcp(scenario, peer_a, alias_a)
+        server_b: Final = register_mcp(scenario, peer_b, alias_b)
+        server_c: Final = register_mcp(scenario, peer_c, alias_c)
+        key: Final = scenario.key(object_permission={"mcp_servers": [server_a, server_b, server_c]})
+        open_aliases: Final = open_before | _open_aliases()
+
+        first_path: Final = f"/mcp/{alias_a}"
+        listed, called = _list_and_call(gateway, client, first_path, key, f"{alias_a}-add", {"a": 20, "b": 22})
+        assert listed.ok, listed.raw
+        assert _without_foreign_open_servers(listed.tools, open_aliases) == {
+            f"{alias_a}-add",
+            f"{alias_a}-multiply",
+            f"{alias_a}-fail",
+        }, listed.tools
+        assert called.ok and called.text == "42", called.raw
+        _refused_call(
+            gateway,
+            client,
+            first_path,
+            key,
+            f"{alias_b}-add",
+            {"a": 20, "b": 22},
+            f"Error: Tool {alias_b}-add not found",
+        )
+        calls_a: Final = tool_calls(peer_a.drain())
+        assert len(calls_a) == 1, calls_a
+        _assert_upstream_call(calls_a[0], "add", {"a": 20, "b": 22})
+        assert tool_calls(peer_b.drain()) == ()
+        assert tool_calls(peer_c.drain()) == ()
+
+        combined_path: Final = f"/mcp/{alias_a},{alias_b}"
+        combined_listed, combined_called = _list_and_call(
+            gateway, client, combined_path, key, f"{alias_b}-multiply", {"a": 6, "b": 7}
+        )
+        assert combined_listed.ok, combined_listed.raw
+        assert _without_foreign_open_servers(combined_listed.tools, open_aliases) == {
+            f"{alias_a}-add",
+            f"{alias_a}-multiply",
+            f"{alias_a}-fail",
+            f"{alias_b}-add",
+            f"{alias_b}-multiply",
+            f"{alias_b}-fail",
+        }, combined_listed.tools
+        assert combined_called.ok and combined_called.text == "42", combined_called.raw
+        _refused_call(gateway, client, combined_path, key, f"{alias_c}-add", {"a": 20, "b": 22})
+        assert tool_calls(peer_a.drain()) == ()
+        calls_b: Final = tool_calls(peer_b.drain())
+        assert len(calls_b) == 1, calls_b
+        _assert_upstream_call(calls_b[0], "multiply", {"a": 6, "b": 7})
+        assert tool_calls(peer_c.drain()) == ()
+
+
+@pytest.mark.parametrize("client", ("raw-jsonrpc", "official-sdk"))
+def test_standard_comma_path_with_a_trailing_segment_scopes_a_proxy_admin_session(
+    gateway: Gateway, client: str
+) -> None:
+    open_before: Final = _open_aliases()
+    with mcp_peer() as peer_a, mcp_peer() as peer_b, mcp_peer() as peer_c, gateway.scenario() as scenario:
+        alias_a: Final = "rt1pa" + uuid.uuid4().hex[:10]
+        alias_b: Final = "rt1pb" + uuid.uuid4().hex[:10]
+        alias_c: Final = "rt1pc" + uuid.uuid4().hex[:10]
+        register_mcp(scenario, peer_a, alias_a)
+        register_mcp(scenario, peer_b, alias_b)
+        register_mcp(scenario, peer_c, alias_c)
+        open_aliases: Final = open_before | _open_aliases()
+        path: Final = f"/mcp/{alias_a},{alias_c}/tools"
+
+        listed, called = _list_and_call(gateway, client, path, gateway.key, f"{alias_c}-add", {"a": 20, "b": 22})
+        assert listed.ok, listed.raw
+        assert _without_foreign_open_servers(listed.tools, open_aliases) == {
+            f"{alias_a}-add",
+            f"{alias_a}-multiply",
+            f"{alias_a}-fail",
+            f"{alias_c}-add",
+            f"{alias_c}-multiply",
+            f"{alias_c}-fail",
+        }, listed.tools
+        assert called.ok and called.text == "42", called.raw
+        _refused_call(gateway, client, path, gateway.key, f"{alias_b}-add", {"a": 20, "b": 22})
+        assert tool_calls(peer_a.drain()) == ()
+        assert tool_calls(peer_b.drain()) == ()
+        calls_c: Final = tool_calls(peer_c.drain())
+        assert len(calls_c) == 1, calls_c
+        _assert_upstream_call(calls_c[0], "add", {"a": 20, "b": 22})
+
+
+@pytest.mark.parametrize("client", ("raw-jsonrpc", "official-sdk"))
+def test_standard_comma_path_with_a_trailing_segment_admits_a_scoped_key(gateway: Gateway, client: str) -> None:
+    pytest.skip(
+        "BUG: a non-admin key gets 401 'Only proxy admin can be used...' on /mcp/{a},{b}/tools because "
+        "mcp_inference_routes only admits the single-segment /mcp/{subpath}, while "
+        "_get_mcp_servers_in_path parses the trailing path"
+    )
+    open_before: Final = _open_aliases()
+    with mcp_peer() as peer_a, mcp_peer() as peer_b, mcp_peer() as peer_c, gateway.scenario() as scenario:
+        alias_a: Final = "rt1sa" + uuid.uuid4().hex[:10]
+        alias_b: Final = "rt1sb" + uuid.uuid4().hex[:10]
+        alias_c: Final = "rt1sc" + uuid.uuid4().hex[:10]
+        server_a: Final = register_mcp(scenario, peer_a, alias_a)
+        server_b: Final = register_mcp(scenario, peer_b, alias_b)
+        server_c: Final = register_mcp(scenario, peer_c, alias_c)
+        key: Final = scenario.key(object_permission={"mcp_servers": [server_a, server_b, server_c]})
+        open_aliases: Final = open_before | _open_aliases()
+        path: Final = f"/mcp/{alias_a},{alias_c}/tools"
+
+        listed, called = _list_and_call(gateway, client, path, key, f"{alias_c}-add", {"a": 20, "b": 22})
+        assert listed.ok, listed.raw
+        assert _without_foreign_open_servers(listed.tools, open_aliases) == {
+            f"{alias_a}-add",
+            f"{alias_a}-multiply",
+            f"{alias_a}-fail",
+            f"{alias_c}-add",
+            f"{alias_c}-multiply",
+            f"{alias_c}-fail",
+        }, listed.tools
+        assert called.ok and called.text == "42", called.raw
+        _refused_call(gateway, client, path, key, f"{alias_b}-add", {"a": 20, "b": 22})
+        assert tool_calls(peer_a.drain()) == ()
+        assert tool_calls(peer_b.drain()) == ()
+        calls_c: Final = tool_calls(peer_c.drain())
+        assert len(calls_c) == 1, calls_c
+        _assert_upstream_call(calls_c[0], "add", {"a": 20, "b": 22})
+
+
+@pytest.mark.parametrize("client", ("raw-jsonrpc", "official-sdk"))
+def test_legacy_server_path_resolves_lists_groups_and_fails_closed_on_unknown_names(
+    gateway: Gateway, client: str
+) -> None:
+    open_before: Final = _open_aliases()
+    with mcp_peer() as peer_a, mcp_peer() as peer_b, mcp_peer() as peer_c, gateway.scenario() as scenario:
+        alias_a: Final = "rt2a" + uuid.uuid4().hex[:10]
+        alias_b: Final = "rt2b" + uuid.uuid4().hex[:10]
+        alias_c: Final = "rt2c" + uuid.uuid4().hex[:10]
+        group: Final = "rt2g" + uuid.uuid4().hex[:10]
+        server_a: Final = register_mcp(scenario, peer_a, alias_a, mcp_access_groups=[group])
+        server_b: Final = register_mcp(scenario, peer_b, alias_b, mcp_access_groups=[group])
+        server_c: Final = register_mcp(scenario, peer_c, alias_c)
+        key: Final = scenario.key(object_permission={"mcp_servers": [server_a, server_b, server_c]})
+        open_aliases: Final = open_before | _open_aliases()
+
+        direct_path: Final = f"/{alias_a},{alias_b}/mcp"
+        direct_listed, direct_called = _list_and_call(
+            gateway, client, direct_path, key, f"{alias_a}-add", {"a": 20, "b": 22}
+        )
+        assert direct_listed.ok, direct_listed.raw
+        assert _without_foreign_open_servers(direct_listed.tools, open_aliases) == {
+            f"{alias_a}-add",
+            f"{alias_a}-multiply",
+            f"{alias_a}-fail",
+            f"{alias_b}-add",
+            f"{alias_b}-multiply",
+            f"{alias_b}-fail",
+        }, direct_listed.tools
+        assert direct_called.ok and direct_called.text == "42", direct_called.raw
+        direct_calls_a: Final = tool_calls(peer_a.drain())
+        assert len(direct_calls_a) == 1, direct_calls_a
+        _assert_upstream_call(direct_calls_a[0], "add", {"a": 20, "b": 22})
+        assert tool_calls(peer_b.drain()) == ()
+        assert tool_calls(peer_c.drain()) == ()
+
+        mixed_path: Final = f"/{alias_a},%20{alias_a}%20,nope{uuid.uuid4().hex}/mcp"
+        mixed_listed, mixed_called = _list_and_call(
+            gateway, client, mixed_path, key, f"{alias_a}-multiply", {"a": 6, "b": 7}
+        )
+        assert mixed_listed.ok, mixed_listed.raw
+        assert _without_foreign_open_servers(mixed_listed.tools, open_aliases) == {
+            f"{alias_a}-add",
+            f"{alias_a}-multiply",
+            f"{alias_a}-fail",
+        }, mixed_listed.tools
+        assert mixed_called.ok and mixed_called.text == "42", mixed_called.raw
+        mixed_calls_a: Final = tool_calls(peer_a.drain())
+        assert len(mixed_calls_a) == 1, mixed_calls_a
+        _assert_upstream_call(mixed_calls_a[0], "multiply", {"a": 6, "b": 7})
+        assert tool_calls(peer_b.drain()) == ()
+        assert tool_calls(peer_c.drain()) == ()
+
+        unknown_path: Final = f"/nope1{uuid.uuid4().hex},nope2{uuid.uuid4().hex}/mcp"
+        assert peer_a.drain() == ()
+        assert peer_b.drain() == ()
+        assert peer_c.drain() == ()
+        if client == "official-sdk":
+            with pytest.raises(builtins.ExceptionGroup) as caught:
+                official_client_outcomes(gateway, key, unknown_path, f"{alias_a}-add", {"a": 20, "b": 22})
+            leaves: Final = _leaf_exceptions(caught.value)
+            assert len(leaves) == 1, repr(leaves)
+            leaf: Final = leaves[0]
+            assert isinstance(leaf, MCPError), repr(leaves)
+            assert leaf.error.code == -32601, repr(leaves)
+            assert leaf.error.message == "Not Found" and leaf.error.data is None, repr(leaves)
+        else:
+            unknown_initialized: Final = _streamable_rpc(gateway, unknown_path, key, "initialize", dict(INITIALIZE))
+            assert unknown_initialized.status_code == 404, unknown_initialized.text
+            unknown_listed: Final = _streamable_rpc(gateway, unknown_path, key, "tools/list", {})
+            assert unknown_listed.status_code == 404, unknown_listed.text
+        assert peer_a.drain() == ()
+        assert peer_b.drain() == ()
+        assert peer_c.drain() == ()
+
+        group_path: Final = f"/{group}/mcp"
+        group_listed, group_called = _list_and_call(
+            gateway, client, group_path, key, f"{alias_b}-multiply", {"a": 6, "b": 7}
+        )
+        assert group_listed.ok, group_listed.raw
+        assert _without_foreign_open_servers(group_listed.tools, open_aliases) == {
+            f"{alias_a}-add",
+            f"{alias_a}-multiply",
+            f"{alias_a}-fail",
+            f"{alias_b}-add",
+            f"{alias_b}-multiply",
+            f"{alias_b}-fail",
+        }, group_listed.tools
+        assert group_called.ok and group_called.text == "42", group_called.raw
+        assert tool_calls(peer_a.drain()) == ()
+        group_calls_b: Final = tool_calls(peer_b.drain())
+        assert len(group_calls_b) == 1, group_calls_b
+        _assert_upstream_call(group_calls_b[0], "multiply", {"a": 6, "b": 7})
+        assert tool_calls(peer_c.drain()) == ()
+
+        mixed_group_path: Final = f"/{group},{alias_c}/mcp"
+        mixed_group_listed, mixed_group_called = _list_and_call(
+            gateway, client, mixed_group_path, key, f"{alias_c}-add", {"a": 20, "b": 22}
+        )
+        assert mixed_group_listed.ok, mixed_group_listed.raw
+        assert _without_foreign_open_servers(mixed_group_listed.tools, open_aliases) == {
+            f"{alias_a}-add",
+            f"{alias_a}-multiply",
+            f"{alias_a}-fail",
+            f"{alias_b}-add",
+            f"{alias_b}-multiply",
+            f"{alias_b}-fail",
+            f"{alias_c}-add",
+            f"{alias_c}-multiply",
+            f"{alias_c}-fail",
+        }, mixed_group_listed.tools
+        assert mixed_group_called.ok and mixed_group_called.text == "42", mixed_group_called.raw
+        assert tool_calls(peer_a.drain()) == ()
+        assert tool_calls(peer_b.drain()) == ()
+        mixed_group_calls_c: Final = tool_calls(peer_c.drain())
+        assert len(mixed_group_calls_c) == 1, mixed_group_calls_c
+        _assert_upstream_call(mixed_group_calls_c[0], "add", {"a": 20, "b": 22})
+
+
+@pytest.mark.parametrize("client", ("raw-jsonrpc", "official-sdk"))
+def test_legacy_toolset_path_scopes_the_session_to_the_toolsets_tools(gateway: Gateway, client: str) -> None:
+    pytest.skip(
+        "BUG: a /{toolset}/mcp session lists every granted server's tools instead of the toolset's "
+        "tools (the toolset's tool filter is not applied on the legacy path)"
+    )
+    open_before: Final = _open_aliases()
+    with mcp_peer() as peer_a, mcp_peer() as peer_b, mcp_peer() as peer_c, gateway.scenario() as scenario:
+        alias_a: Final = "rt2ta" + uuid.uuid4().hex[:10]
+        alias_b: Final = "rt2tb" + uuid.uuid4().hex[:10]
+        alias_c: Final = "rt2tc" + uuid.uuid4().hex[:10]
+        server_a: Final = register_mcp(scenario, peer_a, alias_a)
+        server_b: Final = register_mcp(scenario, peer_b, alias_b)
+        server_c: Final = register_mcp(scenario, peer_c, alias_c)
+        toolset_name: Final = "rt2t" + uuid.uuid4().hex[:10]
+        toolset_id: Final = create_toolset(scenario, ((server_a, "add"),), toolset_name=toolset_name)
+        toolset_key: Final = scenario.key(
+            object_permission={
+                "mcp_servers": [server_a, server_b, server_c],
+                "mcp_toolsets": [toolset_id],
+            }
+        )
+        open_aliases: Final = open_before | _open_aliases()
+
+        toolset_path: Final = f"/{toolset_name}/mcp"
+        toolset_listed, toolset_called = _list_and_call(
+            gateway, client, toolset_path, toolset_key, f"{alias_a}-add", {"a": 20, "b": 22}
+        )
+        assert toolset_listed.ok, toolset_listed.raw
+        assert _without_foreign_open_servers(toolset_listed.tools, open_aliases) == {f"{alias_a}-add"}, (
+            toolset_listed.tools
+        )
+        assert toolset_called.ok and toolset_called.text == "42", toolset_called.raw
+        _refused_call(
+            gateway,
+            client,
+            toolset_path,
+            toolset_key,
+            f"{alias_a}-multiply",
+            {"a": 6, "b": 7},
+            f"Error: Tool 'multiply' is not allowed for your key/team on server '{alias_a}'. Contact proxy admin for access.",
+        )
+        _refused_call(gateway, client, toolset_path, toolset_key, f"{alias_b}-add", {"a": 20, "b": 22})
+        toolset_calls_a: Final = tool_calls(peer_a.drain())
+        assert len(toolset_calls_a) == 1, toolset_calls_a
+        _assert_upstream_call(toolset_calls_a[0], "add", {"a": 20, "b": 22})
+        assert tool_calls(peer_b.drain()) == ()
+        assert tool_calls(peer_c.drain()) == ()
+
+
 @pytest.mark.parametrize("peer_kind", ("http", "sse", "stdio"))
 def test_prompts_resources_and_templates_are_proxied_from_rich_peer(gateway: Gateway, peer_kind: PeerKind) -> None:
     with peer_of(peer_kind, rich=True) as peer, gateway.scenario() as scenario:
@@ -177,14 +572,19 @@ def test_pinned_revision_pairs_list_and_call_through_gateway(
         key: Final = scenario.key(object_permission={"mcp_servers": [identity]})
         endpoint: Final = str(gateway.client.base_url).rstrip("/") + ("/mcp/sse" if ingress == "sse" else "/mcp")
         client: Final = MCPClient(
-            server_url=endpoint, transport_type=MCPTransport(ingress), protocol_version=downstream,
-            extra_headers={"Authorization": f"Bearer {key}", "x-mcp-servers": identity}, timeout=15,
+            server_url=endpoint,
+            transport_type=MCPTransport(ingress),
+            protocol_version=downstream,
+            extra_headers={"Authorization": f"Bearer {key}", "x-mcp-servers": identity},
+            timeout=15,
         )
 
         async def exercise() -> None:
             tools: Final = await client.list_tools(raise_on_error=True)
             assert f"{alias}-add" in tuple(tool.name for tool in tools)
-            result: Final = await client.call_tool(CallToolRequestParams(name=f"{alias}-add", arguments={"a": 3, "b": 4}))
+            result: Final = await client.call_tool(
+                CallToolRequestParams(name=f"{alias}-add", arguments={"a": 3, "b": 4})
+            )
             assert result.is_error is False
             assert result.content[0].text == "7"
 
diff --git a/tests/integration/mcp/test_oauth_configuration.py b/tests/integration/mcp/test_oauth_configuration.py
index fe2b1069f04..e3a42d630cb 100644
--- a/tests/integration/mcp/test_oauth_configuration.py
+++ b/tests/integration/mcp/test_oauth_configuration.py
@@ -9,15 +9,263 @@ from pathlib import Path
 from typing import Final, Literal
 from urllib.parse import parse_qs, urlsplit
 
+import httpx
 import pytest
 from integration._support.client import Gateway, Scenario, eventually
 from integration._support.database import read_rows
 from integration._support.mcp import McpPeer, call_tool, mcp_peer, register_mcp, tool_names
+from integration._support.oauth_server import oauth_server
 from integration._support.process import owned_proxy
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import TypeAdapter
 
 _Upstream = Callable[[Request], Reply]
+CLIENT_REDIRECT: Final = "http://127.0.0.1:9/cb"
+
+
+def _response_object(response: httpx.Response) -> dict[str, object]:
+    return TypeAdapter(dict[str, object]).validate_python(response.json())
+
+
+def _start_callback_flow(
+    gateway: Gateway,
+    alias: str,
+    key: str,
+    client_id: str,
+    expected_authorization_url: str | None = None,
+) -> tuple[str, str, str]:
+    started: Final = gateway.client.get(
+        f"/{alias}/authorize",
+        params={
+            "client_id": client_id,
+            "redirect_uri": CLIENT_REDIRECT,
+            "response_type": "code",
+            "state": "client-state",
+            "code_challenge": "A" * 43,
+            "code_challenge_method": "S256",
+        },
+        headers={"x-litellm-api-key": key},
+    )
+    assert started.status_code in (302, 307), started.text
+    location: Final = urlsplit(started.headers["location"])
+    if expected_authorization_url is not None:
+        assert location.scheme + "://" + location.netloc + location.path == expected_authorization_url, started.text
+    relay_state: Final = parse_qs(location.query)["state"][0]
+    cookie_name: Final = f"mcp_oauth_state_{relay_state}"
+    assert cookie_name in started.cookies, started.headers
+    return relay_state, cookie_name, started.cookies[cookie_name]
+
+
+def _assert_cleared_oauth_state_cookie(response: httpx.Response, cookie_name: str) -> None:
+    cleared: Final = response.headers["set-cookie"]
+    assert cleared.startswith(f'{cookie_name}="";'), cleared
+    assert "Max-Age=0" in cleared, cleared
+    assert "Path=/" in cleared, cleared
+    assert "HttpOnly" in cleared, cleared
+    assert "SameSite=lax" in cleared, cleared
+
+
+def test_callback_forwards_the_code_only_for_the_sealed_issuer_and_relays_idp_errors(
+    gateway: Gateway,
+) -> None:
+    with mcp_peer() as peer, oauth_server() as auth, gateway.scenario() as scenario:
+        alias: Final = "rt4" + uuid.uuid4().hex[:10]
+        identity: Final = register_mcp(
+            scenario,
+            peer,
+            alias,
+            issuer=auth.issuer,
+            authorization_url=auth.issuer + "/authorize",
+            token_url=auth.issuer + "/token",
+            registration_url=auth.issuer + "/register",
+            auth_type="oauth2",
+            oauth2_flow="authorization_code",
+            credentials={"client_id": "ac-client", "client_secret": "ac-secret", "scopes": ["tools.call"]},
+        )
+        key: Final = scenario.key(object_permission={"mcp_servers": [identity]})
+        registered: Final = gateway.client.post(
+            f"/{alias}/register", json={"redirect_uris": [CLIENT_REDIRECT], "client_name": "integration"}
+        )
+        assert registered.status_code in (200, 201), registered.text
+        client_id: Final = str(_response_object(registered)["client_id"])
+
+        wrong_relay, wrong_cookie, wrong_cookie_value = _start_callback_flow(gateway, alias, key, client_id)
+        wrong_code: Final = "callback-code-" + uuid.uuid4().hex
+        wrong: Final = gateway.client.get(
+            "/callback",
+            params={"code": wrong_code, "state": wrong_relay, "iss": "http://127.0.0.1:1/other"},
+            cookies={wrong_cookie: wrong_cookie_value},
+        )
+        assert wrong.status_code == 400, wrong.text
+        assert wrong.headers["content-type"] == "text/html; charset=utf-8", wrong.headers
+        assert wrong.text == (
+            "<html><body><h2>Authentication failed</h2><p><strong>Error:</strong> invalid_issuer</p>"
+            "<p>This authorization response came from a different identity provider than the one this "
+            "MCP server is configured to use.</p>"
+            "<p>You can close this window and try again.</p></body></html>"
+        ), wrong.text
+        assert "location" not in wrong.headers, wrong.headers
+        assert wrong_code not in wrong.text, wrong.text
+        _assert_cleared_oauth_state_cookie(wrong, wrong_cookie)
+
+        correct_relay, correct_cookie, correct_cookie_value = _start_callback_flow(gateway, alias, key, client_id)
+        correct: Final = gateway.client.get(
+            "/callback",
+            params={"code": "c", "state": correct_relay, "iss": auth.issuer},
+            cookies={correct_cookie: correct_cookie_value},
+        )
+        assert correct.status_code == 302, correct.text
+        correct_location: Final = urlsplit(correct.headers["location"])
+        assert correct_location.scheme + "://" + correct_location.netloc + correct_location.path == CLIENT_REDIRECT, (
+            correct.headers["location"]
+        )
+        correct_query: Final = parse_qs(correct_location.query)
+        assert set(correct_query) == {"code", "state"}, correct.headers["location"]
+        assert correct_query["state"] == ["client-state"], correct.headers["location"]
+        assert len(correct_query["code"]) == 1 and correct_query["code"][0].startswith("llm_scope_"), correct.headers[
+            "location"
+        ]
+        _assert_cleared_oauth_state_cookie(correct, correct_cookie)
+
+        error_relay, error_cookie, error_cookie_value = _start_callback_flow(gateway, alias, key, client_id)
+        error: Final = gateway.client.get(
+            "/callback",
+            params={"error": "access_denied", "error_description": "x", "state": error_relay},
+            cookies={error_cookie: error_cookie_value},
+        )
+        assert error.status_code == 302, error.text
+        error_location: Final = urlsplit(error.headers["location"])
+        assert parse_qs(error_location.query) == {
+            "error": ["access_denied"],
+            "error_description": ["x"],
+            "state": ["client-state"],
+        }, error.headers["location"]
+        _assert_cleared_oauth_state_cookie(error, error_cookie)
+
+        error_issuer_relay, error_issuer_cookie, error_issuer_cookie_value = _start_callback_flow(
+            gateway, alias, key, client_id
+        )
+        error_issuer: Final = gateway.client.get(
+            "/callback",
+            params={
+                "error": "access_denied",
+                "error_description": "x",
+                "state": error_issuer_relay,
+                "iss": "http://127.0.0.1:1/other",
+            },
+            cookies={error_issuer_cookie: error_issuer_cookie_value},
+        )
+        assert error_issuer.status_code == 400, error_issuer.text
+        assert error_issuer.headers["content-type"] == "text/html; charset=utf-8", error_issuer.headers
+        assert error_issuer.text == (
+            "<html><body><h2>Authentication failed</h2><p><strong>Error:</strong> invalid_issuer</p>"
+            "<p>Unexpected authorization issuer</p>"
+            "<p>You can close this window and try again.</p></body></html>"
+        ), error_issuer.text
+        assert "location" not in error_issuer.headers, error_issuer.headers
+        _assert_cleared_oauth_state_cookie(error_issuer, error_issuer_cookie)
+
+        missing: Final = gateway.client.get("/callback")
+        assert missing.status_code == 400, missing.text
+        assert missing.headers["content-type"] == "text/html; charset=utf-8", missing.headers
+        assert missing.text == (
+            "<html><body><h2>Authentication failed</h2><p><strong>Error:</strong> invalid_request</p>"
+            "<p>Missing authorization &#x27;code&#x27; and &#x27;state&#x27; parameter(s).</p>"
+            "<p>You can close this window and try again.</p></body></html>"
+        ), missing.text
+
+
+def test_callback_refuses_a_response_without_iss_when_the_authorization_server_advertises_iss_support(
+    gateway: Gateway,
+) -> None:
+    wire_holder: Final[list[Wire]] = []
+
+    def respond(request: Request) -> Reply:
+        path: Final = urlsplit(request.target).path
+        if request.method == "GET" and path in (
+            "/.well-known/oauth-authorization-server",
+            "/.well-known/openid-configuration",
+        ):
+            issuer: Final = wire_holder[0].url
+            return Reply(
+                body=json.dumps(
+                    {
+                        "issuer": issuer,
+                        "authorization_endpoint": issuer + "/authorize",
+                        "token_endpoint": issuer + "/token",
+                        "registration_endpoint": issuer + "/register",
+                        "response_types_supported": ["code"],
+                        "grant_types_supported": ["authorization_code", "refresh_token"],
+                        "code_challenge_methods_supported": ["S256"],
+                        "token_endpoint_auth_methods_supported": ["client_secret_post"],
+                        "authorization_response_iss_parameter_supported": True,
+                    }
+                ).encode()
+            )
+        return Reply(status=404, body=json.dumps({"error": "not_found"}).encode())
+
+    with wire_server(respond) as wire, mcp_peer() as peer, gateway.scenario() as scenario:
+        wire_holder.append(wire)
+        alias: Final = "rt5" + uuid.uuid4().hex[:10]
+        identity: Final = register_mcp(
+            scenario,
+            peer,
+            alias,
+            issuer=wire.url,
+            auth_type="oauth2",
+            oauth2_flow="authorization_code",
+            credentials={"client_id": "ac-client", "client_secret": "ac-secret", "scopes": ["tools.call"]},
+        )
+        key: Final = scenario.key(object_permission={"mcp_servers": [identity]})
+
+        relay, cookie, cookie_value = _start_callback_flow(
+            gateway, alias, key, "ac-client", expected_authorization_url=wire.url + "/authorize"
+        )
+        code: Final = "callback-code-" + uuid.uuid4().hex
+        missing_issuer: Final = gateway.client.get(
+            "/callback",
+            params={"code": code, "state": relay},
+            cookies={cookie: cookie_value},
+        )
+        assert missing_issuer.status_code == 400, missing_issuer.text
+        assert missing_issuer.headers["content-type"] == "text/html; charset=utf-8", missing_issuer.headers
+        assert missing_issuer.text == (
+            "<html><body><h2>Authentication failed</h2><p><strong>Error:</strong> invalid_issuer</p>"
+            "<p>This authorization response came from a different identity provider than the one this "
+            "MCP server is configured to use.</p>"
+            "<p>You can close this window and try again.</p></body></html>"
+        ), missing_issuer.text
+        assert "location" not in missing_issuer.headers, missing_issuer.headers
+        assert code not in missing_issuer.text, missing_issuer.text
+        _assert_cleared_oauth_state_cookie(missing_issuer, cookie)
+
+        trusted_relay, trusted_cookie, trusted_cookie_value = _start_callback_flow(
+            gateway, alias, key, "ac-client", expected_authorization_url=wire.url + "/authorize"
+        )
+        trusted_code: Final = "callback-code-" + uuid.uuid4().hex
+        trusted: Final = gateway.client.get(
+            "/callback",
+            params={"code": trusted_code, "state": trusted_relay, "iss": wire.url},
+            cookies={trusted_cookie: trusted_cookie_value},
+        )
+        assert trusted.status_code == 302, trusted.text
+        trusted_location: Final = urlsplit(trusted.headers["location"])
+        assert trusted_location.scheme + "://" + trusted_location.netloc + trusted_location.path == CLIENT_REDIRECT, (
+            trusted.headers["location"]
+        )
+        trusted_query: Final = parse_qs(trusted_location.query)
+        assert set(trusted_query) == {"code", "state"}, trusted.headers["location"]
+        assert trusted_query["state"] == ["client-state"], trusted.headers["location"]
+        assert len(trusted_query["code"]) == 1 and trusted_query["code"][0].startswith("llm_scope_"), trusted.headers[
+            "location"
+        ]
+        assert trusted_query["code"][0] != trusted_code, trusted.headers["location"]
+        _assert_cleared_oauth_state_cookie(trusted, trusted_cookie)
+
+        wire_requests: Final = wire.drain()
+        wire_method_paths: Final = tuple((request.method, urlsplit(request.target).path) for request in wire_requests)
+        assert wire_method_paths == (("GET", "/.well-known/oauth-authorization-server"),), wire_method_paths
+        assert not any(method == "POST" and path == "/token" for method, path in wire_method_paths), wire_method_paths
 
 
 @pytest.mark.covers("other.mcp.oauth.discovery_cannot_erase_configured_authorization_endpoint")