test(e2e): run the official MCP conformance suite through the gateway and cover prompt/resource primitives

tests/e2e/coverage_registry/mcp.yaml

@@ -151,3 +151,11 @@
   assertions: [uses_per_user_token]
   source: "v2 authorization_code egress arm (per-user DB token)"
   rationale: Calls on an interactive server must present the user's upstream token, never the caller's virtual key or IdP JWT
+- id: mcp.protocol.api_key.passes_official_conformance
+  module: mcp
+  tier: P0
+  operation: protocol
+  auth_family: api_key
+  assertions: [passes_official_conformance]
+  source: "@modelcontextprotocol/conformance server scenarios via tests/e2e/mcp/test_mcp_conformance_e2e.py"
+  rationale: The gateway re-serves the MCP protocol to hosts, so it must stay spec-conformant as a server; the official suite catches lifecycle/error-shape regressions our behavior tests do not encode

tests/e2e/e2e_config.py

@@ -46,6 +46,7 @@ MCP_STUB_URL = os.environ.get("E2E_MCP_STUB_URL", "http://mcp-stub:8765/mcp")
 # 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.
 _MCP_STUB_BASE = MCP_STUB_URL.removesuffix("/mcp")
+MCP_STUB_CONFORMANCE_URL = f"{_MCP_STUB_BASE}/conformance/mcp"
 MCP_STUB_OAUTHUSER_URL = f"{_MCP_STUB_BASE}/oauthuser/mcp"
 MCP_STUB_TOKEN_URL = f"{_MCP_STUB_BASE}/oauth/token"
 

tests/e2e/mcp/auth_forwarder.py

@@ -0,0 +1,88 @@
+"""In-process reverse proxy that stamps the LiteLLM key onto every request.
+
+The official conformance suite's client offers no way to send custom headers,
+but the gateway's MCP routes require the virtual key. This forwarder plays the
+role of an MCP host's HTTP layer configured with the key header: it listens on
+a loopback port, injects `x-litellm-api-key: Bearer <key>`, and relays
+everything (including SSE streams, unbuffered) to the proxy under test.
+
+Runs inside the pytest process on a background uvicorn thread; tests get the
+local base URL from `serve()` and hand `{base}/{alias}/mcp` to the conformance
+CLI as the server under test.
+"""
+
+from __future__ import annotations
+
+import socket
+import threading
+import time
+from collections.abc import Generator
+from contextlib import contextmanager
+from typing import cast
+
+import httpx
+import uvicorn
+from starlette.applications import Starlette
+from starlette.background import BackgroundTask
+from starlette.requests import Request
+from starlette.responses import StreamingResponse
+from starlette.routing import Route
+
+from e2e_config import PROXY_BASE_URL, REQUEST_TIMEOUT
+
+_HOP_BY_HOP_REQUEST_HEADERS = frozenset({"host", "content-length"})
+_HOP_BY_HOP_RESPONSE_HEADERS = frozenset({"content-length", "transfer-encoding", "content-encoding"})
+
+
+def _build_app(upstream: httpx.AsyncClient, litellm_key: str) -> Starlette:
+    async def forward(request: Request) -> StreamingResponse:
+        url = httpx.URL(path=request.url.path, query=request.url.query.encode())
+        headers = {
+            name: value
+            for name, value in request.headers.items()
+            if name.lower() not in _HOP_BY_HOP_REQUEST_HEADERS
+        }
+        headers["x-litellm-api-key"] = f"Bearer {litellm_key}"
+        proxied = upstream.build_request(request.method, url, headers=headers, content=await request.body())
+        response = await upstream.send(proxied, stream=True)
+        response_headers = {
+            name: value
+            for name, value in response.headers.items()
+            if name.lower() not in _HOP_BY_HOP_RESPONSE_HEADERS
+        }
+        return StreamingResponse(
+            response.aiter_raw(),
+            status_code=response.status_code,
+            headers=response_headers,
+            background=BackgroundTask(response.aclose),
+        )
+
+    methods = ["GET", "POST", "DELETE", "PUT", "PATCH", "HEAD", "OPTIONS"]
+    return Starlette(routes=[Route("/{path:path}", forward, methods=methods)])
+
+
+def _free_loopback_port() -> int:
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return cast("int", sock.getsockname()[1])
+
+
+@contextmanager
+def serve(litellm_key: str) -> Generator[str]:
+    """Serve the forwarder for the block's duration; yields its base URL."""
+    upstream = httpx.AsyncClient(base_url=PROXY_BASE_URL, timeout=httpx.Timeout(REQUEST_TIMEOUT))
+    port = _free_loopback_port()
+    server = uvicorn.Server(
+        uvicorn.Config(_build_app(upstream, litellm_key), host="127.0.0.1", port=port, log_level="warning")
+    )
+    thread = threading.Thread(target=server.run, daemon=True)
+    thread.start()
+    deadline = time.monotonic() + 15
+    while not server.started:
+        assert time.monotonic() < deadline, "auth forwarder never started"
+        time.sleep(0.05)
+    try:
+        yield f"http://127.0.0.1:{port}"
+    finally:
+        server.should_exit = True
+        thread.join(timeout=10)

