test(e2e): cover MCP core happy paths: upstream auth, access control, allowed_tools, aggregate routing

2026-10-11 03:38:38 +00:00 · 2026-07-14 21:02:03 -07:00 · 2026-07-14 21:02:03 -07:00 · 59fa990c86
commit 59fa990c86
parent b92b7cc943
9 changed files with 565 additions and 18 deletions
--- a/tests/e2e/CLAUDE.md
+++ b/tests/e2e/CLAUDE.md
@ -13,7 +13,7 @@ Each subdirectory under `tests/e2e/` is one suite, scoped to an endpoint family
 - `realtime/` - realtime websocket sessions, including the pipecat audio path
 - `quota_management/` - quota enforcement and accounting, one subfolder per behavior: `ratelimit/` (rpm/tpm blocks, window reset, pacing headers on live traffic), `budgets/` (budget definition, enforcement, and reset windows: key, team, tag, soft, multi-window), and `spend_tracking/` (spend logging and cost attribution on `/spend/*`)
 - `management/` - key/team/user/organization management routes: create/update/delete persistence via the info routes, team membership, and llm-only-key route denials; also the dashboard UI behavior on top of them, driven through the proxy-served UI at /ui with playwright (optional dep behind importorskip)
- `mcp/` - the MCP gateway: server registration via `/v1/mcp/server`, tool listing/calling over the streamable-http protocol under both auth headers, the interactive (authorization_code) OAuth flow driven with the mcp SDK's own OAuth client, and per-server enforcement such as `max_concurrent_requests`; `stub/` holds the deterministic upstream MCP mounts and stub OAuth2 IdP the compose stack runs for it
+- `mcp/` - the MCP gateway: server registration via `/v1/mcp/server`, tool listing/calling over the streamable-http protocol under both auth headers (per-server and aggregate `/mcp` namespaces), upstream credential injection (static api_key and OAuth2 client_credentials), access control (object_permission grants, `allowed_tools`), and per-server enforcement such as `max_concurrent_requests`; `stub/` holds the deterministic upstream MCP mounts (anonymous, disjoint-toolset, credential-guarded) plus the stub OAuth2 token endpoint the compose stack runs for it
 - `logging/` - logging-integration delivery (datadog and friends)
 - `security/` - secret handling and log-leak protection
 - `router/` - routing and reliability behavior (fallbacks, cooldowns)
--- a/tests/e2e/coverage_registry/mcp.yaml
+++ b/tests/e2e/coverage_registry/mcp.yaml
@ -135,6 +135,46 @@
  assertions: [succeeds]
  source: "server.py:1089"
  rationale: Smoke; rarely used; same auth model as tools
+- id: mcp.call_tool.api_key.injects_upstream_credential
+  module: mcp
+  tier: P0
+  operation: call_tool
+  auth_family: api_key
+  assertions: [injects_upstream_credential]
+  source: "outbound_credentials/adapter.py:81"
+  rationale: Stored shared-key credential must reach the upstream as X-API-Key and the caller's virtual key must not; most common upstream-auth config
+- id: mcp.list_tools.api_key.aggregates_servers
+  module: mcp
+  tier: P0
+  operation: list_tools
+  auth_family: api_key
+  assertions: [aggregates_servers]
+  source: "server.py:1921 + auth/user_api_key_auth_mcp.py:722"
+  rationale: The aggregate /mcp namespace with x-mcp-servers scoping is how production MCP hosts (Claude Code, Cursor) mount the gateway
+- id: mcp.call_tool.api_key.routes_to_target_server
+  module: mcp
+  tier: P0
+  operation: call_tool
+  auth_family: api_key
+  assertions: [routes_to_target_server]
+  source: "server.py:2625"
+  rationale: An aggregate-session call must dispatch to the server that owns the prefixed tool, never a sibling
+- id: mcp.list_tools.api_key.filters_allowed_tools
+  module: mcp
+  tier: P0
+  operation: list_tools
+  auth_family: api_key
+  assertions: [filters_allowed_tools]
+  source: "server.py:1352 filter_tools_by_allowed_tools"
+  rationale: Server-level allowed_tools is the admin's tool-governance surface; a filter regression exposes tools the admin excluded
+- id: mcp.call_tool.api_key.blocks_tool_outside_allowed_tools
+  module: mcp
+  tier: P0
+  operation: call_tool
+  auth_family: api_key
+  assertions: [blocks_tool_outside_allowed_tools]
+  source: "mcp_server_manager.py:3454 check_allowed_or_banned_tools"
+  rationale: Filtering the listing alone is cosmetic; the call path must reject excluded tools or governance is bypassable by name
 - id: mcp.list_tools.oauth.completes_authorization_code_flow
  module: mcp
  tier: P0
