test(e2e): mcp suite for key-without-access denial (#33752)

Add an e2e suite at tests/e2e/mcp/ that proves MCP authorization over the
api_key auth family. An admin registers an upstream MCP server through the
management API (POST /v1/mcp/server, persisted in the DB and picked up without
a restart) and queues its deletion. Two keys are created against that one
server: one granted access through object_permission.mcp_servers and one with
no MCP grant. The permitted key is a live control proving the upstream is
reachable and the tool is callable, so a denial on the ungranted key is an
authorization decision rather than a dead server. The denied key then sees
none of the server's tools on tools/list and is refused a tools/call with a
403 access_denied.

A deterministic self-hosted FastMCP upstream (add/multiply over
streamable-http) is added to the e2e compose stack so the suite runs offline
with a known tool set. KeyGenerateBody gains an optional typed
object_permission so the shared gateway can create a key with an MCP grant.

tests/e2e/CLAUDE.md

@@ -13,6 +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 server surface over api_key auth: an admin registers an upstream MCP server through the management API and grants keys access via `object_permission.mcp_servers`, then the suite asserts tool listing and calling honor that permission (a key without the grant sees none of the server's tools and is refused a `tools/call` with a 403)
 - `logging/` - logging-integration delivery (datadog and friends)
 - `security/` - secret handling and log-leak protection
 - `router/` - routing and reliability behavior (fallbacks, cooldowns)

tests/e2e/docker-compose.yml

@@ -1,5 +1,7 @@
 # local setup to run e2e tests
 configs:
+  mcp_upstream_server:
+    file: ../mcp_tests/mcp_e2e_upstream_server.py
   litellm_config:
     content: |
       general_settings:
@@ -131,7 +133,27 @@ services:
         target: /app/config.yaml
     command: ["--config", "/app/config.yaml", "--port", "4000"]
 
-# throwaway db 
+# deterministic self-hosted upstream MCP server (FastMCP add/multiply over
+# streamable-http), reachable by the litellm container at mcp-upstream:8090/mcp.
+# Not a depends_on of litellm on purpose: only the mcp suite needs it, and it
+# boots long before the proxy is live, so it must not gate the other suites'
+# stack. The suite registers it through /v1/mcp/server at test time.
+  mcp-upstream:
+    image: ghcr.io/berriai/litellm:main-latest
+    entrypoint: ["python3", "/app/mcp_upstream_server.py"]
+    environment:
+      MCP_HOST: 0.0.0.0
+      MCP_PORT: "8090"
+    configs:
+      - source: mcp_upstream_server
+        target: /app/mcp_upstream_server.py
+    healthcheck:
+      test: ["CMD", "python3", "-c", "import socket; socket.create_connection(('127.0.0.1', 8090), 2).close()"]
+      interval: 3s
+      timeout: 3s
+      retries: 40
+
+# throwaway db
   db:
     image: postgres:16
     environment:

tests/e2e/mcp/conftest.py

@@ -0,0 +1,16 @@
+"""MCP suite's `client` fixture.
+
+The shared lifecycle (resources/scoped_key), proxy liveness handling, and the
+`e2e`/`covers` markers live in the parent tests/e2e/conftest.py. McpClient holds
+the shared Gateway, so the `resources` fixture tears down whatever this suite
+creates (keys via the Gateway, MCP servers via the deferred cleanups).
+"""
+
+import pytest
+
+from mcp_client import McpClient, build_client
+
+
+@pytest.fixture(scope="session")
+def client() -> McpClient:
+    return build_client()

tests/e2e/mcp/mcp_client.py

@@ -0,0 +1,153 @@
+"""Client for the MCP e2e suite: admin server registration plus the api_key tool
+surface.
+
+An admin registers an upstream MCP server through the management API
+(`/v1/mcp/server`, persisted in the DB) and grants a virtual key access to it via
+`object_permission.mcp_servers`. Keys then reach the server through the REST bridge
+the proxy exposes for api_key auth (`/mcp-rest/tools/list`, `/mcp-rest/tools/call`),
+which `user_api_key_auth` gates the same way the JSON-RPC `/mcp` surface does. The
+request/response bodies are co-located here because only this suite speaks MCP.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from pydantic import BaseModel, ConfigDict, Field, RootModel
+
+from e2e_gateway import Gateway, build_gateway
+from e2e_http import Headers, NoBody, Result, unwrap
+from models import KeyGenerateBody, ObjectPermission
+
+
+class ApiKeyHeaders(Headers):
+    x_litellm_api_key: str = Field(serialization_alias="x-litellm-api-key")
+
+
+class McpServerNewBody(BaseModel):
+    server_name: str
+    alias: str
+    url: str
+    transport: str = "http"
+
+
+class McpServerNewResponse(BaseModel):
+    server_id: str
+
+
+class McpServerRow(BaseModel):
+    server_id: str
+    alias: str | None = None
+    url: str | None = None
+
+
+class McpServersListResponse(RootModel[list[McpServerRow]]):
+    pass
+
+
+class McpToolMcpInfo(BaseModel):
+    server_id: str | None = None
+    alias: str | None = None
+
+
+class McpToolEntry(BaseModel):
+    name: str
+    description: str | None = None
+    mcp_info: McpToolMcpInfo | None = None
+
+
+class McpToolsListResponse(BaseModel):
+    tools: list[McpToolEntry] = []
+    error: str | None = None
+    message: str | None = None
+
+    def tool_names_for_server(self, server_id: str) -> frozenset[str]:
+        return frozenset(
+            tool.name
+            for tool in self.tools
+            if tool.mcp_info is not None and tool.mcp_info.server_id == server_id
+        )
+
+
+class McpCallToolBody(BaseModel):
+    name: str
+    arguments: dict[str, int]
+    server_id: str
+
+
+class McpCallContent(BaseModel):
+    type: str | None = None
+    text: str | None = None
+
+
+class McpCallToolResponse(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    content: list[McpCallContent] = []
+    is_error: bool | None = Field(default=None, alias="isError")
+
+    @property
+    def first_text(self) -> str | None:
+        return self.content[0].text if self.content else None
+
+
+@dataclass(frozen=True, slots=True)
+class McpClient:
+    gateway: Gateway
+
+    def register_server(self, *, server_name: str, alias: str, url: str) -> str:
+        return unwrap(
+            self.gateway.transport.post(
+                "/v1/mcp/server",
+                headers=self.gateway.transport.master,
+                json=McpServerNewBody(server_name=server_name, alias=alias, url=url),
+                response_type=McpServerNewResponse,
+            )
+        ).server_id
+
+    def delete_server(self, server_id: str) -> None:
+        _ = self.gateway.transport.delete(
+            f"/v1/mcp/server/{server_id}",
+            headers=self.gateway.transport.master,
+            json=NoBody(),
+            response_type=NoBody,
+        )
+
+    def registered_servers(self) -> list[McpServerRow]:
+        return unwrap(
+            self.gateway.transport.get(
+                "/v1/mcp/server",
+                headers=self.gateway.transport.master,
+                params=NoBody(),
+                response_type=McpServersListResponse,
+            )
+        ).root
+
+    def generate_key(self, *, user_id: str, mcp_servers: list[str] | None) -> str:
+        object_permission = (
+            ObjectPermission(mcp_servers=mcp_servers) if mcp_servers is not None else None
+        )
+        return self.gateway.generate_key(
+            KeyGenerateBody(models=[], user_id=user_id, object_permission=object_permission)
+        )
+
+    def list_tools(self, key: str) -> Result[McpToolsListResponse]:
+        return self.gateway.transport.get(
+            "/mcp-rest/tools/list",
+            headers=ApiKeyHeaders(x_litellm_api_key=key),
+            params=NoBody(),
+            response_type=McpToolsListResponse,
+        )
+
+    def call_tool(
+        self, key: str, *, server_id: str, name: str, arguments: dict[str, int]
+    ) -> Result[McpCallToolResponse]:
+        return self.gateway.transport.post(
+            "/mcp-rest/tools/call",
+            headers=ApiKeyHeaders(x_litellm_api_key=key),
+            json=McpCallToolBody(name=name, arguments=arguments, server_id=server_id),
+            response_type=McpCallToolResponse,
+        )
+
+
+def build_client() -> McpClient:
+    return McpClient(gateway=build_gateway())

tests/e2e/mcp/test_mcp_key_access_e2e.py

@@ -0,0 +1,103 @@
+"""Live e2e: a virtual key without MCP access is denied an MCP server's tools.
+
+An admin registers an upstream MCP server through the management API (persisted in
+the DB, picked up without a restart) and queues its deletion. Two keys are created
+against that one server: one granted access through `object_permission.mcp_servers`
+and one with no MCP grant at all. The permitted key is the control that proves the
+upstream is alive and the tool is callable, so a failure on the denied key is an
+authorization denial rather than a dead server. The denied key must then see none
+of the server's tools on `tools/list` and must be refused with a 403 on
+`tools/call`.
+
+Both the recorded state (the server is registered; the permitted key resolves its
+tools) and the enforced behavior (the unpermitted key sees nothing and is blocked)
+are asserted, so a regression that leaks tools to an ungranted key or drops the
+call-time permission check fails here.
+"""
+
+import os
+
+import pytest
+
+from e2e_config import unique_marker
+from e2e_http import UnknownApiError, unwrap
+from lifecycle import ResourceManager
+from mcp_client import McpClient
+
+pytestmark = pytest.mark.e2e
+
+MCP_UPSTREAM_URL = os.environ.get("E2E_MCP_UPSTREAM_URL", "http://mcp-upstream:8090/mcp")
+MATH_TOOLS = frozenset({"add", "multiply"})
+
+
+def _register_math_server(client: McpClient, resources: ResourceManager) -> str:
+    name = f"e2e_math_{unique_marker()}"
+    server_id = client.register_server(server_name=name, alias=name, url=MCP_UPSTREAM_URL)
+    resources.defer(lambda: client.delete_server(server_id))
+    return server_id
+
+
+def _key(client: McpClient, resources: ResourceManager, *, mcp_servers: list[str] | None) -> str:
+    label = "allowed" if mcp_servers else "denied"
+    key = client.generate_key(user_id=f"e2e-mcp-{label}-{unique_marker()}", mcp_servers=mcp_servers)
+    resources.defer(lambda: client.gateway.delete_key(key))
+    return key
+
+
+def _assert_registered(client: McpClient, server_id: str) -> None:
+    registered = {row.server_id for row in client.registered_servers()}
+    assert server_id in registered, f"registered server {server_id} absent from /v1/mcp/server: {registered}"
+
+
+class TestMcpKeyWithoutAccessIsDenied:
+    @pytest.mark.covers("mcp.list_tools.api_key.denied_without_permission")
+    def test_list_tools_denied_without_permission(
+        self, client: McpClient, resources: ResourceManager
+    ) -> None:
+        server_id = _register_math_server(client, resources)
+        _assert_registered(client, server_id)
+
+        permitted_key = _key(client, resources, mcp_servers=[server_id])
+        denied_key = _key(client, resources, mcp_servers=None)
+
+        permitted_tools = unwrap(client.list_tools(permitted_key)).tool_names_for_server(server_id)
+        assert MATH_TOOLS <= permitted_tools, (
+            f"granted key did not see the server's tools (upstream dead or grant not applied): "
+            f"{permitted_tools}"
+        )
+
+        denied_tools = unwrap(client.list_tools(denied_key)).tool_names_for_server(server_id)
+        assert denied_tools == frozenset(), (
+            f"ungranted key saw the server's tools; tools/list leaked across the permission "
+            f"boundary: {denied_tools}"
+        )
+
+    @pytest.mark.covers("mcp.call_tool.api_key.denied_without_permission")
+    def test_call_tool_denied_without_permission(
+        self, client: McpClient, resources: ResourceManager
+    ) -> None:
+        server_id = _register_math_server(client, resources)
+        _assert_registered(client, server_id)
+
+        permitted_key = _key(client, resources, mcp_servers=[server_id])
+        denied_key = _key(client, resources, mcp_servers=None)
+
+        permitted_tools = unwrap(client.list_tools(permitted_key)).tool_names_for_server(server_id)
+        assert "add" in permitted_tools, (
+            f"granted key did not discover the add tool (upstream dead or grant not applied): "
+            f"{permitted_tools}"
+        )
+
+        permitted_call = unwrap(
+            client.call_tool(permitted_key, server_id=server_id, name="add", arguments={"a": 3, "b": 4})
+        )
+        assert permitted_call.is_error is not True, f"granted key's tool call errored: {permitted_call}"
+        assert permitted_call.first_text == "7", (
+            f"granted key's add(3, 4) did not return 7 (upstream not reachable): {permitted_call}"
+        )
+
+        match client.call_tool(denied_key, server_id=server_id, name="add", arguments={"a": 3, "b": 4}):
+            case UnknownApiError(status_code=403, body=body):
+                assert "access_denied" in body, f"403 was not an MCP access denial: {body}"
+            case other:
+                pytest.fail(f"ungranted key's tool call was not refused with 403 access_denied: {other}")

tests/e2e/models.py

@@ -39,6 +39,10 @@ class KeyMetadata(BaseModel):
     logging: list[KeyLoggingCallback] | None = None
 
 
+class ObjectPermission(BaseModel):
+    mcp_servers: list[str] | None = None
+
+
 class KeyGenerateBody(BaseModel):
     models: list[str] = []
     duration: str | None = None
@@ -57,6 +61,7 @@ class KeyGenerateBody(BaseModel):
     rpm_limit: int | None = None
     allowed_routes: list[str] | None = None
     metadata: KeyMetadata | None = None
+    object_permission: ObjectPermission | None = None
 
 
 class KeyGenerateResponse(BaseModel):

tests/mcp_tests/mcp_e2e_upstream_server.py

@@ -0,0 +1,40 @@
+"""Deterministic upstream MCP server for the mcp e2e suite.
+
+A tiny FastMCP server exposing `add` and `multiply` over streamable-http so the
+suite has a self-hosted, offline upstream to register and exercise. DNS-rebinding
+protection is turned off because the litellm container reaches this over the
+compose network by service name (`mcp-upstream:8090`), not localhost, and the
+stack is an isolated throwaway. Bind host/port come from MCP_HOST/MCP_PORT.
+"""
+
+import os
+
+from mcp.server.fastmcp import FastMCP
+from mcp.server.transport_security import TransportSecuritySettings
+
+mcp: FastMCP = FastMCP(
+    "e2e-math",
+    host=os.getenv("MCP_HOST", "0.0.0.0"),
+    port=int(os.getenv("MCP_PORT", "8090")),
+    transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
+)
+
+
+@mcp.tool()
+def add(a: int, b: int) -> int:
+    """Add two integers"""
+    return a + b
+
+
+@mcp.tool()
+def multiply(a: int, b: int) -> int:
+    """Multiply two integers"""
+    return a * b
+
+
+def main() -> None:
+    mcp.run(transport="streamable-http")
+
+
+if __name__ == "__main__":
+    main()