tests/e2e/mcp/mcp_client.py

@@ -121,6 +121,60 @@ async def _call_tool(url: str, headers: dict[str, str], tool: str, arguments: To
                 return McpToolText(text=_first_text(result), is_error=bool(result.isError))
 
 
+async def _list_prompts(url: str, headers: dict[str, str]) -> tuple[tuple[str, tuple[str, ...]], ...]:
+    """(name, argument names) per prompt, sorted by name."""
+    async with _http_client(headers) as http_client:
+        async with streamable_http_client(url, http_client=http_client) as (read, write, _):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+                listed = await session.list_prompts()
+                return tuple(
+                    sorted(
+                        (prompt.name, tuple(arg.name for arg in (prompt.arguments or [])))
+                        for prompt in listed.prompts
+                    )
+                )
+
+
+async def _get_prompt(url: str, headers: dict[str, str], name: str, arguments: dict[str, str]) -> str:
+    """The text of the rendered prompt's first message."""
+    async with _http_client(headers) as http_client:
+        async with streamable_http_client(url, http_client=http_client) as (read, write, _):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+                result = await session.get_prompt(name, arguments)
+                first = result.messages[0].content if result.messages else None
+                if isinstance(first, TextContent):
+                    return first.text
+                return f"<non-text prompt content: {type(first).__name__}>"
+
+
+async def _list_resources(url: str, headers: dict[str, str]) -> tuple[tuple[str, str], ...]:
+    """(uri, name) per resource, sorted by uri."""
+    async with _http_client(headers) as http_client:
+        async with streamable_http_client(url, http_client=http_client) as (read, write, _):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+                listed = await session.list_resources()
+                return tuple(sorted((str(resource.uri), resource.name) for resource in listed.resources))
+
+
+async def _read_resource(url: str, headers: dict[str, str], uri: str) -> str:
+    """The text of the resource's first content block."""
+    from pydantic import AnyUrl
+
+    async with _http_client(headers) as http_client:
+        async with streamable_http_client(url, http_client=http_client) as (read, write, _):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+                result = await session.read_resource(AnyUrl(uri))
+                first = result.contents[0] if result.contents else None
+                text = getattr(first, "text", None)
+                if isinstance(text, str):
+                    return text
+                return f"<non-text resource content: {type(first).__name__}>"
+
+
 # ---------- interactive (authorization_code) OAuth: the MCP-host side ----------
 
 # Where the "browser" lands at the end of the authorize dance. Nothing listens
@@ -339,6 +393,22 @@ class McpClient:
         prior poll_oauth_tool_names dance), over its own fresh MCP session."""
         return asyncio.run(_oauth_call_tool(_mcp_url(alias), headers, storage, tool, arguments))
 
+    def list_prompts(self, alias: str, headers: dict[str, str]) -> tuple[tuple[str, tuple[str, ...]], ...]:
+        """prompts/list over its own fresh MCP session: (name, argument names) sorted by name."""
+        return asyncio.run(_list_prompts(_mcp_url(alias), headers))
+
+    def get_prompt(self, alias: str, headers: dict[str, str], name: str, arguments: dict[str, str]) -> str:
+        """prompts/get over its own fresh MCP session: the rendered first message's text."""
+        return asyncio.run(_get_prompt(_mcp_url(alias), headers, name, arguments))
+
+    def list_resources(self, alias: str, headers: dict[str, str]) -> tuple[tuple[str, str], ...]:
+        """resources/list over its own fresh MCP session: (uri, name) sorted by uri."""
+        return asyncio.run(_list_resources(_mcp_url(alias), headers))
+
+    def read_resource(self, alias: str, headers: dict[str, str], uri: str) -> str:
+        """resources/read over its own fresh MCP session: the first content block's text."""
+        return asyncio.run(_read_resource(_mcp_url(alias), headers, uri))
+
     def stub_stats(self, alias: str, headers: dict[str, str], stats_tool: str, marker: str) -> StubToolStats:
         outcome = self.call_tool(alias, headers, stats_tool, {"marker": marker})
         return StubToolStats.model_validate_json(outcome.text)

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, three 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,6 +10,13 @@ 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.
+- `/conformance/mcp` — anonymous, serving the official
+  @modelcontextprotocol/conformance suite's hardcoded fixture contract
+  (test_simple_text / test_error_handling tools, test://static-text,
+  test://static-binary and the test://template/{id}/data resources,
+  test_simple_prompt / test_prompt_with_arguments prompts), so the gateway can
+  be conformance-tested end to end and the suite's own prompt/resource tests
+  have a deterministic upstream that serves all three MCP primitives.
 - `/oauthuser/mcp` — the interactive (authorization_code) upstream: rejects
   anything but `Bearer OAUTH_USER_ACCESS_TOKEN`, which only the
   authorization_code grant hands out, so a served request proves the whole