--- a/tests/e2e/e2e_config.py
+++ b/tests/e2e/e2e_config.py
@ -42,10 +42,15 @@ DD_SINK_URL = os.environ.get("E2E_DD_SINK_URL", "http://localhost:9915").rstrip(
 # test runs somewhere the compose stub is not visible from.
 MCP_STUB_URL = os.environ.get("E2E_MCP_STUB_URL", "http://mcp-stub:8765/mcp")

-# The stub's sibling surfaces (see mcp/stub/stub_server.py): a Bearer-guarded
-# upstream for the interactive OAuth flow, plus the stub IdP's token endpoint.
-# Derived from MCP_STUB_URL so one override relocates the whole stub.
+# The stub's sibling mounts (see mcp/stub/stub_server.py): a second anonymous
+# upstream with a disjoint tool set, an X-API-Key-guarded upstream, a
+# Bearer-guarded upstream, and the OAuth2 client_credentials token endpoint
+# that mints the only token the Bearer guard accepts. Derived from MCP_STUB_URL
+# so one override relocates the whole stub.
 _MCP_STUB_BASE = MCP_STUB_URL.removesuffix("/mcp")
+MCP_STUB_SECOND_URL = f"{_MCP_STUB_BASE}/second/mcp"
+MCP_STUB_APIKEY_URL = f"{_MCP_STUB_BASE}/apikey/mcp"
+MCP_STUB_OAUTH_URL = f"{_MCP_STUB_BASE}/oauth/mcp"
 MCP_STUB_OAUTHUSER_URL = f"{_MCP_STUB_BASE}/oauthuser/mcp"
 MCP_STUB_TOKEN_URL = f"{_MCP_STUB_BASE}/oauth/token"

@ -58,6 +63,11 @@ MCP_STUB_AUTHORIZE_BROWSER_URL = os.environ.get(
 )

 # Deterministic test-only credentials; must mirror mcp/stub/stub_server.py.
+MCP_STUB_UPSTREAM_API_KEY = "e2e-stub-upstream-api-key"
+MCP_STUB_OAUTH_CLIENT_ID = "e2e-stub-oauth-client-id"
+MCP_STUB_OAUTH_CLIENT_SECRET = "e2e-stub-oauth-client-secret"
+MCP_STUB_OAUTH_SCOPE = "tools:read"
+MCP_STUB_OAUTH_ACCESS_TOKEN = "e2e-stub-minted-access-token"
 MCP_STUB_OAUTH_USER_CLIENT_ID = "e2e-stub-user-client-id"
 MCP_STUB_OAUTH_USER_CLIENT_SECRET = "e2e-stub-user-client-secret"
 MCP_STUB_OAUTH_USER_ACCESS_TOKEN = "e2e-stub-user-access-token"
--- a/tests/e2e/mcp/mcp_client.py
+++ b/tests/e2e/mcp/mcp_client.py
@ -3,11 +3,14 @@
 Management routes (/v1/mcp/server CRUD) go through the shared Gateway
 transport. The MCP protocol itself (initialize, tools/list, tools/call over
 streamable HTTP) goes through the official mcp SDK, the same client library
-production MCP hosts use, aimed at the gateway's per-server URL namespace
-{PROXY}/{alias}/mcp.
+production MCP hosts use, aimed at either of the gateway's two URL namespaces:
+the per-server {PROXY}/{alias}/mcp, or the aggregate {PROXY}/mcp where one
+session spans every server the key may use and the `x-mcp-servers` header
+narrows it to named aliases.

 Every protocol method takes the request headers as a plain dict, built inside
-the test body, so the exact wire format is visible where it is asserted. The gateway accepts the LiteLLM
+the test body (including `x-mcp-servers` for aggregate scoping), so the exact
+wire format is visible where it is asserted. The gateway accepts the LiteLLM
 virtual key as either `x-litellm-api-key: Bearer sk-...` or
 `Authorization: Bearer sk-...` (both Bearer-prefixed on the MCP routes,
 matching the docs).
@ -62,6 +65,9 @@ class McpToolText:
    is_error: bool


+CallToolOutcome = McpToolText | McpDenied
+
+
 class StubToolStats(BaseModel):
    """JSON the stub's `stats` tool returns for one marker (see stub/stub_server.py)."""

@ -79,6 +85,10 @@ def _mcp_url(alias: str) -> str:
    return f"{PROXY_BASE_URL}/{alias}/mcp"


+def _aggregate_url() -> str:
+    return f"{PROXY_BASE_URL}/mcp"
+
+
 def _first_text(result: CallToolResult) -> str:
    first = result.content[0] if result.content else None
    if isinstance(first, TextContent):
@ -312,6 +322,47 @@ class McpClient:
        behave like independent clients."""
        return asyncio.run(_call_tool(_mcp_url(alias), headers, tool, arguments))

+    def call_tool_once(
+        self, alias: str, headers: dict[str, str], tool: str, arguments: ToolArguments
+    ) -> CallToolOutcome:
+        """A tools/call whose rejection is an outcome, not a crash: the gateway
+        surfaces permission denials through the protocol stream, which the SDK
+        raises client-side; denial tests assert on the McpDenied value."""
+        try:
+            return self.call_tool(alias, headers, tool, arguments)
+        except Exception as exc:  # noqa: BLE001 - the SDK raises ExceptionGroup-wrapped transport errors; modelled as a value
+            return McpDenied(status_code=_http_status(exc), message=str(exc))
+
+    def aggregate_list_tools_once(self, headers: dict[str, str]) -> ListToolsOutcome:
+        """tools/list over one aggregate {PROXY}/mcp session; the caller builds
+        the `x-mcp-servers` header when it wants to narrow the session."""
+        try:
+            return McpToolNames(names=asyncio.run(_list_tool_names(_aggregate_url(), headers)))
+        except Exception as exc:  # noqa: BLE001 - the SDK raises ExceptionGroup-wrapped transport errors; modelled as a value
+            return McpDenied(status_code=_http_status(exc), message=str(exc))
+
+    def poll_aggregate_tool_names(self, headers: dict[str, str], *, until_listed: str) -> tuple[str, ...]:
+        """Aggregate tools/list to the shared deadline, until `until_listed`
+        appears (the settle signal that the last-created record has propagated
+        to the gateway); returns that full listing."""
+        deadline = time.monotonic() + self.gateway.poll_timeout
+        outcome: ListToolsOutcome = McpDenied(status_code=None, message="never attempted")
+        while time.monotonic() < deadline:
+            outcome = self.aggregate_list_tools_once(headers)
+            match outcome:
+                case McpToolNames(names=names) if until_listed in names:
+                    return names
+                case _:
+                    time.sleep(self.gateway.poll_interval)
+        pytest.fail(
+            f"aggregate MCP listing never included {until_listed!r} "
+            f"within {self.gateway.poll_timeout}s; last outcome: {outcome}"
+        )
+
+    def aggregate_call_tool(self, headers: dict[str, str], tool: str, arguments: ToolArguments) -> McpToolText:
+        """One tools/call over its own fresh aggregate {PROXY}/mcp session."""
+        return asyncio.run(_call_tool(_aggregate_url(), headers, tool, arguments))
+
    def poll_oauth_tool_names(
        self, alias: str, headers: dict[str, str], storage: InMemoryTokenStorage
    ) -> tuple[str, ...]:
--- a/tests/e2e/mcp/stub/stub_server.py
+++ b/tests/e2e/mcp/stub/stub_server.py
@ -1,6 +1,6 @@
 """Deterministic MCP upstreams for the mcp e2e suite.

-One process, one port, two streamable-http MCP mounts plus a deterministic
+One process, one port, five streamable-http MCP mounts plus a deterministic
 OAuth2 IdP, so the compose stack keeps a single `mcp-stub` service:

 - `/mcp` — anonymous. `echo` answers immediately so auth tests can assert an
@ -10,7 +10,16 @@ OAuth2 IdP, so the compose stack keeps a single `mcp-stub` service:
  `max_concurrent_requests` cap must bound); `stats` reads those counters back,
  so tests observe upstream concurrency through the proxy itself and the stub
  needs no side-channel port.
