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

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)

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

tests/e2e/e2e_config.py

@@ -42,10 +42,15 @@
 # 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 @@
 )
 
 # 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"

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 @@ def call_tool(self, alias: str, headers: dict[str, str], tool: str, arguments: T
         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, ...]:

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,22 +10,32 @@
   `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.
 - `/oauth/authorize` — the auto-approving authorization endpoint: validates
   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.
-
-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
+- `/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 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.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 recorded_headers() -> str:
     _ = 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 @@ async def lifespan(_: Starlette) -> AsyncGenerator[None]:
                     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,