@@ -61,6 +68,7 @@ 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)
+conformance_mcp = FastMCP("e2e-stub-conformance", 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 +117,50 @@ def stats(marker: str) -> str:
     )
 
 
+@conformance_mcp.tool()
+def test_simple_text() -> str:
+    """Return a plain text result (the conformance suite's simple-text fixture)."""
+    return "Hello from the e2e conformance stub"
+
+
+@conformance_mcp.tool()
+def test_error_handling(should_error: bool = True) -> str:
+    """Raise so the result carries isError (the conformance suite's error fixture)."""
+    if should_error:
+        raise ValueError("Intentional error from the e2e conformance stub")
+    return "no error"
+
+
+@conformance_mcp.resource("test://static-text")
+def static_text_resource() -> str:
+    """The conformance suite's static text resource fixture."""
+    return "Static text resource from the e2e conformance stub"
+
+
+@conformance_mcp.resource("test://static-binary")
+def static_binary_resource() -> bytes:
+    """The conformance suite's static binary resource fixture."""
+    return b"\x89binary-fixture-bytes\x00\x01"
+
+
+@conformance_mcp.resource("test://template/{id}/data")
+def template_resource(id: str) -> str:
+    """The conformance suite's resource template fixture: {id} must substitute."""
+    return f"Template resource data for id={id}"
+
+
+@conformance_mcp.prompt()
+def test_simple_prompt() -> str:
+    """The conformance suite's no-argument prompt fixture."""
+    return "This is a simple prompt from the e2e conformance stub"
+
+
+@conformance_mcp.prompt()
+def test_prompt_with_arguments(arg1: str, arg2: str) -> str:
+    """The conformance suite's argument-substitution prompt fixture."""
+    return f"Prompt rendered with arg1={arg1} and arg2={arg2}"
+
+
 def _register_guarded_tools(server: FastMCP, mount: str) -> None:
     def echo(text: str) -> str:
         """Return `text` unchanged."""
@@ -224,7 +276,7 @@ async def oauth_token(request: Request) -> JSONResponse:
 
 
 def build_app() -> Starlette:
-    servers = (main_mcp, oauthuser_mcp)
+    servers = (main_mcp, conformance_mcp, oauthuser_mcp)
     apps = {server.name: server.streamable_http_app() for server in servers}
 
     @contextlib.asynccontextmanager
@@ -247,6 +299,7 @@ def build_app() -> Starlette:
                     expected=f"Bearer {OAUTH_USER_ACCESS_TOKEN}",
                 ),
             ),
+            Mount("/conformance", app=apps["e2e-stub-conformance"]),
             Mount("/", app=apps["e2e-stub"]),
         ],
         lifespan=lifespan,

tests/e2e/mcp/test_mcp_conformance_e2e.py