- `/oauthuser/mcp` — the interactive (authorization_code) upstream: rejects
+- `/second/mcp` — anonymous, with a deliberately disjoint tool set
+  (`second_ping`). Aggregate-routing tests register it as a second gateway
+  server; a call only this mount can answer proves which upstream served it.
+- `/apikey/mcp` — rejects any request whose `X-API-Key` is not exactly
+  UPSTREAM_API_KEY, the header the gateway injects for `auth_type: api_key`.
+- `/oauth/mcp` — rejects any request whose `Authorization` is not exactly
+  `Bearer OAUTH_ACCESS_TOKEN`. That token is only obtainable from
+  `/oauth/token`, so a served request proves the gateway ran the
+  client_credentials exchange rather than forwarding something it already had.
+- `/oauthuser/mcp` — the interactive (authorization_code) sibling: rejects
  anything but `Bearer OAUTH_USER_ACCESS_TOKEN`, which only the
  authorization_code grant hands out, so a served request proves the whole
  browser dance (authorize redirect, code, PKCE-verified token exchange) ran.
@ -18,14 +27,15 @@ OAuth2 IdP, so the compose stack keeps a single `mcp-stub` service:
  client_id/response_type, records the one-time code with its PKCE challenge,
  and 302s straight back to the caller's redirect_uri with code and state (the
  "user" of this IdP always consents instantly).
- `/oauth/token` — the token endpoint: authorization_code validates the code,
-  redirect_uri, client credentials, and (when a challenge was recorded) the
-  S256 code_verifier, then answers with OAUTH_USER_ACCESS_TOKEN +
-  OAUTH_USER_REFRESH_TOKEN; refresh_token re-issues OAUTH_USER_ACCESS_TOKEN
-  for the known refresh token.
+- `/oauth/token` — the token endpoint for all three grants:
+  client_credentials answers with OAUTH_ACCESS_TOKEN;
+  authorization_code validates the code, redirect_uri, client credentials,
+  and (when a challenge was recorded) the S256 code_verifier, then answers
+  with OAUTH_USER_ACCESS_TOKEN + OAUTH_USER_REFRESH_TOKEN; refresh_token
+  re-issues OAUTH_USER_ACCESS_TOKEN for the known refresh token.

