test(mcp): enforce complete revision and transport conformance

.circleci/scripts/install_mcp_conformance.sh

@@ -18,6 +18,7 @@ PY
 mkdir -p "$root"
 tar -xzf "$archive" --strip-components=1 -C "$root"
 npm ci --ignore-scripts --prefix "$root"
+npm ci --ignore-scripts --prefix "$root/examples/servers/typescript"
 # The current reference misclassifies legacy initialize _meta as stateless traffic.
 # Use the last official legacy reference; leave its source and lockfile untouched.
 curl --fail --location --silent --show-error \

tests/integration/README.md

@@ -37,3 +37,50 @@ The mcp shard runs the MCP gateway against SDK peers owned by each test (`_suppo
 Browser contracts live in `tests/e2e/ui/tests/integrationCritical` and run only through `tests/e2e/ui/integration.config.ts`. The expected browser results are listed in `expected.json` in that directory and checked by `.circleci/scripts/verify_integration_browser.py`. The CircleCI browser shard builds the checked-out dashboard, starts the owned proxy with that build, and verifies one exact browser result without retries or skips. The default Playwright selection excludes this directory. The focused project flow asserts the submitted create and clear values, fresh SQL state and actual blocked/restored serving while preserving model restrictions
 
 Two always-on `-replica` CircleCI jobs (management, database) run their groups in replica mode, where every proxy connects through a real `litellm_writer` role and a real read-only `litellm_reader` role against the same PostgreSQL. Nothing is captured there: the job passes when the tests pass, and a write routed to the read-only reader fails the test that issued it. A deeper check runs on demand as the `routing_parity` workflow, triggered through the CircleCI API v2 pipeline endpoint on the PR branch with `{"parameters": {"routing_parity_base": "<40-hex merge-base sha>"}}`. The workflow fans out over the seven groups, and each `routing-parity-<group>` job runs its own group twice against the same test harness, once with `litellm/`, `enterprise/`, and `litellm-proxy-extras/` checked out from the base revision and once from the head, with a pytest plugin snapshotting `pg_stat_statements` into `routing-observed.json` per side. The `check` step then compares the two observations and writes `routing-diff.txt`: a statement seen on both sides fails when its role set changed, globally or for the same test (per-test capture is skipped under xdist), unless it is listed in `tests/integration/routing/either_role.json`, where each entry names the statement and a one-line reason it legitimately runs on whichever role asks for it, printed under `== either role ==`. Queries seen on only one side are listed, never failed, `pg_stat_statements` evictions and a role that never ran a statement are failures
+
+
+### MCP conformance
+
+The `mcp` group runs a pinned official conformance client against the official reference
+both directly and through a source-built gateway. Install it with
+`MCP_CONFORMANCE_ROOT=/tmp/mcp-conformance bash .circleci/scripts/install_mcp_conformance.sh`
+after the existing integration dependency setup, then use the normal MCP integration command.
+The destination must be fresh. The installer verifies each immutable archive before extraction
+and uses both upstream lockfiles. CircleCI installs this automatically and publishes runner logs,
+raw checks, negotiation observations and helper coverage with the existing integration artifacts.
+
+The runner is pinned to `7169291ec0b68eb370fddcd9947313ab0d5e4156`; the unmodified legacy
+reference is pinned separately to `8f3994c75ff1aed1e39f91cff9358e2bc2c81dcd`. The newer
+reference mistakes a valid legacy initialize containing `_meta: {}` for stateless traffic.
+Full download URLs and archive SHA-256 values live in the installer.
+
+Coverage is deliberately explicit:
+
+- The official client negotiates 2025-11-25. Its `--spec-version` flag selects assertions;
+  it does not pin its SDK handshake. Official cases cover each declared upstream revision.
+- Session termination is a separate mandatory gateway contract. Its raw HTTP control uses
+  the current pinned reference, which correctly returns 404 after deletion. The legacy
+  reference returns 400 for that case. Both references run unmodified with their own lockfiles.
+- The SDK matrix supplies the older-client gap: all 16 declared ordered revision pairs,
+  HTTP and SSE ingress, and HTTP, SSE and stdio upstreams. It checks the seven operations in
+  `capabilities.py` and records requested and returned revisions on both connections.
+- The official simple-text scenario accepts error text, and its reference rejects its omitted
+  arguments. Exact text with `{}` and omitted-argument forwarding therefore have explicit SDK
+  cases. This limitation cannot be treated as a successful official simple-text result.
+- The schema scenario looks up an unprefixed fixture name. A direct/gateway SDK comparison
+  instead requires the full JSON Schema 2020-12 input schema to survive unchanged.
+- The official DNS-rebinding scenario explicitly targets unauthenticated localhost servers.
+  This authenticated gateway instead has explicit allowed/denied Origin execution cases with
+  `LITELLM_CORS_ORIGINS` configured, covering HTTP and SSE. The default wildcard is unchanged.
+- Logging, completion, resource subscriptions, sampling and elicitation are not advertised by
+  this gateway contract; their capability-specific scenarios do not establish legacy support.
+  Modern 2026-07-28 and extension scenarios belong to later activation gates.
+
+The adapter changes only authentication and the fixture name to the gateway's advertised prefix.
+It preserves protocol versions, arguments, metadata, Origin/Host headers and response bytes.
+No discovery warm-up or expected-failure exemption is used. Missing, failed, skipped or wrongly
+negotiated required cases fail the MCP integration result.
+
+An installed workflow alone is not a merge gate. After these changes reach the default branch,
+verify the hosted MCP job's exact status name and add that check to the existing main ruleset.
+Do not mark LIT-7744 complete until all applicable cases pass and that requirement is active.

tests/integration/_support/conformance.py

@@ -4,8 +4,9 @@ import os
 import queue
 import signal
 import subprocess
-from collections.abc import Iterator
+from collections.abc import AsyncIterator, Iterator
 from contextlib import contextmanager
+from itertools import product
 from pathlib import Path
 from typing import TYPE_CHECKING, Final, Literal
 
@@ -35,10 +36,20 @@ def read_checks(directory: Path, scenario: str) -> tuple[ConformanceCheck, ...]:
     checks: Final = TypeAdapter(tuple[ConformanceCheck, ...]).validate_json(reports[0].read_bytes())
     assert len({check.id for check in checks}) == len(checks), f"conformance {scenario}: duplicate checks"
     required: Final = {
+        "server-session-lifecycle": (
+            "server-session-initialized-accepted",
+            "server-session-delete-accepted",
+            "server-session-terminated-returns-404",
+        ),
+        "server-sse-multiple-streams": (
+            "server-accepts-multiple-post-streams",
+            "server-sse-streams-functional",
+            "wire-schema-valid",
+        ),
         "server-initialize": ("server-initialize", "server-session-id-visible-ascii", "wire-schema-valid"),
         "tools-list": ("tools-list", "tools-name-format", "wire-schema-valid"),
         "tools-call-image": ("tools-call-image", "wire-schema-valid"),
-    }.get(scenario, (scenario,))
+    }.get(scenario, (scenario, "wire-schema-valid") if scenario in OFFICIAL_SCENARIOS else (scenario,))
     for identity in required:
         assert tuple(check.status for check in checks if check.id == identity) == ("SUCCESS",), (
             f"conformance {scenario}: required check {identity} did not pass: {checks}"
@@ -96,7 +107,7 @@ def reference_server(root: Path, directory: Path, port: int) -> Iterator["McpPee
     with (directory / "reference.log").open("w") as log:
         process: Final = subprocess.Popen(
             ["node", "--import", "tsx", "everything-server.ts"],
-            cwd=root / "legacy-reference/examples/servers/typescript",
+            cwd=root / "examples/servers/typescript",
             env={"PATH": os.environ["PATH"], "PORT": str(port)},
             stdout=log,
             stderr=subprocess.STDOUT,
@@ -130,7 +141,12 @@ def reference_server(root: Path, directory: Path, port: int) -> Iterator["McpPee
 
 
 @contextmanager
-def authenticated_endpoint(target: str, key: str, alias: str) -> Iterator[str]:
+def authenticated_endpoint(
+    target: str,
+    key: str | None,
+    alias: str | None,
+    negotiations: queue.Queue[tuple[str, str]] | None = None,
+) -> Iterator[str]:
     from integration._support.asgi import asgi_server
     from starlette.applications import Starlette
     from starlette.requests import Request
@@ -141,15 +157,29 @@ def authenticated_endpoint(target: str, key: str, alias: str) -> Iterator[str]:
     async def forward(scope: Scope, receive: Receive, send: Send) -> None:
         request: Final = Request(scope, receive)
         original: Final = await request.body()
-        body: Final = prefixed_request(original, alias)
+        body: Final = prefixed_request(original, alias) if alias is not None else original
         headers: Final = tuple(
             (name, value)
             for name, value in request.headers.raw
-            if name.lower() != b"authorization" and (name.lower() != b"content-length" or body == original)
-        ) + ((b"authorization", f"Bearer {key}".encode()),)
+            if (key is None or name.lower() != b"authorization")
+            and (name.lower() != b"content-length" or body == original)
+        ) + (((b"authorization", f"Bearer {key}".encode()),) if key is not None else ())
         async with httpx.AsyncClient(timeout=35, trust_env=False) as client:
             async with client.stream(request.method, target, headers=headers, content=body) as response:
-                streamed: Final = StreamingResponse(response.aiter_raw(), status_code=response.status_code)
+
+                async def observed_chunks() -> AsyncIterator[bytes]:
+                    capture: Final = (
+                        negotiations is not None and response.status_code == 200 and is_initialization(body)
+                    )
+                    chunks: Final[list[bytes]] = []
+                    async for chunk in response.aiter_raw():
+                        if capture:
+                            chunks.append(chunk)
+                        yield chunk
+                    if capture and negotiations is not None:
+                        negotiations.put(read_negotiation(body, b"".join(chunks), response.headers["content-type"]))
+
+                streamed: Final = StreamingResponse(observed_chunks(), status_code=response.status_code)
                 streamed.raw_headers = [
                     (name, value)
                     for name, value in response.headers.raw
@@ -172,3 +202,137 @@ def prefixed_request(body: bytes, alias: str) -> bytes:
     if not isinstance(params, dict) or not isinstance(params.get("name"), str):
         return body
     return json.dumps({**request, "params": {**params, "name": f"{alias}-{params['name']}"}}).encode()
+
+
+def require_negotiations(expected: str, observed: tuple[tuple[str, str], ...]) -> None:
+    assert observed and all(pair == (expected, expected) for pair in observed), (
+        f"conformance negotiation: expected {expected} on the wire, observed {observed}"
+    )
+
+
+class ExecutionReport(BaseModel):
+    collected: tuple[str, ...]
+    passed: tuple[str, ...]
+    skipped: tuple[str, ...]
+    complete: bool
+
+
+def require_passes(directory: Path, required: tuple[str, ...]) -> None:
+    report: Final = ExecutionReport.model_validate_json((directory / "execution.json").read_bytes())
+    assert required and report.complete, "conformance execution was incomplete"
+    assert set(required) <= set(report.collected), "conformance cases were not collected"
+    assert set(required) <= set(report.passed), "conformance cases did not pass"
+    assert not set(required).intersection(report.skipped), "conformance cases were skipped"
+
+
+def translation_cases() -> tuple[tuple[str, str, str, str], ...]:
+    from litellm.proxy._experimental.mcp_server.capabilities import REVISION_SUPPORT, TRANSLATION_PAIRS
+    from litellm.types.mcp import MCP_LEGACY_VERSIONS, MCPTransport
+
+    completed: Final = {revision for revision, support in REVISION_SUPPORT.items() if support.completed}
+    assert completed == set(MCP_LEGACY_VERSIONS), "New revisions need conformance cases before activation"
+    assert TRANSLATION_PAIRS == frozenset(product(completed, repeat=2)), "Declared translation coverage changed"
+    return tuple(
+        (downstream, upstream, peer, ingress)
+        for (downstream, upstream), peer, ingress in product(
+            sorted(TRANSLATION_PAIRS), ("http", "sse", "stdio"), ("http", "sse")
+        )
+        if MCPTransport(peer) in REVISION_SUPPORT[upstream].transports
+        and MCPTransport(ingress) in REVISION_SUPPORT[downstream].transports
+    )
+
+
+# Applicable legacy server scenarios for the operations advertised by capabilities.py.
+# The simple-text runner/reference mismatch has an explicit SDK gap case instead.
+OFFICIAL_SCENARIOS: Final = (
+    "server-initialize",
+    "server-sse-multiple-streams",
+    "ping",
+    "tools-list",
+    "tools-call-image",
+    "tools-call-audio",
+    "tools-call-embedded-resource",
+    "tools-call-mixed-content",
+    "tools-call-error",
+    "tools-call-with-progress",
+    "resources-list",
+    "resources-read-text",
+    "resources-read-binary",
+    "resources-templates-read",
+    "prompts-list",
+    "prompts-get-simple",
+    "prompts-get-with-args",
+    "prompts-get-embedded-resource",
+    "prompts-get-with-image",
+)
+
+
+def official_cases() -> tuple[tuple[str, str], ...]:
+    from litellm.types.mcp import MCP_LEGACY_VERSIONS
+
+    return tuple(product(OFFICIAL_SCENARIOS, MCP_LEGACY_VERSIONS))
+
+
+def required_conformance_nodes() -> tuple[str, ...]:
+    official: Final = tuple(
+        "tests/integration/mcp/test_mcp_official_conformance.py::test_official_scenario_through_gateway["
+        + "-".join(case)
+        + "]"
+        for case in official_cases()
+    )
+    matrix: Final = tuple(
+        "tests/integration/mcp/test_mcp_transports.py::test_pinned_revision_pairs_list_and_call_through_gateway["
+        + "-".join(case)
+        + "]"
+        for case in translation_cases()
+    )
+    return (
+        official
+        + matrix
+        + ("tests/integration/mcp/test_mcp_protocol_errors.py::test_omitted_tool_arguments_reach_the_upstream",)
+        + tuple(
+            "tests/integration/mcp/test_mcp_protocol_errors.py::test_configured_origin_policy_rejects_before_tool_execution["
+            + ingress
+            + "]"
+            for ingress in ("server_mcp", "sse")
+        )
+        + tuple(
+            "tests/integration/mcp/test_mcp_official_conformance.py::" + name
+            for name in (
+                "test_conformance_bridge_preserves_headers_payload_and_error_status[200]",
+                "test_conformance_bridge_preserves_headers_payload_and_error_status[403]",
+                "test_official_runner_rejects_unknown_scenario",
+                "test_official_gateway_session_lifecycle",
+                "test_stalled_reference_is_killed_and_cannot_report_clean_teardown",
+                "test_reference_children_are_stopped_after_the_root_exits",
+            )
+        )
+    )
+
+
+def read_negotiation(request: bytes, response: bytes, content_type: str) -> tuple[str, str]:
+    from httpx_sse import EventSource
+    from mcp.types import InitializeRequestParams, InitializeResult
+
+    body: Final = json.loads(request)
+    received: Final = httpx.Response(200, content=response, headers={"content-type": content_type})
+    messages: Final = (
+        tuple(event.json() for event in EventSource(received).iter_sse() if event.data)
+        if content_type.startswith("text/event-stream")
+        else (received.json(),)
+    )
+    replies: Final = tuple(
+        message["result"] for message in messages if message.get("id") == body["id"] and "result" in message
+    )
+    assert len(replies) == 1, "conformance negotiation: missing or duplicated initialize response"
+    return InitializeRequestParams.model_validate(body["params"]).protocol_version, InitializeResult.model_validate(
+        replies[0]
+    ).protocol_version
+
+
+def is_initialization(body: bytes) -> bool:
+    try:
+        parsed: Final = json.loads(body)
+    except (ValueError, UnicodeDecodeError):
+        return False
+    return isinstance(parsed, dict) and parsed.get("method") == "initialize"

tests/integration/_support/mcp.py

@@ -18,9 +18,10 @@ from integration._support.wire import Reply, Request, wire_server
 from mcp import ClientSession
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamable_http_client
+from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
 from mcp.server.mcpserver import Context, MCPServer
 from mcp.server.transport_security import TransportSecuritySettings
-from mcp.types import SamplingMessage, TextContent
+from mcp.types import InitializeRequestParams, InitializeResult, SamplingMessage, TextContent
 from mcp_tests.mcp_e2e_upstream_server import add, multiply
 from pydantic import BaseModel
 from sse_starlette.sse import AppStatus
@@ -63,8 +64,23 @@ class Confirmation(BaseModel):
     confirmed: bool
 
 
-def math_service(name: str = "integration-math", *, rich: bool = False) -> MCPServer:
+def math_service(
+    name: str = "integration-math",
+    *,
+    rich: bool = False,
+    record: Callable[[dict[str, object]], None],
+) -> MCPServer:
     service: Final = MCPServer(name)
+
+    async def capture_negotiation(ctx: ServerRequestContext[object, object], call_next: CallNext) -> HandlerResult:
+        result: Final = await call_next(ctx)
+        if ctx.method == "initialize":
+            requested: Final = InitializeRequestParams.model_validate(ctx.params).protocol_version
+            returned: Final = InitializeResult.model_validate(result).protocol_version
+            record({"body": {}, "negotiation": {"requested": requested, "returned": returned}})
+        return result
+
+    service.middleware.append(capture_negotiation)
     service.add_tool(add)
     service.add_tool(multiply)
 
@@ -184,14 +200,14 @@ def _draining_sse_watcher(app: Callable[[Scope, Receive, Send], object]):
 
 @contextmanager
 def mcp_peer(transport: Literal["http", "sse"] = "http", *, rich: bool = False) -> Iterator[McpPeer]:
-    service: Final = math_service(rich=rich)
+    observed: Final[queue.Queue[dict[str, object]]] = queue.Queue()
+    service: Final = math_service(rich=rich, record=observed.put)
     security: Final = TransportSecuritySettings(enable_dns_rebinding_protection=False)
     app: Final = (
         _draining_sse_watcher(service.sse_app(transport_security=security))
         if transport == "sse"
         else service.streamable_http_app(stateless_http=True, json_response=True, transport_security=security)
     )
-    observed: Final[queue.Queue[dict[str, object]]] = queue.Queue()
     with asgi_server(_capturing(app, observed), before_stop=_drain_sse_streams if transport == "sse" else None) as url:
         yield McpPeer(url + ("/sse" if transport == "sse" else "/mcp"), observed, transport)
 

tests/integration/_support/mcp_stdio_peer.py

@@ -38,7 +38,12 @@ class Recording:
 
 async def main() -> None:
     record: Final = Path(sys.argv[1])
-    service: Final = math_service("integration-stdio", rich=sys.argv[2] == "rich")
+
+    def record_negotiation(event: dict[str, object]) -> None:
+        with record.open("a") as sink:
+            sink.write(json.dumps(event) + "\n")
+
+    service: Final = math_service("integration-stdio", rich=sys.argv[2] == "rich", record=record_negotiation)
     stdin: Final = anyio.wrap_file(sys.stdin)
     async with stdio_server(stdin=Recording(stdin, record)) as (read_stream, write_stream):
         lowlevel: Final = service._lowlevel_server

tests/integration/conformance_coverage.toml

@@ -3,7 +3,13 @@ branch = true
 parallel = true
 relative_files = true
 patch = ["subprocess"]
-include = ["tests/integration/_support/conformance.py", "tests/integration/_support/process.py"]
+include = [
+    "tests/integration/_support/conformance.py",
+    "tests/integration/_support/process.py",
+    "tests/integration/_support/mcp.py",
+    "tests/integration/_support/mcp_stdio_peer.py",
+    "tests/integration/run.py",
+]
 
 [tool.coverage.report]
 show_missing = true

tests/integration/mcp/test_mcp_official_conformance.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import os
+import queue
 import uuid
 from pathlib import Path
 from typing import Final
@@ -8,49 +9,110 @@ from typing import Final
 import httpx
 import pytest
 from integration._support.client import Gateway
-from integration._support.conformance import authenticated_endpoint, reference_server, run_scenario
-from integration._support.mcp import official_client_outcomes, register_mcp
+from integration._support.conformance import (
+    authenticated_endpoint,
+    official_cases,
+    reference_server,
+    require_negotiations,
+    run_scenario,
+)
+from integration._support.mcp import McpPeer, mcp_peer, official_client_outcomes, register_mcp
 from integration._support.wire import Reply, wire_server
 from mcp import ClientSession
 from mcp.client.streamable_http import streamable_http_client
 from mcp.types import ImageContent
+from pydantic import TypeAdapter
 
 
-@pytest.mark.parametrize("name", ("server-initialize", "tools-list", "tools-call-image"))
-def test_official_scenario_through_gateway(gateway: Gateway, tmp_path: Path, unused_tcp_port: int, name: str) -> None:
+@pytest.mark.parametrize(
+    ("name", "upstream"), tuple(pytest.param(*case, id="-".join(case)) for case in official_cases())
+)
+def test_official_scenario_through_gateway(
+    gateway: Gateway, tmp_path: Path, unused_tcp_port: int, name: str, upstream: str
+) -> None:
     root: Final = Path(os.environ["MCP_CONFORMANCE_ROOT"])
     output: Final = Path(os.environ.get("INTEGRATION_RESULTS_DIR", str(tmp_path))) / f"conformance-{uuid.uuid4().hex}"
-    with reference_server(root, output, unused_tcp_port) as reference, gateway.scenario() as scenario:
+    with reference_server(root / "legacy-reference", output, unused_tcp_port) as reference, gateway.scenario() as scenario:
         alias: Final = "official" + uuid.uuid4().hex[:8]
-        identity: Final = register_mcp(scenario, reference, alias, mcp_info={"protocol_version": "2025-11-25"})
+        upstream_wire: Final[queue.Queue[tuple[str, str]]] = queue.Queue()
+        downstream_wire: Final[queue.Queue[tuple[str, str]]] = queue.Queue()
+        with authenticated_endpoint(reference.url, None, None, upstream_wire) as recorded_reference:
+            identity: Final = register_mcp(
+                scenario, McpPeer(recorded_reference, queue.Queue()), alias, mcp_info={"protocol_version": upstream}
+            )
+            key: Final = scenario.key(object_permission={"mcp_servers": [identity]})
+            direct: Final = run_scenario(root, reference.url, name, output / "direct")
+            endpoint: Final = str(gateway.client.base_url).rstrip("/") + f"/{alias}/mcp"
+            with authenticated_endpoint(endpoint, key, alias, downstream_wire) as authenticated:
+                proxied: Final = run_scenario(root, authenticated, name, output / "gateway")
+            direct_checks: Final = {check.id: check for check in direct}
+            gateway_checks: Final = {check.id: check for check in proxied}
+            if name in ("tools-list", "prompts-list"):
+                field: Final = "tools" if name == "tools-list" else "prompts"
+                names: Final = TypeAdapter(tuple[str, ...]).validate_python(direct_checks[name].details[field])
+                assert names, "Official reference listed no fixtures"
+                assert gateway_checks[name].details[field] == [f"{alias}-{fixture}" for fixture in names]
+            if name == "tools-list":
+
+                async def schema(url: str, tool_name: str, token: str | None = None) -> dict[str, object]:
+                    async with httpx.AsyncClient(headers={"Authorization": f"Bearer {token}"} if token else {}) as http:
+                        async with streamable_http_client(url, http_client=http) as streams:
+                            async with ClientSession(streams[0], streams[1]) as session:
+                                await session.initialize()
+                                listed: Final = await session.list_tools()
+                                return next(tool.input_schema for tool in listed.tools if tool.name == tool_name)
+
+                original_schema: Final = asyncio.run(schema(reference.url, "json_schema_2020_12_tool"))
+                assert original_schema["$schema"] == "https://json-schema.org/draft/2020-12/schema"
+                assert original_schema["$defs"] and original_schema["additionalProperties"] is False
+                assert asyncio.run(schema(endpoint, f"{alias}-json_schema_2020_12_tool", key)) == original_schema
+            if name == "tools-call-image":
+
+                async def check_image(url: str, tool: str, token: str | None = None) -> None:
+                    async with httpx.AsyncClient(headers={"Authorization": f"Bearer {token}"} if token else {}) as http:
+                        async with streamable_http_client(url, http_client=http) as streams:
+                            async with ClientSession(streams[0], streams[1]) as session:
+                                await session.initialize()
+                                result: Final = await session.call_tool(tool, {})
+                                assert result.is_error is False and len(result.content) == 1, result
+                                content: Final = result.content[0]
+                                assert isinstance(content, ImageContent) and content.mime_type == "image/png", content
+                                assert content.data == (
+                                    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlE"
+                                    "QVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="
+                                ), content
+
+                asyncio.run(check_image(reference.url, "test_image_content"))
+                asyncio.run(check_image(endpoint, f"{alias}-test_image_content", key))
+            if name == "tools-call-error":
+                assert gateway_checks[name].details["result"] == direct_checks[name].details["result"], (
+                    "An unrelated error masked the fixture error"
+                )
+            listed, called = official_client_outcomes(gateway, key, f"/{alias}/mcp", f"{alias}-test_simple_text", {})
+            assert f"{alias}-test_simple_text" in listed.tools, listed
+            assert called.ok and called.text == "This is a simple text response for testing.", called
+            downstream_observed: Final = tuple(downstream_wire.get_nowait() for _ in range(downstream_wire.qsize()))
+            upstream_observed: Final = tuple(upstream_wire.get_nowait() for _ in range(upstream_wire.qsize()))
+            (output / "negotiation.json").write_text(
+                json.dumps({"client_gateway": downstream_observed, "gateway_reference": upstream_observed}, indent=2)
+                + "\n"
+            )
+            require_negotiations("2025-11-25", downstream_observed)
+            require_negotiations(upstream, upstream_observed)
+
+
+def test_official_gateway_session_lifecycle(gateway: Gateway, tmp_path: Path, unused_tcp_port: int) -> None:
+    root: Final = Path(os.environ["MCP_CONFORMANCE_ROOT"])
+    output: Final = Path(os.environ.get("INTEGRATION_RESULTS_DIR", str(tmp_path))) / f"lifecycle-{uuid.uuid4().hex}"
+    with reference_server(root, output, unused_tcp_port) as reference:
+        run_scenario(root, reference.url, "server-session-lifecycle", output / "direct")
+    with mcp_peer() as upstream, gateway.scenario() as scenario:
+        alias: Final = "lifecycle" + uuid.uuid4().hex[:8]
+        identity: Final = register_mcp(scenario, upstream, alias)
         key: Final = scenario.key(object_permission={"mcp_servers": [identity]})
-        direct: Final = run_scenario(root, reference.url, name, output / "direct")
         endpoint: Final = str(gateway.client.base_url).rstrip("/") + f"/{alias}/mcp"
         with authenticated_endpoint(endpoint, key, alias) as authenticated:
-            proxied: Final = run_scenario(root, authenticated, name, output / "gateway")
-        if name == "tools-call-image":
-
-            async def check_image(url: str, tool: str, token: str | None = None) -> None:
-                async with httpx.AsyncClient(headers={"Authorization": f"Bearer {token}"} if token else {}) as http:
-                    async with streamable_http_client(url, http_client=http) as streams:
-                        async with ClientSession(streams[0], streams[1]) as session:
-                            await session.initialize()
-                            result: Final = await session.call_tool(tool, {})
-                            assert result.is_error is False and len(result.content) == 1, result
-                            content: Final = result.content[0]
-                            assert isinstance(content, ImageContent) and content.mime_type == "image/png", content
-                            assert content.data == (
-                                "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlE"
-                                "QVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="
-                            ), content
-
-            asyncio.run(check_image(reference.url, "test_image_content"))
-            asyncio.run(check_image(endpoint, f"{alias}-test_image_content", key))
-        assert tuple(check.status for check in direct if check.id == name) == ("SUCCESS",)
-        assert tuple(check.status for check in proxied if check.id == name) == ("SUCCESS",)
-        listed, called = official_client_outcomes(gateway, key, f"/{alias}/mcp", f"{alias}-test_simple_text", {})
-        assert f"{alias}-test_simple_text" in listed.tools, listed
-        assert called.ok and called.text == "This is a simple text response for testing.", called
+            run_scenario(root, authenticated, "server-session-lifecycle", output / "gateway")
 
 
 @pytest.mark.parametrize("status", (200, 403))
@@ -106,7 +168,7 @@ def test_stalled_reference_is_killed_and_cannot_report_clean_teardown(tmp_path:
 
     children: Final = frozenset(child.pid for child in psutil.Process().children())
     with ExitStack() as cleanup:
-        cleanup.enter_context(reference_server(Path(os.environ["MCP_CONFORMANCE_ROOT"]), tmp_path, unused_tcp_port))
+        cleanup.enter_context(reference_server(Path(os.environ["MCP_CONFORMANCE_ROOT"]) / "legacy-reference", tmp_path, unused_tcp_port))
         started: Final = tuple(child for child in psutil.Process().children() if child.pid not in children)
         assert len(started) == 1, started
         victim: Final = started[0]
@@ -119,7 +181,7 @@ def test_stalled_reference_is_killed_and_cannot_report_clean_teardown(tmp_path:
 def test_reference_children_are_stopped_after_the_root_exits(tmp_path: Path, unused_tcp_port: int) -> None:
     import psutil
 
-    source: Final = tmp_path / "legacy-reference/examples/servers/typescript"
+    source: Final = tmp_path / "examples/servers/typescript"
     source.mkdir(parents=True)
     (source / "node_modules").symlink_to(
         Path(os.environ["MCP_CONFORMANCE_ROOT"]) / "legacy-reference/examples/servers/typescript/node_modules",

tests/integration/mcp/test_mcp_protocol_errors.py

@@ -1,12 +1,13 @@
 import json
 import queue
 import uuid
+from pathlib import Path
 from typing import Final
 
 import pytest
 
 from integration._support.client import Gateway
-from integration._support.mcp import McpPeer, call_tool, register_mcp, tool_names
+from integration._support.mcp import EntryPoint, McpPeer, call_tool, register_mcp, tool_names
 from integration._support.wire import Reply, Request, wire_server
 
 
@@ -112,14 +113,26 @@ def test_configured_revision_blocks_unadvertised_handshake_and_keeps_allowed_con
         with owned_proxy(gateway, tmp_path, {"DISABLE_SCHEMA_UPDATE": "true"}, config=config_path) as restricted:
             endpoint: Final = str(restricted.client.base_url).rstrip("/") + ("/mcp/sse" if ingress == "sse" else "/mcp")
             headers: Final = {"Authorization": f"Bearer {key}", "x-mcp-servers": identity}
-            denied: Final = MCPClient(server_url=endpoint, transport_type=MCPTransport(ingress), protocol_version="2025-11-25", extra_headers=headers)
-            allowed: Final = MCPClient(server_url=endpoint, transport_type=MCPTransport(ingress), protocol_version="2024-11-05", extra_headers=headers)
+            denied: Final = MCPClient(
+                server_url=endpoint,
+                transport_type=MCPTransport(ingress),
+                protocol_version="2025-11-25",
+                extra_headers=headers,
+            )
+            allowed: Final = MCPClient(
+                server_url=endpoint,
+                transport_type=MCPTransport(ingress),
+                protocol_version="2024-11-05",
+                extra_headers=headers,
+            )
 
             async def exercise() -> None:
                 with pytest.raises(MCPError, match="Unsupported MCP protocol version"):
                     await denied.list_tools(raise_on_error=True)
                 assert f"{alias}-add" in tuple(tool.name for tool in await allowed.list_tools(raise_on_error=True))
-                result: Final = await allowed.call_tool(CallToolRequestParams(name=f"{alias}-add", arguments={"a": 2, "b": 5}))
+                result: Final = await allowed.call_tool(
+                    CallToolRequestParams(name=f"{alias}-add", arguments={"a": 2, "b": 5})
+                )
                 assert result.is_error is False and result.content[0].text == "7"
 
             asyncio.run(exercise())
@@ -167,3 +180,28 @@ def test_omitted_tool_arguments_reach_the_upstream(gateway: Gateway) -> None:
         denied: Final = asyncio.run(invoke(aggregate, f"{alias}-fail", denied_key))
         assert denied.is_error is True and "not allowed" in denied.content[0].text.lower(), denied
         assert tool_calls(reference.drain()) == (), "Denied caller reached the upstream"
+
+
+@pytest.mark.parametrize("ingress", ("server_mcp", "sse"))
+def test_configured_origin_policy_rejects_before_tool_execution(
+    gateway: Gateway, tmp_path: Path, ingress: EntryPoint
+) -> None:
+    from integration._support.mcp import McpCaller, mcp_peer, tool_calls
+    from integration._support.process import owned_proxy
+
+    with mcp_peer() as reference, gateway.scenario() as scenario:
+        alias: Final = "origin" + uuid.uuid4().hex[:8]
+        identity: Final = register_mcp(scenario, reference, alias)
+        key: Final = scenario.key(object_permission={"mcp_servers": [identity]})
+        with owned_proxy(gateway, tmp_path, {"LITELLM_CORS_ORIGINS": "https://allowed.example"}) as isolated:
+            reference.drain()
+            denied: Final = McpCaller(isolated, key, ingress, alias, {"Origin": "https://untrusted.example"}).call(
+                f"{alias}-add", {"a": 3, "b": 4}
+            )
+            assert denied.status == 403, denied
+            assert tool_calls(reference.drain()) == (), "Disallowed Origin executed a tool"
+            allowed: Final = McpCaller(isolated, key, ingress, alias, {"Origin": "https://allowed.example"}).call(
+                f"{alias}-add", {"a": 3, "b": 4}
+            )
+            assert allowed.ok and allowed.text == "7", allowed
+            assert len(tool_calls(reference.drain())) == 1

tests/integration/mcp/test_mcp_transports.py

@@ -4,6 +4,7 @@ from typing import Final
 
 import pytest
 from integration._support.client import Gateway
+from integration._support.conformance import require_negotiations, translation_cases
 from integration._support.mcp import (
     ENTRY_POINTS,
     PEER_KINDS,
@@ -157,41 +158,88 @@ def test_server_initiated_sampling_and_elicitation_surface_as_errors_not_success
         assert len(tool_calls(peer.drain())) == 1
 
 
-@pytest.mark.parametrize("downstream", ("2024-11-05", "2025-03-26", "2025-06-18", "2025-11-25"))
-@pytest.mark.parametrize("upstream", ("2024-11-05", "2025-03-26", "2025-06-18", "2025-11-25"))
-@pytest.mark.parametrize("peer_kind", ("http", "sse", "stdio"))
-@pytest.mark.parametrize("ingress", ("http", "sse"))
+@pytest.mark.parametrize(
+    ("downstream", "upstream", "peer_kind", "ingress"),
+    tuple(pytest.param(*case, id="-".join(case)) for case in translation_cases()),
+)
 def test_pinned_revision_pairs_list_and_call_through_gateway(
-    gateway: Gateway, downstream: str, upstream: str, peer_kind: PeerKind, ingress: str
+    gateway: Gateway, downstream: str, upstream: str, peer_kind: PeerKind, ingress: str, tmp_path
 ) -> None:
     import asyncio
+    import os
+    from pathlib import Path
 
-    from mcp.types import CallToolRequestParams
+    import httpx
+    from mcp import ClientSession
+    from mcp.client.sse import sse_client
+    from mcp.client.streamable_http import streamable_http_client
+    from mcp.types import (
+        ClientCapabilities,
+        Implementation,
+        InitializeRequest,
+        InitializeRequestParams,
+        InitializeResult,
+        InitializedNotification,
+    )
 
-    from litellm.experimental_mcp_client.client import MCPClient
-    from litellm.types.mcp import MCPTransport
-
-    with peer_of(peer_kind) as peer, gateway.scenario() as scenario:
+    with peer_of(peer_kind, rich=True) as peer, gateway.scenario() as scenario:
         alias: Final = "versions" + uuid.uuid4().hex[:8]
         identity: Final = register_mcp(scenario, peer, alias, mcp_info={"protocol_version": upstream})
         key: Final = scenario.key(object_permission={"mcp_servers": [identity]})
         endpoint: Final = str(gateway.client.base_url).rstrip("/") + ("/mcp/sse" if ingress == "sse" else "/mcp")
-        client: Final = MCPClient(
-            server_url=endpoint, transport_type=MCPTransport(ingress), protocol_version=downstream,
-            extra_headers={"Authorization": f"Bearer {key}", "x-mcp-servers": identity}, timeout=15,
-        )
+        headers: Final = {"Authorization": f"Bearer {key}", "x-mcp-servers": identity}
 
-        async def exercise() -> None:
-            tools: Final = await client.list_tools(raise_on_error=True)
-            assert f"{alias}-add" in tuple(tool.name for tool in tools)
-            result: Final = await client.call_tool(CallToolRequestParams(name=f"{alias}-add", arguments={"a": 3, "b": 4}))
-            assert result.is_error is False
-            assert result.content[0].text == "7"
+        async def exercise() -> tuple[str, str]:
+            async with httpx.AsyncClient(headers=headers, timeout=15) as http:
+                transport: Final = (
+                    sse_client(endpoint, headers=headers)
+                    if ingress == "sse"
+                    else streamable_http_client(endpoint, http_client=http)
+                )
+                async with transport as streams, ClientSession(streams[0], streams[1]) as session:
+                    initialized: Final = await session.send_request(
+                        InitializeRequest(
+                            params=InitializeRequestParams(
+                                protocol_version=downstream,
+                                capabilities=ClientCapabilities(),
+                                client_info=Implementation(name="conformance-gap-client", version="1"),
+                            )
+                        ),
+                        InitializeResult,
+                    )
+                    require_negotiations(downstream, ((downstream, initialized.protocol_version),))
+                    session.adopt(initialized)
+                    await session.send_notification(InitializedNotification())
+                    tools: Final = await session.list_tools()
+                    assert f"{alias}-add" in tuple(tool.name for tool in tools.tools)
+                    result: Final = await session.call_tool(f"{alias}-add", {"a": 3, "b": 4})
+                    assert result.is_error is False
+                    assert result.content[0].text == "7"
+                    prompts: Final = await session.list_prompts()
+                    assert tuple(prompt.name for prompt in prompts.prompts) == (f"{alias}-greeting",)
+                    prompt: Final = await session.get_prompt(f"{alias}-greeting", {"name": "Ada"})
+                    assert prompt.messages[0].content.text == "Hello, Ada"
+                    resources: Final = await session.list_resources()
+                    assert tuple(str(resource.uri) for resource in resources.resources) == ("status://ready",)
+                    resource: Final = await session.read_resource(resources.resources[0].uri)
+                    assert resource.contents[0].text == "ready"
+                    templates: Final = await session.list_resource_templates()
+                    assert tuple(template.uri_template for template in templates.resource_templates) == (
+                        "greeting://{name}",
+                    )
+                    return downstream, initialized.protocol_version
 
         peer.drain()
-        asyncio.run(exercise())
+        negotiated: Final = asyncio.run(exercise())
         observed: Final = peer.drain()
-        negotiations: Final = tuple(item["body"] for item in observed if item["body"].get("method") == "initialize")
-        assert negotiations, "The operation must reach the upstream negotiation"
-        assert all(request["params"]["protocolVersion"] == upstream for request in negotiations), negotiations
+        negotiations: Final = tuple(
+            (item["negotiation"]["requested"], item["negotiation"]["returned"])
+            for item in observed
+            if "negotiation" in item
+        )
+        require_negotiations(upstream, negotiations)
         assert len(tool_calls(observed)) == 1
+        output: Final = Path(os.environ.get("INTEGRATION_RESULTS_DIR", str(tmp_path)))
+        (output / f"negotiation-{downstream}-{upstream}-{peer_kind}-{ingress}.json").write_text(
+            json.dumps({"client_gateway": negotiated, "gateway_upstream": negotiations}, indent=2) + "\n"
+        )

tests/integration/run.py

@@ -92,6 +92,10 @@ def main() -> int:
     if not evidence["complete"]:
         sys.stderr.write("Integration run did not complete: a collected node neither passed nor skipped\n")
         return 1
+    if options.group == "mcp":
+        from integration._support.conformance import require_passes, required_conformance_nodes
+
+        require_passes(output, required_conformance_nodes())
     return 0
 
 

tests/unit/integration_support/test_conformance.py

@@ -6,7 +6,15 @@ from typing import Final
 from unittest.mock import Mock, patch
 
 import pytest
-from tests.integration._support.conformance import read_checks, verify_archive, prefixed_request
+from tests.integration._support.conformance import (
+    is_initialization,
+    read_negotiation,
+    prefixed_request,
+    read_checks,
+    require_negotiations,
+    require_passes,
+    verify_archive,
+)
 from pydantic import ValidationError
 
 
@@ -161,6 +169,121 @@ def test_name_integration_preserves_other_requests(body: bytes) -> None:
     assert prefixed_request(body, "official") == body
 
 
+@pytest.mark.parametrize("observed", ((), (("2025-11-25", "2025-11-25"),), (("2025-03-26", "2025-11-25"),)))
+def test_wrong_or_missing_negotiation_cannot_pass(observed: tuple[tuple[str, str], ...]) -> None:
+    with pytest.raises(AssertionError, match="negotiation"):
+        require_negotiations("2025-03-26", observed)
+
+
+def test_actual_requested_and_returned_revision_are_required() -> None:
+    assert require_negotiations("2025-03-26", (("2025-03-26", "2025-03-26"),)) is None
+
+
+@pytest.mark.parametrize(
+    "collected,passed,skipped,complete",
+    (
+        ([], [], [], True),
+        (["one"], ["one"], [], True),
+        (["one", "two"], ["one"], ["two"], True),
+        (["one", "two"], ["one", "two"], [], False),
+    ),
+)
+def test_missing_skipped_or_incomplete_gate_cannot_pass(
+    tmp_path: Path, collected: list[str], passed: list[str], skipped: list[str], complete: bool
+) -> None:
+    (tmp_path / "execution.json").write_text(
+        json.dumps({"collected": collected, "passed": passed, "skipped": skipped, "complete": complete})
+    )
+    with pytest.raises(AssertionError, match="conformance"):
+        require_passes(tmp_path, ("one", "two"))
+
+
+def test_complete_gate_requires_every_declared_case(tmp_path: Path) -> None:
+    (tmp_path / "execution.json").write_text(
+        json.dumps(
+            {
+                "collected": ["one", "two", "unrelated"],
+                "passed": ["one", "two"],
+                "skipped": ["unrelated"],
+                "complete": True,
+            }
+        )
+    )
+    assert require_passes(tmp_path, ("one", "two")) is None
+
+
+@pytest.mark.parametrize("streamed", (False, True))
+def test_negotiation_evidence_reads_the_actual_reply(streamed: bool) -> None:
+    request: Final = json.dumps(
+        {
+            "id": 1,
+            "params": {
+                "protocolVersion": "2025-03-26",
+                "capabilities": {},
+                "clientInfo": {"name": "test", "version": "1"},
+            },
+        }
+    ).encode()
+    result: Final = json.dumps(
+        {
+            "id": 1,
+            "result": {
+                "protocolVersion": "2025-11-25",
+                "capabilities": {},
+                "serverInfo": {"name": "test", "version": "1"},
+            },
+        }
+    ).encode()
+    response: Final = b"data:\n\n: heartbeat\n\nevent: message\ndata: " + result + b"\n\n" if streamed else result
+    assert read_negotiation(request, response, "text/event-stream" if streamed else "application/json") == (
+        "2025-03-26",
+        "2025-11-25",
+    )
+
+
+def test_unrelated_rpc_response_cannot_supply_negotiation() -> None:
+    with pytest.raises(AssertionError, match="initialize response"):
+        read_negotiation(b'{"id":1}', b'{"id":2,"result":{}}', "application/json")
+
+
+@pytest.mark.parametrize(
+    "body,expected",
+    ((b"", False), (b"[]", False), (b'{"method":"tools/list"}', False), (b'{"method":"initialize"}', True)),
+)
+def test_only_initialize_is_captured(body: bytes, expected: bool) -> None:
+    assert is_initialization(body) is expected
+
+@pytest.mark.parametrize(
+    "scenario,identities",
+    (
+        (
+            "server-session-lifecycle",
+            (
+                "server-session-initialized-accepted",
+                "server-session-delete-accepted",
+                "server-session-terminated-returns-404",
+            ),
+        ),
+        (
+            "server-sse-multiple-streams",
+            ("server-accepts-multiple-post-streams", "server-sse-streams-functional", "wire-schema-valid"),
+        ),
+    ),
+)
+def test_complete_transport_checks_pass_and_missing_checks_fail(
+    tmp_path: Path, scenario: str, identities: tuple[str, ...]
+) -> None:
+    report: Final = tmp_path / "checks.json"
+    report.write_text(json.dumps([{"id": identity, "status": "SUCCESS"} for identity in identities]))
+    assert len(read_checks(tmp_path, scenario)) == len(identities)
+    for missing in identities:
+        report.write_text(
+            json.dumps([{"id": identity, "status": "SUCCESS"} for identity in identities if identity != missing])
+        )
+        with pytest.raises(AssertionError, match="required check"):
+            read_checks(tmp_path, scenario)
+
+
 @pytest.mark.parametrize(
     "scenario,identities",
     (