@@ -0,0 +1,121 @@
+"""Live e2e: the official MCP conformance suite, run through the gateway.
+
+Covers mcp.protocol.api_key.passes_official_conformance: the gateway is a
+protocol middlebox, so beyond this suite's behavior tests it must stay
+conformant to the MCP spec as re-served to hosts. This test drives the
+official @modelcontextprotocol/conformance server scenarios (the same suite
+the SDKs gate their CI on) against a gateway-registered server backed by the
+stub's /conformance mount, which implements the suite's hardcoded fixture
+contract (test_simple_text, test://static-text, test_simple_prompt, ...).
+
+The conformance CLI cannot send custom headers, so the run goes through
+auth_forwarder.serve(), an in-process reverse proxy that stamps the virtual
+key onto every request the way a configured MCP host's HTTP layer would; the
+gateway path itself stays fully authenticated and unmodified.
+
+EXPECTED_GAPS is the checked-in baseline, exact in both directions: a
+scenario failing that is not listed fails this test (a regression), and a
+listed scenario that starts passing also fails it (so the baseline ratchets
+down instead of going stale). The current entries are real gateway findings,
+verified by hand: logging/setLevel and completion/complete are not relayed
+(-32601), prompts/get resolves alias-prefixed names only (the unprefixed
+fallback that tools/call has is missing, so the suite's hardcoded prompt
+names 403), and binary resource read-through drops the base64 blob field
+(the stub serves it; the gateway's answer has no blob).
+
+Scenarios needing client capabilities the gateway does not advertise for its
+callers (sampling, elicitation) and SSE-timing scenarios are intentionally
+not run; the curated list below is the subset a gateway can honestly own.
+"""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+
+import auth_forwarder
+from e2e_config import MCP_STUB_CONFORMANCE_URL, unique_marker
+from mcp_client import McpClient
+from models import KeyGenerateBody, McpServerCreateBody
+
+pytestmark = pytest.mark.e2e
+
+CONFORMANCE_PACKAGE = "@modelcontextprotocol/conformance@0.1.11"
+
+SCENARIOS = (
+    "server-initialize",
+    "ping",
+    "logging-set-level",
+    "completion-complete",
+    "tools-list",
+    "tools-call-simple-text",
+    "tools-call-error",
+    "resources-list",
+    "resources-read-text",
+    "resources-read-binary",
+    "resources-templates-read",
+    "prompts-list",
+    "prompts-get-simple",
+    "prompts-get-with-args",
+)
+
+EXPECTED_GAPS = {
+    "logging-set-level": "gateway answers logging/setLevel with -32601 Method not found instead of relaying it",
+    "completion-complete": "gateway answers completion/complete with -32601 Method not found instead of relaying it",
+    "prompts-get-simple": "prompts/get lacks the unprefixed-name fallback tools/call has; unprefixed names 403",
+    "prompts-get-with-args": "prompts/get lacks the unprefixed-name fallback tools/call has; unprefixed names 403",
+    "resources-read-binary": "gateway drops the base64 blob field on binary resource read-through",
+}
+
+
+@pytest.fixture(scope="module")
+def conformance_target(client: McpClient) -> Iterator[str]:
+    """A conformance-fixture server registered on the gateway, fronted by the
+    key-stamping forwarder; yields the URL the conformance CLI tests. Module
+    scoped so all scenarios share one server, one key, and one forwarder."""
+    if shutil.which("npx") is None:
+        pytest.skip("npx not available; the official conformance suite runs on Node")
+    alias = f"e2emcpconf{unique_marker()}"
+    created = client.create_server(
+        McpServerCreateBody(alias=alias, url=MCP_STUB_CONFORMANCE_URL, allow_all_keys=True)
+    )
+    key = client.gateway.generate_key(KeyGenerateBody())
+    try:
+        _ = client.poll_tool_names(alias, {"x-litellm-api-key": f"Bearer {key}"})
+        with auth_forwarder.serve(key) as forwarder_base:
+            yield f"{forwarder_base}/{alias}/mcp"
+    finally:
+        client.delete_server(created.server_id)
+        client.gateway.delete_key(key)
+
+
+class TestMcpOfficialConformance:
+    """Each curated official scenario passes through the gateway, except the
+    exact set of known gaps pinned in EXPECTED_GAPS."""
+
+    @pytest.mark.covers("mcp.protocol.api_key.passes_official_conformance")
+    @pytest.mark.parametrize("scenario", SCENARIOS)
+    def test_official_scenario(self, scenario: str, conformance_target: str, tmp_path: Path) -> None:
+        run = subprocess.run(
+            ["npx", "-y", CONFORMANCE_PACKAGE, "server", "--url", conformance_target, "--scenario", scenario],
+            capture_output=True,
+            text=True,
+            timeout=300,
+            cwd=tmp_path,
+        )
+        failed = run.returncode != 0
+        expected_reason = EXPECTED_GAPS.get(scenario)
+        if expected_reason is not None:
+            assert failed, (
+                f"conformance scenario {scenario!r} now PASSES; the gateway gap "
+                f"({expected_reason}) appears fixed, so remove it from EXPECTED_GAPS to ratchet the baseline"
+            )
+            return
+        assert not failed, (
+            f"conformance scenario {scenario!r} failed through the gateway (exit {run.returncode}); "
+            f"output tail:\n{run.stdout[-1500:]}"
+        )

tests/e2e/mcp/test_mcp_prompts_resources_e2e.py