-The guarded mount records the headers of the most recent authorized request;
-its `recorded_headers` tool reads them back through the proxy, so tests can
+The guarded mounts record the headers of the most recent authorized request;
+their `recorded_headers` tool reads them back through the proxy, so tests can
 assert exactly which credentials the gateway attached upstream (and that the
 caller's LiteLLM virtual key never left the gateway).

@ -55,12 +65,20 @@ from starlette.responses import JSONResponse, RedirectResponse, Response
 from starlette.routing import Mount, Route
 from starlette.types import ASGIApp, Receive, Scope, Send

+UPSTREAM_API_KEY = "e2e-stub-upstream-api-key"
+OAUTH_CLIENT_ID = "e2e-stub-oauth-client-id"
+OAUTH_CLIENT_SECRET = "e2e-stub-oauth-client-secret"
+OAUTH_SCOPE = "tools:read"
+OAUTH_ACCESS_TOKEN = "e2e-stub-minted-access-token"
 OAUTH_USER_CLIENT_ID = "e2e-stub-user-client-id"
 OAUTH_USER_CLIENT_SECRET = "e2e-stub-user-client-secret"
 OAUTH_USER_ACCESS_TOKEN = "e2e-stub-user-access-token"
 OAUTH_USER_REFRESH_TOKEN = "e2e-stub-user-refresh-token"

 main_mcp = FastMCP("e2e-stub", host="0.0.0.0", port=8765, stateless_http=True)
+second_mcp = FastMCP("e2e-stub-second", host="0.0.0.0", port=8765, stateless_http=True)
+apikey_mcp = FastMCP("e2e-stub-apikey", host="0.0.0.0", port=8765, stateless_http=True)
+oauth_mcp = FastMCP("e2e-stub-oauth", host="0.0.0.0", port=8765, stateless_http=True)
 oauthuser_mcp = FastMCP("e2e-stub-oauthuser", host="0.0.0.0", port=8765, stateless_http=True)


@ -109,6 +127,10 @@ def stats(marker: str) -> str:
    )


+@second_mcp.tool()
+def second_ping() -> str:
+    """Identify the /second upstream; no other mount serves this tool."""
+    return "pong-from-second"
 def _register_guarded_tools(server: FastMCP, mount: str) -> None:
    def echo(text: str) -> str:
        """Return `text` unchanged."""
@ -122,6 +144,8 @@ def _register_guarded_tools(server: FastMCP, mount: str) -> None:
    _ = server.tool()(recorded_headers)


+_register_guarded_tools(apikey_mcp, "apikey")
+_register_guarded_tools(oauth_mcp, "oauth")
 _register_guarded_tools(oauthuser_mcp, "oauthuser")


@ -216,6 +240,15 @@ async def oauth_token(request: Request) -> JSONResponse:
    exactly so a failure points at the precise field the proxy sent wrong."""
    form = await request.form()
    grant_type = form.get("grant_type")
+    if grant_type == "client_credentials":
+        granted = (
+            form.get("client_id") == OAUTH_CLIENT_ID
+            and form.get("client_secret") == OAUTH_CLIENT_SECRET
+            and form.get("scope") == OAUTH_SCOPE
+        )
+        if not granted:
+            return JSONResponse({"error": "invalid_client"}, status_code=401)
+        return JSONResponse({"access_token": OAUTH_ACCESS_TOKEN, "token_type": "Bearer", "expires_in": 3600})
    if grant_type == "authorization_code":
        return _authorization_code_grant(form)
    if grant_type == "refresh_token":
@ -224,7 +257,7 @@ async def oauth_token(request: Request) -> JSONResponse:


 def build_app() -> Starlette:
-    servers = (main_mcp, oauthuser_mcp)
+    servers = (main_mcp, second_mcp, apikey_mcp, oauth_mcp, oauthuser_mcp)
    apps = {server.name: server.streamable_http_app() for server in servers}

    @contextlib.asynccontextmanager
@ -247,6 +280,25 @@ def build_app() -> Starlette:
                    expected=f"Bearer {OAUTH_USER_ACCESS_TOKEN}",
                ),
            ),
+            Mount(
+                "/oauth",
+                app=_require_header(
+                    apps["e2e-stub-oauth"],
+                    mount="oauth",
+                    header="authorization",
+                    expected=f"Bearer {OAUTH_ACCESS_TOKEN}",
+                ),
+            ),
+            Mount(
+                "/apikey",
+                app=_require_header(
+                    apps["e2e-stub-apikey"],
+                    mount="apikey",
+                    header="x-api-key",
+                    expected=UPSTREAM_API_KEY,
+                ),
+            ),
+            Mount("/second", app=apps["e2e-stub-second"]),
            Mount("/", app=apps["e2e-stub"]),
        ],
        lifespan=lifespan,
--- a/tests/e2e/mcp/test_mcp_access_control_e2e.py
+++ b/tests/e2e/mcp/test_mcp_access_control_e2e.py
@ -0,0 +1,147 @@
+"""Live e2e: the MCP gateway's two governance axes.
+
+Covers mcp.list_tools.api_key.denied_without_permission +
+mcp.call_tool.api_key.denied_without_permission (server-level access: a key
+without an object_permission grant for a non-allow-all server must neither
+see its tools nor call them) and mcp.list_tools.api_key.filters_allowed_tools
+ mcp.call_tool.api_key.blocks_tool_outside_allowed_tools (tool-level access:
+a server's `allowed_tools` subset must bound both the listing and the call
+path), all against the deterministic mcp-stub compose service.
+
+The denial test is built so it cannot pass for the wrong reason. A sibling
+key with the grant drives the same server through the same machinery first
+(so record propagation and a working grant path are proven), and the denied
+key first lists an allow-all control server (so the key itself is proven
+valid and propagated). Only then do the denial assertions run: whatever the
+denied key is refused on cannot be blamed on propagation lag or a broken key,
+which is the fail-before-fix evidence for the permission guard; removing the
+guard makes the granted-vs-denied outcomes identical and the test fail.
+
+The allowed_tools test asserts the exact filtered listing (a broken filter
+serves all three stub tools and fails the equality immediately) and that the
+excluded tool is refused on the call path, since filtering only the listing
+would leave governance bypassable by anyone who knows a tool's name.
+
+Both denial contracts are pinned to the gateway's observed live behavior: a
+denied listing is a served-but-empty tool list (the session is admitted, the
+tools are filtered), and a denied call is an in-band tool error ("not allowed
+to call this tool" / "is not allowed for server"), not a transport-level 4xx,
+because a mid-session JSON-RPC call cannot carry an HTTP status.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from e2e_config import MCP_STUB_URL, unique_marker
+from lifecycle import ResourceManager
+from mcp_client import McpClient, McpToolNames, McpToolText
+from models import KeyGenerateBody, KeyObjectPermission, McpServerCreateBody
+
+pytestmark = pytest.mark.e2e
+
+STUB_TOOLS = ("echo", "slow_echo", "stats")
+
+
+class TestMcpServerAccessControl:
+    """A non-allow-all server is invisible and uncallable to keys without its
+    object_permission grant, while a granted key uses it normally."""
+
+    @pytest.mark.covers("mcp.list_tools.api_key.denied_without_permission")
+    @pytest.mark.covers("mcp.call_tool.api_key.denied_without_permission")
+    def test_key_without_grant_sees_no_tools_and_cannot_call(
+        self, client: McpClient, resources: ResourceManager
+    ) -> None:
+        marker = unique_marker()
+        guarded_alias = f"e2emcpacl{marker}"
+        guarded = client.create_server(
+            McpServerCreateBody(alias=guarded_alias, url=MCP_STUB_URL, allow_all_keys=False)
+        )
+        resources.defer(lambda: client.delete_server(guarded.server_id))
+
+        control_alias = f"e2emcpaclctl{marker}"
+        control = client.create_server(
+            McpServerCreateBody(alias=control_alias, url=MCP_STUB_URL, allow_all_keys=True)
+        )
+        resources.defer(lambda: client.delete_server(control.server_id))
+
+        assert client.server_info(guarded.server_id).allow_all_keys is False
+
+        granted_key = client.gateway.generate_key(
+            KeyGenerateBody(object_permission=KeyObjectPermission(mcp_servers=[guarded.server_id]))
+        )
+        resources.defer(lambda: client.gateway.delete_key(granted_key))
+        denied_key = client.gateway.generate_key(KeyGenerateBody())
+        resources.defer(lambda: client.gateway.delete_key(denied_key))
+
+        granted_headers = {"x-litellm-api-key": f"Bearer {granted_key}"}
+        denied_headers = {"x-litellm-api-key": f"Bearer {denied_key}"}
+
+        names = client.poll_tool_names(guarded_alias, granted_headers)
+        expected = tuple(sorted(f"{guarded_alias}-{tool}" for tool in STUB_TOOLS))
+        assert names == expected, f"granted key listed {names}, expected exactly {expected}"
+
+        payload = f"e2e-{marker}"
+        result = client.call_tool(guarded_alias, granted_headers, f"{guarded_alias}-echo", {"text": payload})
+        assert result.is_error is False, f"granted key's echo errored: {result.text[:300]}"
+        assert result.text == payload
+
+        _ = client.poll_tool_names(control_alias, denied_headers)
+
+        denied_list = client.list_tools_once(guarded_alias, denied_headers)
+        assert denied_list == McpToolNames(names=()), (
+            f"key without the grant must be served an empty listing, got: {denied_list}"
+        )
+
+        denied_call = client.call_tool_once(guarded_alias, denied_headers, f"{guarded_alias}-echo", {"text": payload})
+        match denied_call:
+            case McpToolText(is_error=True, text=text) if "not allowed to call this tool" in text:
+                pass
+            case other:
+                pytest.fail(f"expected the in-band 'not allowed to call this tool' error, got: {other}")
+
+
+class TestMcpAllowedToolsFilter:
+    """`allowed_tools` bounds the server to a subset of its upstream tools, on
+    the listing and on the call path."""
+
+    @pytest.mark.covers("mcp.list_tools.api_key.filters_allowed_tools")
+    @pytest.mark.covers("mcp.call_tool.api_key.blocks_tool_outside_allowed_tools")
+    def test_allowed_tools_filters_listing_and_blocks_excluded_call(
+        self, client: McpClient, resources: ResourceManager
+    ) -> None:
+        alias = f"e2emcptoolgov{unique_marker()}"
+        created = client.create_server(
+            McpServerCreateBody(
+                alias=alias,
+                url=MCP_STUB_URL,
+                allow_all_keys=True,
+                allowed_tools=["echo", "stats"],
+            )
+        )
+        resources.defer(lambda: client.delete_server(created.server_id))
+
+        stored = client.server_info(created.server_id)
+        assert stored.allowed_tools == ["echo", "stats"]
+
+        key = client.gateway.generate_key(KeyGenerateBody())
+        resources.defer(lambda: client.gateway.delete_key(key))
+
+        headers = {"x-litellm-api-key": f"Bearer {key}"}
+        names = client.poll_tool_names(alias, headers)
+        expected = tuple(sorted((f"{alias}-echo", f"{alias}-stats")))
+        assert names == expected, f"allowed_tools listing was {names}, expected exactly {expected}"
+
+        payload = f"e2e-{unique_marker()}"
+        result = client.call_tool(alias, headers, f"{alias}-echo", {"text": payload})
+        assert result.is_error is False, f"allowed tool errored: {result.text[:300]}"
+        assert result.text == payload
+
+        blocked = client.call_tool_once(
+            alias, headers, f"{alias}-slow_echo", {"text": payload, "marker": unique_marker(), "sleep_seconds": 0}
+        )
+        match blocked:
+            case McpToolText(is_error=True, text=text) if "is not allowed for server" in text:
+                pass
+            case other:
+                pytest.fail(f"expected the in-band 'not allowed for server' error for the excluded tool, got: {other}")
--- a/tests/e2e/mcp/test_mcp_aggregate_e2e.py
+++ b/tests/e2e/mcp/test_mcp_aggregate_e2e.py
@ -0,0 +1,97 @@
+"""Live e2e: the aggregate /mcp namespace over multiple servers.
+
+Covers mcp.list_tools.api_key.aggregates_servers +
+mcp.call_tool.api_key.routes_to_target_server: one MCP session at
+{PROXY}/mcp, scoped with the `x-mcp-servers` header, must list the
+alias-prefixed union of every named server's tools and dispatch each call to
+the server that owns the prefixed tool. This is the mount shape production
+MCP hosts use (one gateway endpoint, many upstreams), where per-server URLs
+cannot catch cross-server routing bugs by construction.
+
+The two registered servers point at stub mounts with deliberately disjoint
+tool sets (tests/e2e/mcp/stub/): only the /second mount serves `second_ping`.
+A successful `{second_alias}-second_ping` call answered with the /second
+mount's canned reply is therefore proof of correct dispatch, and the
+fail-before-fix evidence: a gateway that routed by anything other than the
+prefixed owner would hit the main mount, which cannot answer that tool. The
+scoped re-listing at the end pins the header contract (naming one alias must
+hide the sibling's tools), which keeps the exact-equality listing assertions
+meaningful when other suites register their own servers concurrently.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from e2e_config import MCP_STUB_SECOND_URL, MCP_STUB_URL, unique_marker
+from lifecycle import ResourceManager
+from mcp_client import McpClient, McpDenied, McpToolNames
+from models import KeyGenerateBody, McpServerCreateBody
+
+pytestmark = pytest.mark.e2e
+
+STUB_TOOLS = ("echo", "slow_echo", "stats")
+SECOND_STUB_TOOLS = ("second_ping",)
+
+
+class TestMcpAggregateNamespace:
+    """One aggregate session spans every server named in `x-mcp-servers`,
+    lists their tools under alias prefixes, and routes calls by prefix."""
+
+    @pytest.mark.covers("mcp.list_tools.api_key.aggregates_servers")
+    @pytest.mark.covers("mcp.call_tool.api_key.routes_to_target_server")
+    def test_one_session_lists_both_servers_and_routes_calls(
+        self, client: McpClient, resources: ResourceManager
+    ) -> None:
+        marker = unique_marker()
+        first_alias = f"e2emcpagga{marker}"
+        first = client.create_server(McpServerCreateBody(alias=first_alias, url=MCP_STUB_URL, allow_all_keys=True))
+        resources.defer(lambda: client.delete_server(first.server_id))
+
+        second_alias = f"e2emcpaggb{marker}"
+        second = client.create_server(
+            McpServerCreateBody(alias=second_alias, url=MCP_STUB_SECOND_URL, allow_all_keys=True)
+        )
+        resources.defer(lambda: client.delete_server(second.server_id))
+
+        assert client.server_info(first.server_id).url == MCP_STUB_URL
+        assert client.server_info(second.server_id).url == MCP_STUB_SECOND_URL
+
+        key = client.gateway.generate_key(KeyGenerateBody())
+        resources.defer(lambda: client.gateway.delete_key(key))
+
+        headers = {
+            "x-litellm-api-key": f"Bearer {key}",
+            "x-mcp-servers": f"{first_alias},{second_alias}",
+        }
+
+        _ = client.poll_aggregate_tool_names(headers, until_listed=f"{first_alias}-echo")
+        names = client.poll_aggregate_tool_names(headers, until_listed=f"{second_alias}-second_ping")
+        expected = tuple(
+            sorted(
+                [f"{first_alias}-{tool}" for tool in STUB_TOOLS]
+                + [f"{second_alias}-{tool}" for tool in SECOND_STUB_TOOLS]
+            )
+        )
+        assert names == expected, f"aggregate session listed {names}, expected exactly {expected}"
+
+        payload = f"e2e-{marker}"
+        echoed = client.aggregate_call_tool(headers, f"{first_alias}-echo", {"text": payload})
+        assert echoed.is_error is False, f"aggregate echo errored: {echoed.text[:300]}"
+        assert echoed.text == payload
+
+        pinged = client.aggregate_call_tool(headers, f"{second_alias}-second_ping", {})
+        assert pinged.is_error is False, f"aggregate second_ping errored: {pinged.text[:300]}"
+        assert pinged.text == "pong-from-second", (
+            f"call must dispatch to the /second upstream that owns the tool, got {pinged.text!r}"
+        )
+
+        second_only_headers = {"x-litellm-api-key": f"Bearer {key}", "x-mcp-servers": second_alias}
+        scoped = client.aggregate_list_tools_once(second_only_headers)
+        match scoped:
+            case McpToolNames(names=only):
+                assert only == (f"{second_alias}-second_ping",), (
+                    f"x-mcp-servers scoped to {second_alias} must hide the sibling's tools, listed {only}"
+                )
+            case McpDenied() as denied:
+                pytest.fail(f"scoped aggregate listing was refused: {denied}")
--- a/tests/e2e/mcp/test_mcp_upstream_auth_e2e.py
+++ b/tests/e2e/mcp/test_mcp_upstream_auth_e2e.py
@ -0,0 +1,147 @@
+"""Live e2e: the gateway attaches the server's stored upstream credentials.
+
+Covers mcp.call_tool.api_key.injects_upstream_credential (a static shared-key
+credential is injected as X-API-Key) and mcp.list_tools.oauth.succeeds +
+mcp.call_tool.oauth.succeeds (OAuth2 client_credentials: the gateway exchanges
+the stored client id/secret at the token endpoint and sends the minted access
+token upstream), against the guarded mcp-stub mounts (tests/e2e/mcp/stub/).
+
+Both tests follow the suite lifecycle: register the server with credentials
+over the management API and defer its deletion, assert the recorded state
+round-trips with the secret redacted (GET /v1/mcp/server/{id} echoes the auth
+config but nulls `credentials`), then drive initialize + tools/list +
+tools/call through the gateway and assert the enforced behavior. The guarded
+stub mounts 401 any request that does not carry exactly the expected
+credential, so a served call is itself proof of injection; that is the
+fail-before-fix evidence built into the design, since a gateway that stops
+attaching the credential (or attaches the wrong one) cannot list a single
+tool here. The `recorded_headers` read-back then makes the assertion explicit
+and adds the boundary check: the caller's virtual key must appear in no
+header the upstream received (the LIT-3794 class of bug, where the proxy
+forwards its caller's own credential upstream).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from e2e_config import (
+    MCP_STUB_APIKEY_URL,
+    MCP_STUB_OAUTH_ACCESS_TOKEN,
+    MCP_STUB_OAUTH_CLIENT_ID,
+    MCP_STUB_OAUTH_CLIENT_SECRET,
+    MCP_STUB_OAUTH_SCOPE,
+    MCP_STUB_OAUTH_URL,
+    MCP_STUB_TOKEN_URL,
+    MCP_STUB_UPSTREAM_API_KEY,
+    unique_marker,
+)
+from lifecycle import ResourceManager
+from mcp_client import McpClient
+from models import KeyGenerateBody, McpServerCreateBody, McpServerCredentials
+
+pytestmark = pytest.mark.e2e
+
+GUARDED_STUB_TOOLS = ("echo", "recorded_headers")
+
+
+class TestMcpUpstreamSharedKeyInjection:
+    """A server stored with `auth_type: api_key` reaches its guarded upstream:
+    the gateway injects the stored secret as X-API-Key on every egress request
+    and the secret never travels back out of the management API."""
+
+    @pytest.mark.covers("mcp.call_tool.api_key.injects_upstream_credential")
+    def test_stored_api_key_reaches_upstream_and_never_leaks(
+        self, client: McpClient, resources: ResourceManager
+    ) -> None:
+        alias = f"e2emcpkeyauth{unique_marker()}"
+        created = client.create_server(
+            McpServerCreateBody(
+                alias=alias,
+                url=MCP_STUB_APIKEY_URL,
+                allow_all_keys=True,
+                auth_type="api_key",
+                credentials=McpServerCredentials(auth_value=MCP_STUB_UPSTREAM_API_KEY),
+            )
+        )
+        resources.defer(lambda: client.delete_server(created.server_id))
+
+        stored = client.server_info(created.server_id)
+        assert stored.auth_type == "api_key"
+        assert stored.url == MCP_STUB_APIKEY_URL
+        assert stored.credentials is None, f"stored secret must be redacted on read-back, got {stored.credentials}"
+
+        key = client.gateway.generate_key(KeyGenerateBody())
+        resources.defer(lambda: client.gateway.delete_key(key))
+
+        headers = {"x-litellm-api-key": f"Bearer {key}"}
+        names = client.poll_tool_names(alias, headers)
+        expected = tuple(sorted(f"{alias}-{tool}" for tool in GUARDED_STUB_TOOLS))
+        assert names == expected, f"guarded upstream listed {names}, expected exactly {expected}"
+
+        payload = f"e2e-{unique_marker()}"
+        result = client.call_tool(alias, headers, f"{alias}-echo", {"text": payload})
+        assert result.is_error is False, f"echo through the guarded upstream errored: {result.text[:300]}"
+        assert result.text == payload
+
+        upstream_headers = client.stub_recorded_headers(alias, headers, f"{alias}-recorded_headers")
+        assert upstream_headers.get("x-api-key") == MCP_STUB_UPSTREAM_API_KEY
+        leaked = sorted(name for name, value in upstream_headers.items() if key in value)
+        assert leaked == [], f"caller's virtual key crossed the gateway boundary in header(s) {leaked}"
+
+
+class TestMcpOauth2ClientCredentials:
+    """A server stored with `auth_type: oauth2` in the client_credentials (M2M)
+    flow works end to end: the gateway exchanges the stored client id/secret
+    (with the configured scope) at the token endpoint and presents the minted
+    access token upstream; the caller's own credential stays at the gateway."""
+
+    @pytest.mark.covers("mcp.list_tools.oauth.succeeds")
+    @pytest.mark.covers("mcp.call_tool.oauth.succeeds")
+    def test_gateway_exchanges_client_credentials_and_sends_minted_token(
+        self, client: McpClient, resources: ResourceManager
+    ) -> None:
+        alias = f"e2emcpoauth{unique_marker()}"
+        created = client.create_server(
+            McpServerCreateBody(
+                alias=alias,
+                url=MCP_STUB_OAUTH_URL,
+                allow_all_keys=True,
+                auth_type="oauth2",
+                oauth2_flow="client_credentials",
+                token_url=MCP_STUB_TOKEN_URL,
+                credentials=McpServerCredentials(
+                    client_id=MCP_STUB_OAUTH_CLIENT_ID,
+                    client_secret=MCP_STUB_OAUTH_CLIENT_SECRET,
+                    scopes=[MCP_STUB_OAUTH_SCOPE],
+                ),
+            )
+        )
+        resources.defer(lambda: client.delete_server(created.server_id))
+
+        stored = client.server_info(created.server_id)
+        assert stored.auth_type == "oauth2"
+        assert stored.oauth2_flow == "client_credentials"
+        assert stored.token_url == MCP_STUB_TOKEN_URL
+        assert stored.credentials is None, f"client secret must be redacted on read-back, got {stored.credentials}"
+
+        key = client.gateway.generate_key(KeyGenerateBody())
+        resources.defer(lambda: client.gateway.delete_key(key))
+
+        headers = {"x-litellm-api-key": f"Bearer {key}"}
+        names = client.poll_tool_names(alias, headers)
+        expected = tuple(sorted(f"{alias}-{tool}" for tool in GUARDED_STUB_TOOLS))
+        assert names == expected, f"oauth upstream listed {names}, expected exactly {expected}"
+
+        payload = f"e2e-{unique_marker()}"
+        result = client.call_tool(alias, headers, f"{alias}-echo", {"text": payload})
+        assert result.is_error is False, f"echo through the oauth upstream errored: {result.text[:300]}"
+        assert result.text == payload
+
+        upstream_headers = client.stub_recorded_headers(alias, headers, f"{alias}-recorded_headers")
+        assert upstream_headers.get("authorization") == f"Bearer {MCP_STUB_OAUTH_ACCESS_TOKEN}", (
+            "upstream must receive exactly the token the stub IdP mints for the stored client credentials, "
+            f"got {upstream_headers.get('authorization')!r}"
+        )
+        leaked = sorted(name for name, value in upstream_headers.items() if key in value)
+        assert leaked == [], f"caller's virtual key crossed the gateway boundary in header(s) {leaked}"
--- a/tests/e2e/models.py
+++ b/tests/e2e/models.py
@ -120,7 +120,8 @@ class McpServerCredentials(BaseModel):

 class McpServerCreateBody(BaseModel):
    """POST /v1/mcp/server. `allow_all_keys` opts the server out of per-key
-    object_permission grants so any virtual key on the proxy may use it."""
+    object_permission grants so any virtual key on the proxy may use it.
+    `allowed_tools` restricts the server to a subset of its upstream tools."""

    alias: str
    url: str
@ -132,6 +133,7 @@ class McpServerCreateBody(BaseModel):
    authorization_url: str | None = None
    token_url: str | None = None
    oauth2_flow: Literal["client_credentials", "authorization_code"] | None = None
+    allowed_tools: list[str] | None = None


 class McpServerInfo(BaseModel):
@ -148,6 +150,7 @@ class McpServerInfo(BaseModel):
    authorization_url: str | None = None
    token_url: str | None = None
    oauth2_flow: str | None = None
+    allowed_tools: list[str] = []


 # ---------- customers ----------