@@ -0,0 +1,99 @@
+"""Live e2e: the MCP gateway's prompt and resource primitives.
+
+Covers mcp.list_prompts.api_key.succeeds, mcp.get_prompt.api_key.succeeds,
+mcp.list_resources.api_key.succeeds, and mcp.read_resource.api_key.succeeds,
+against the stub's /conformance mount (tests/e2e/mcp/stub/), the one upstream
+that serves all three MCP primitives.
+
+The gateway's namespacing contract differs per primitive, and the assertions
+pin it exactly as observed live: prompt names are alias-prefixed like tool
+names (and must be fetched by the prefixed name), while resource URIs pass
+through unprefixed (only the resource's display name gains the prefix). The
+prompt-rendering assertion feeds unique per-run argument values through
+prompts/get and requires them back verbatim in the rendered text, so a cached
+or canned response cannot pass; the resource assertion requires the exact
+fixture body.
+
+The binary-resource read (test://static-binary) is deliberately NOT asserted
+here: the gateway currently drops the base64 blob field on read-through
+(verified against the stub directly, which serves it), and that gap is pinned
+in the conformance suite baseline (test_mcp_conformance_e2e.py) instead.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from e2e_config import MCP_STUB_CONFORMANCE_URL, unique_marker
+from lifecycle import ResourceManager
+from mcp_client import McpClient
+from models import KeyGenerateBody, McpServerCreateBody
+
+pytestmark = pytest.mark.e2e
+
+
+class TestMcpPromptsAndResources:
+    """A registered server's prompts and resources are listed, fetched, and
+    read through the gateway with the same lifecycle contract as tools."""
+
+    @pytest.mark.covers("mcp.list_prompts.api_key.succeeds")
+    @pytest.mark.covers("mcp.get_prompt.api_key.succeeds")
+    def test_prompts_list_and_render_with_arguments(
+        self, client: McpClient, resources: ResourceManager
+    ) -> None:
+        alias = f"e2emcpprompts{unique_marker()}"
+        created = client.create_server(
+            McpServerCreateBody(alias=alias, url=MCP_STUB_CONFORMANCE_URL, allow_all_keys=True)
+        )
+        resources.defer(lambda: client.delete_server(created.server_id))
+
+        stored = client.server_info(created.server_id)
+        assert stored.alias == alias
+        assert stored.url == MCP_STUB_CONFORMANCE_URL
+
+        key = client.gateway.generate_key(KeyGenerateBody())
+        resources.defer(lambda: client.gateway.delete_key(key))
+        headers = {"x-litellm-api-key": f"Bearer {key}"}
+        _ = client.poll_tool_names(alias, headers)
+
+        listed = client.list_prompts(alias, headers)
+        expected = (
+            (f"{alias}-test_prompt_with_arguments", ("arg1", "arg2")),
+            (f"{alias}-test_simple_prompt", ()),
+        )
+        assert listed == expected, f"gateway listed prompts {listed}, expected exactly {expected}"
+
+        first_value = f"e2e-{unique_marker()}"
+        second_value = f"e2e-{unique_marker()}"
+        rendered = client.get_prompt(
+            alias, headers, f"{alias}-test_prompt_with_arguments", {"arg1": first_value, "arg2": second_value}
+        )
+        assert rendered == f"Prompt rendered with arg1={first_value} and arg2={second_value}"
+
+    @pytest.mark.covers("mcp.list_resources.api_key.succeeds")
+    @pytest.mark.covers("mcp.read_resource.api_key.succeeds")
+    def test_resources_list_and_read_text(self, client: McpClient, resources: ResourceManager) -> None:
+        alias = f"e2emcpresources{unique_marker()}"
+        created = client.create_server(
+            McpServerCreateBody(alias=alias, url=MCP_STUB_CONFORMANCE_URL, allow_all_keys=True)
+        )
+        resources.defer(lambda: client.delete_server(created.server_id))
+
+        stored = client.server_info(created.server_id)
+        assert stored.alias == alias
+        assert stored.url == MCP_STUB_CONFORMANCE_URL
+
+        key = client.gateway.generate_key(KeyGenerateBody())
+        resources.defer(lambda: client.gateway.delete_key(key))
+        headers = {"x-litellm-api-key": f"Bearer {key}"}
+        _ = client.poll_tool_names(alias, headers)
+
+        listed = client.list_resources(alias, headers)
+        expected = (
+            ("test://static-binary", f"{alias}-static_binary_resource"),
+            ("test://static-text", f"{alias}-static_text_resource"),
+        )
+        assert listed == expected, f"gateway listed resources {listed}, expected exactly {expected}"
+
+        body = client.read_resource(alias, headers, "test://static-text")
+        assert body == "Static text resource from the e2e conformance stub"