diff --git a/.circleci/config.yml b/.circleci/config.yml index 995480f8784..835f780b9a5 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -3213,6 +3213,8 @@ jobs: resource_class: large working_directory: ~/project parallelism: << parameters.parallelism >> + environment: + MCP_CONFORMANCE_CODECOV_FLAG: mcp-conformance steps: - setup_litellm_test_deps - when: @@ -3266,7 +3268,7 @@ jobs: steps: - codecov/upload: file: test-results/integration-mcp/conformance-coverage.xml - flags: mcp-conformance + flags: MCP_CONFORMANCE_CODECOV_FLAG - run: name: Stop owned database and Redis when: always @@ -3276,8 +3278,18 @@ jobs: docker logs redis-cache > test-results/services-<< parameters.suite >>-<< parameters.mode >>/redis.log 2>&1 || true docker rm -f postgres-db redis-cache test -z "$(docker ps -aq --filter name=postgres-db --filter name=redis-cache)" - - store_test_results: - path: test-results + - when: + condition: + equal: [mcp, << parameters.suite >>] + steps: + - store_test_results: + path: test-results/integration-mcp/junit.xml + - unless: + condition: + equal: [mcp, << parameters.suite >>] + steps: + - store_test_results: + path: test-results - store_artifacts: path: test-results diff --git a/.circleci/scripts/install_mcp_conformance.sh b/.circleci/scripts/install_mcp_conformance.sh index 8bc5eb22750..8b0615316af 100644 --- a/.circleci/scripts/install_mcp_conformance.sh +++ b/.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 \ diff --git a/tests/integration/README.md b/tests/integration/README.md index c09904597ab..74ac09a08ad 100644 --- a/tests/integration/README.md +++ b/tests/integration/README.md @@ -14,7 +14,7 @@ Reuse the existing canned provider handlers through `_support/upstream.py`. It r The CircleCI workflow starts its own database and Redis, restricts test-phase egress to its owned services and writes JUnit plus an executed-node manifest. Missing setup, failed cleanup or a selected test with neither a passed call nor a skip fail qualification. Skipped nodes are listed under `skipped` in `execution.json`, so the skip reasons double as the open bug list. Existing GitHub Actions jobs do not own these tests -There is no per-node manifest. The runner fails only when pytest fails, when collection errors, or when a selected file collects zero tests. Older tests still carry `@pytest.mark.covers(...)` decorators; the marker stays registered so they collect, but the IDs are not checked against anything and new tests should not use it. The GitHub Actions coverage census reads the `GROUPS` literal in `run.py` and treats every `tests/integration//test_*.py` file in a scheduled group as owned by CircleCI +Groups other than MCP have no per-node manifest. Those groups fail when pytest fails, when collection errors, or when a selected file collects zero tests. MCP additionally enforces the required baseline described below. Older tests still carry `@pytest.mark.covers(...)` decorators; the marker stays registered so they collect, but the IDs are not checked against anything and new tests should not use it. The GitHub Actions coverage census reads the `GROUPS` literal in `run.py` and treats every `tests/integration//test_*.py` file in a scheduled group as owned by CircleCI Provider sentinels currently use the controlled server, not live recordings. The provider shard also runs the existing strict replay controls for changed requests, exhausted interactions, leftover interactions and no provider connection. Future recorded scenarios must use that replay-only implementation; missing recordings cannot fall back to a real provider. The observation endpoint is destructive and the current selection runs serially against one owned upstream @@ -37,3 +37,92 @@ 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-` 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 regression and conformance baseline + +This gate protects existing behavior while the stateless refactor proceeds. Completion establishes a +passing, enforced baseline; it does not certify exhaustive conformance or future capabilities. +Each later implementation preserves this baseline and adds tests for the behavior it changes + +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. + +`mcp/conformance_baseline.json` is the independent required-coverage contract. Its 19 official +scenarios across four upstream revisions, 96 SDK revision/transport combinations, and nine explicit +checks require 181 passing nodes. Test generation does not read this file, so removing a scenario, +transport case or test file cannot silently remove its requirement. Capability mappings name the +required test families that exercise each revision's operations and extensions. A completed +revision or capability without mapped required coverage fails the gate + +To extend the gate, add meaningful behavior assertions to the existing mapped integration file, +then update the baseline with the required cases and corresponding capability mapping in the same +PR. Review that the mapped test actually exercises the capability. A mapping alone is not proof. +Keep required cases passing without skip/xfail exemptions. Track an existing uncovered requirement +with its owning ticket; do not remove baseline protection or advertise unverified new support + +CircleCI currently runs MCP on one node with four pytest workers. The pytest controller combines +all worker reports into `execution.json`; the full invocation checks all required nodes. Missing +worker results, absent files, skips and incomplete runs fail the baseline. Local file selections +check the required cases in those files and are not evidence of a complete gate. If MCP is later +split across CircleCI nodes, aggregate their results against the entire baseline before accepting +the job; per-file successes alone are insufficient + +Remaining coverage grows with the owning implementation: + +| Behavior | Owner and acceptance boundary | +| --- | --- | +| Modern upstream calls without initialization; affected legacy auth/policy parity | LIT-7745, tested with its implementation | +| Authorized list/call across cold replicas without session affinity | LIT-4500, tested with its implementation | +| Full schema/result preservation and remaining upstream pagination | LIT-7750 and LIT-5594 | +| Input relay, MRTR and event/cancellation behavior | LIT-4508, LIT-7752 and LIT-4510 | +| Expanded OAuth and security regressions | LIT-3467, LIT-3559 and LIT-4506 alongside the affected fixes | +| Applicable combined conformance, canary and rollback | LIT-8305 before the corresponding release | + +The SDK matrix provides successful legacy-operation compatibility coverage. It does not repeat +every official rich-content, error, progress or lifecycle assertion for every older client and +transport. Expand those cases when relevant paths change; this limitation does not block unrelated +implementation. Modern revision and extension activation still requires their applicable tests + +An installed workflow alone is not an enforced merge gate. Require the hosted MCP job's exact +status in the existing main ruleset. Close LIT-7744 after this baseline is merged, passes and is +required. LIT-7745 can be developed alongside gate completion; its shared-path changes must pass +the baseline and their added legacy/modern tests before merging. Later coverage extensions belong +to their implementation tickets rather than keeping LIT-7744 open indefinitely diff --git a/tests/integration/_support/conformance.py b/tests/integration/_support/conformance.py index f864bfce666..00d06adabdf 100644 --- a/tests/integration/_support/conformance.py +++ b/tests/integration/_support/conformance.py @@ -4,16 +4,19 @@ import os import queue import signal import subprocess -from collections.abc import Iterator +from collections.abc import AsyncIterator, Iterator, Mapping from contextlib import contextmanager +from itertools import product from pathlib import Path from typing import TYPE_CHECKING, Final, Literal import httpx import psutil -from pydantic import BaseModel, Field, JsonValue, TypeAdapter +from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter if TYPE_CHECKING: + from litellm.proxy._experimental.mcp_server.capabilities import RevisionSupport + from integration._support.mcp import McpPeer @@ -35,10 +38,19 @@ 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", + ), "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 +108,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 +142,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 +158,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 +203,173 @@ 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" + missing: Final = set(required) - set(report.collected) + assert not missing, f"conformance cases were not collected: {sorted(missing)}" + unsuccessful: Final = set(required) - set(report.passed) + assert not unsuccessful, f"conformance cases did not pass: {sorted(unsuccessful)}" + skipped: Final = set(required).intersection(report.skipped) + assert not skipped, f"conformance cases were skipped: {sorted(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)) + + +class CapabilityContract(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + revisions: tuple[str, ...] = Field(min_length=1) + operations: tuple[str, ...] + extensions: tuple[str, ...] + tests: tuple[str, ...] = Field(min_length=1) + + +class ConformanceBaseline(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + revisions: tuple[str, ...] = Field(min_length=1) + ingress_transports: tuple[str, ...] = Field(min_length=1) + upstream_transports: tuple[str, ...] = Field(min_length=1) + official_scenarios: tuple[str, ...] = Field(min_length=1) + capabilities: tuple[CapabilityContract, ...] = Field(min_length=1) + required_tests: tuple[str, ...] = Field(min_length=1) + + +def require_capability_coverage( + baseline: ConformanceBaseline, + required: tuple[str, ...], + support: Mapping[str, "RevisionSupport"], +) -> None: + families: Final = {node.split("[", 1)[0] for node in required} + for contract in baseline.capabilities: + assert set(contract.revisions) <= set(baseline.revisions), "Conformance capability maps an untested revision" + assert set(contract.tests) <= families, "Conformance capability maps a test outside the required baseline" + for revision, advertised in support.items(): + if not advertised.completed: + continue + contracts: Final = tuple(contract for contract in baseline.capabilities if revision in contract.revisions) + assert contracts, f"Conformance capability coverage missing for revision {revision}" + operations: Final = {operation for contract in contracts for operation in contract.operations} + extensions: Final = {extension for contract in contracts for extension in contract.extensions} + assert advertised.operations <= operations, ( + f"Conformance capability coverage missing for {revision}: {sorted(advertised.operations - operations)}" + ) + assert advertised.extensions <= extensions, ( + f"Conformance capability coverage missing for {revision}: {sorted(advertised.extensions - extensions)}" + ) + + +def required_conformance_nodes() -> tuple[str, ...]: + from litellm.proxy._experimental.mcp_server.capabilities import REVISION_SUPPORT + + baseline: Final = ConformanceBaseline.model_validate_json( + (Path(__file__).resolve().parents[1] / "mcp/conformance_baseline.json").read_bytes() + ) + official: Final = tuple( + "tests/integration/mcp/test_mcp_official_conformance.py::test_official_scenario_through_gateway[" + + "-".join(case) + + "]" + for case in product(baseline.official_scenarios, baseline.revisions) + ) + matrix: Final = tuple( + "tests/integration/mcp/test_mcp_transports.py::test_pinned_revision_pairs_list_and_call_through_gateway[" + + "-".join(case) + + "]" + for case in product( + baseline.revisions, baseline.revisions, baseline.upstream_transports, baseline.ingress_transports + ) + ) + required: Final = official + matrix + baseline.required_tests + assert len(set(required)) == len(required), "Conformance baseline contains duplicate required cases" + require_capability_coverage(baseline, required, REVISION_SUPPORT) + return required + + +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" diff --git a/tests/integration/_support/mcp.py b/tests/integration/_support/mcp.py index a3693433de4..996bd054d7f 100644 --- a/tests/integration/_support/mcp.py +++ b/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 @@ -43,13 +44,17 @@ class McpPeer: spec_path: Path | None = None consumed: list[int] = field(default_factory=lambda: [0]) - def drain(self) -> tuple[dict[str, object], ...]: + def drain(self, *, include_negotiation: bool = False) -> tuple[dict[str, object], ...]: if self.record is not None: lines: Final = self.record.read_text().splitlines() if self.record.exists() else [] fresh: Final = tuple(json.loads(line) for line in lines[self.consumed[0] :]) self.consumed[0] = len(lines) - return fresh - return tuple(self.calls.get_nowait() for _ in range(self.calls.qsize())) + return tuple(item for item in fresh if include_negotiation or "negotiation" not in item) + return tuple( + item + for item in (self.calls.get_nowait() for _ in range(self.calls.qsize())) + if include_negotiation or "negotiation" not in item + ) def registration(self) -> dict[str, object]: if self.transport == "stdio": @@ -63,8 +68,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": {}, "headers": {}, "negotiation": {"requested": requested, "returned": returned}}) + return result + + service.middleware.append(capture_negotiation) service.add_tool(add) service.add_tool(multiply) @@ -184,14 +204,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) @@ -199,12 +219,26 @@ def mcp_peer(transport: Literal["http", "sse"] = "http", *, rich: bool = False) @contextmanager def stdio_peer(directory: Path, *, rich: bool = False) -> Iterator[McpPeer]: record: Final = directory / f"stdio-{os.getpid()}-{time.monotonic_ns()}.jsonl" + coverage_file: Final = os.environ.get("COVERAGE_FILE") + coverage_args: Final = ( + ( + "-c", + "import mcp.server.mcpserver, runpy, sys; " + "sys.argv = ['coverage', *sys.argv[1:]]; " + "runpy.run_module('coverage', run_name='__main__')", + "run", + f"--rcfile={STDIO_PEER.parent.parent / 'conformance_coverage.toml'}", + f"--data-file={Path(coverage_file).resolve()}", + ) + if coverage_file and os.environ.get("COVERAGE_PROCESS_CONFIG") + else () + ) yield McpPeer( "", queue.Queue(), "stdio", sys.executable, - (str(STDIO_PEER), str(record), "rich" if rich else "plain"), + (*coverage_args, str(STDIO_PEER), str(record), "rich" if rich else "plain"), record, ) diff --git a/tests/integration/_support/mcp_stdio_peer.py b/tests/integration/_support/mcp_stdio_peer.py index 59865f9dc74..2dfe2f94421 100644 --- a/tests/integration/_support/mcp_stdio_peer.py +++ b/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 diff --git a/tests/integration/conformance_coverage.toml b/tests/integration/conformance_coverage.toml index ca1c2724341..49bd2df9c72 100644 --- a/tests/integration/conformance_coverage.toml +++ b/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 diff --git a/tests/integration/mcp/conformance_baseline.json b/tests/integration/mcp/conformance_baseline.json new file mode 100644 index 00000000000..8a3def5ddcb --- /dev/null +++ b/tests/integration/mcp/conformance_baseline.json @@ -0,0 +1,72 @@ +{ + "revisions": [ + "2024-11-05", + "2025-03-26", + "2025-06-18", + "2025-11-25" + ], + "ingress_transports": [ + "http", + "sse" + ], + "upstream_transports": [ + "http", + "sse", + "stdio" + ], + "official_scenarios": [ + "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" + ], + "capabilities": [ + { + "revisions": [ + "2024-11-05", + "2025-03-26", + "2025-06-18", + "2025-11-25" + ], + "operations": [ + "tools/list", + "tools/call", + "prompts/list", + "prompts/get", + "resources/list", + "resources/read", + "resources/templates/list" + ], + "extensions": [], + "tests": [ + "tests/integration/mcp/test_mcp_transports.py::test_pinned_revision_pairs_list_and_call_through_gateway" + ] + } + ], + "required_tests": [ + "tests/integration/mcp/test_mcp_protocol_errors.py::test_omitted_tool_arguments_reach_the_upstream", + "tests/integration/mcp/test_mcp_protocol_errors.py::test_configured_origin_policy_rejects_before_tool_execution[server_mcp]", + "tests/integration/mcp/test_mcp_protocol_errors.py::test_configured_origin_policy_rejects_before_tool_execution[sse]", + "tests/integration/mcp/test_mcp_official_conformance.py::test_conformance_bridge_preserves_headers_payload_and_error_status[200]", + "tests/integration/mcp/test_mcp_official_conformance.py::test_conformance_bridge_preserves_headers_payload_and_error_status[403]", + "tests/integration/mcp/test_mcp_official_conformance.py::test_official_runner_rejects_unknown_scenario", + "tests/integration/mcp/test_mcp_official_conformance.py::test_official_gateway_session_lifecycle", + "tests/integration/mcp/test_mcp_official_conformance.py::test_stalled_reference_is_killed_and_cannot_report_clean_teardown", + "tests/integration/mcp/test_mcp_official_conformance.py::test_reference_children_are_stopped_after_the_root_exits" + ] +} diff --git a/tests/integration/mcp/test_mcp_management.py b/tests/integration/mcp/test_mcp_management.py index 66c30a62bde..71e79979919 100644 --- a/tests/integration/mcp/test_mcp_management.py +++ b/tests/integration/mcp/test_mcp_management.py @@ -18,7 +18,7 @@ from integration._support.mcp import ( from integration._support.process import owned_proxy from litellm.models.user import LiteLLM_UserTable -from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken +from litellm.proxy.proxy_server import ExperimentalUIJWTToken ADD: Final = {"a": 4, "b": 5} diff --git a/tests/integration/mcp/test_mcp_official_conformance.py b/tests/integration/mcp/test_mcp_official_conformance.py index 5e585f65136..9ae09f8b753 100644 --- a/tests/integration/mcp/test_mcp_official_conformance.py +++ b/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", diff --git a/tests/integration/mcp/test_mcp_protocol_errors.py b/tests/integration/mcp/test_mcp_protocol_errors.py index fbce03cf60b..503b298259f 100644 --- a/tests/integration/mcp/test_mcp_protocol_errors.py +++ b/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 diff --git a/tests/integration/mcp/test_mcp_transports.py b/tests/integration/mcp/test_mcp_transports.py index 16004cdd501..8d2ddcc1e73 100644 --- a/tests/integration/mcp/test_mcp_transports.py +++ b/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()) - 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 + negotiated: Final = asyncio.run(exercise()) + observed: Final = peer.drain(include_negotiation=True) + 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" + ) diff --git a/tests/integration/run.py b/tests/integration/run.py index 87f93873267..e1e5207669a 100644 --- a/tests/integration/run.py +++ b/tests/integration/run.py @@ -92,6 +92,14 @@ 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 + + required: Final = tuple( + node for node in required_conformance_nodes() if not options.files or node.split("::")[0] in selected + ) + if required: + require_passes(output, required) return 0 diff --git a/tests/unit/integration_support/test_conformance.py b/tests/unit/integration_support/test_conformance.py index 8f0acda74ca..791ef5e04cf 100644 --- a/tests/unit/integration_support/test_conformance.py +++ b/tests/unit/integration_support/test_conformance.py @@ -1,15 +1,132 @@ import hashlib import json import os +import queue from pathlib import Path 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 +@pytest.mark.asyncio +async def test_negotiation_records_preserve_the_peer_call_contract(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2])) + from integration._support.mcp import math_service + from integration.mcp.test_mcp_user_env_vars import UPSTREAM_CALLS + from mcp.server.context import ServerRequestContext + from mcp.types import InitializeResult + + records: Final[list[dict[str, object]]] = [] + service: Final = math_service(record=records.append) + context: Final = ServerRequestContext( + session=Mock(), + lifespan_context=None, + protocol_version="2025-03-26", + method="initialize", + params={"protocolVersion": "2025-03-26", "capabilities": {}, "clientInfo": {"name": "client", "version": "1"}}, + ) + + async def initialize(request: ServerRequestContext[object, object]) -> InitializeResult: + return InitializeResult.model_validate( + { + "protocolVersion": request.protocol_version, + "capabilities": {}, + "serverInfo": {"name": "peer", "version": "1"}, + } + ) + + await service.middleware[-1](context, initialize) + calls: Final = UPSTREAM_CALLS.validate_python(tuple(records)) + assert len(calls) == 1 and calls[0].headers == {} and calls[0].body == {} + assert records[0]["negotiation"] == {"requested": "2025-03-26", "returned": "2025-03-26"} + + +@pytest.mark.parametrize("covered", (False, True)) +@pytest.mark.parametrize("has_data_file", (False, True)) +def test_stdio_coverage_preserves_default_environment( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, covered: bool, has_data_file: bool +) -> None: + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2])) + from integration._support.mcp import STDIO_PEER, stdio_peer + + monkeypatch.delenv("COVERAGE_PROCESS_CONFIG", raising=False) + monkeypatch.delenv("COVERAGE_FILE", raising=False) + monkeypatch.setenv("OPENAI_API_KEY", "must-not-reach-peer") + coverage_file: Final = tmp_path / "coverage data" + if has_data_file: + monkeypatch.setenv("COVERAGE_FILE", str(coverage_file)) + if covered: + monkeypatch.setenv("COVERAGE_PROCESS_CONFIG", "coverage-config") + with stdio_peer(tmp_path) as peer: + assert peer.registration() == { + "transport": "stdio", + "command": peer.command, + "args": [ + *( + [ + "-c", + "import mcp.server.mcpserver, runpy, sys; " + "sys.argv = ['coverage', *sys.argv[1:]]; " + "runpy.run_module('coverage', run_name='__main__')", + "run", + f"--rcfile={STDIO_PEER.parent.parent / 'conformance_coverage.toml'}", + f"--data-file={coverage_file}", + ] + if covered and has_data_file + else [] + ), + str(STDIO_PEER), + str(peer.record), + "plain", + ], + } + + +@pytest.mark.parametrize("file_backed", (False, True)) +def test_peer_drain_preserves_received_requests( + tmp_path: Path, file_backed: bool, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2])) + from integration._support.mcp import McpPeer + + request: Final = {"body": {"method": "tools/call"}, "headers": {"authorization": "synthetic"}} + negotiation: Final = { + "body": {}, + "headers": {}, + "negotiation": {"requested": "2025-03-26", "returned": "2025-03-26"}, + } + records: Final = (request, negotiation) + observed: Final[queue.Queue[dict[str, object]]] = queue.Queue() + path: Final = tmp_path / "peer.jsonl" if file_backed else None + if path is not None: + path.write_text("".join(json.dumps(record) + "\n" for record in records)) + else: + for record in records: + observed.put(record) + peer: Final = McpPeer("http://peer", observed, record=path) + assert peer.drain() == (request,) + assert peer.drain() == () + if path is not None: + with path.open("a") as sink: + sink.write("".join(json.dumps(record) + "\n" for record in records)) + else: + for record in records: + observed.put(record) + assert peer.drain(include_negotiation=True) == records + assert peer.drain(include_negotiation=True) == () + + @pytest.mark.parametrize("explicit", (False, True)) def test_owned_proxy_isolates_automatic_coverage_unless_requested( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, explicit: bool @@ -37,6 +154,42 @@ def test_owned_proxy_isolates_automatic_coverage_unless_requested( assert os.environ["COVERAGE_PROCESS_CONFIG"] == "parent-config" +@pytest.mark.parametrize("selected", ("test_mcp_transports.py", "test_mcp_credentials.py", "")) +@pytest.mark.parametrize("missing_required", (False, True)) +def test_runner_requires_conformance_for_selected_files( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, selected: str, missing_required: bool +) -> None: + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2])) + from integration import run + from integration._support.conformance import required_conformance_nodes + + files: Final = tuple( + str(path.relative_to(Path.cwd())) + for path in sorted((Path.cwd() / "tests/integration/mcp").glob("test_*.py")) + if not selected or path.name == selected + ) + required: Final = tuple(node for node in required_conformance_nodes() if node.split("::")[0] in files) + collected: Final = required + tuple(f"{path}::other" for path in files) + omitted: Final = required[:1] if missing_required else () + (tmp_path / "execution.json").write_text( + json.dumps( + { + "collected": collected, + "passed": tuple(n for n in collected if n not in omitted), + "skipped": omitted, + "complete": True, + } + ) + ) + monkeypatch.setattr("sys.argv", ["run.py", "mcp", "--results", str(tmp_path), *(files if selected else ())]) + with patch("integration.run.subprocess.call", return_value=0): + if omitted: + with pytest.raises(AssertionError, match="conformance"): + run.main() + else: + assert run.main() == 0 + + def test_changed_archive_is_rejected(tmp_path: Path) -> None: archive: Final = tmp_path / "reference.tar.gz" archive.write_bytes(b"changed reference") @@ -161,6 +314,122 @@ 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"), + ), + ), +) +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", ( @@ -181,3 +450,139 @@ def test_missing_secondary_checks_cannot_report_complete_conformance( read_checks(tmp_path, scenario) report.write_text(json.dumps([{"id": identity, "status": "SUCCESS"} for identity in identities])) assert len(read_checks(tmp_path, scenario)) == len(identities) + + +@pytest.mark.parametrize("removed", ("tools-list", "tools-call-with-progress")) +def test_removing_official_scenario_cannot_shrink_required_baseline( + monkeypatch: pytest.MonkeyPatch, removed: str +) -> None: + from tests.integration._support import conformance + + expected: Final = conformance.required_conformance_nodes() + monkeypatch.setattr( + conformance, "OFFICIAL_SCENARIOS", tuple(s for s in conformance.OFFICIAL_SCENARIOS if s != removed) + ) + assert conformance.required_conformance_nodes() == expected + + +def test_removing_transport_cannot_shrink_required_baseline(monkeypatch: pytest.MonkeyPatch) -> None: + from dataclasses import replace + + from tests.integration._support.conformance import required_conformance_nodes + from litellm.proxy._experimental.mcp_server import capabilities + from litellm.types.mcp import MCPTransport + + expected: Final = required_conformance_nodes() + monkeypatch.setattr( + capabilities, + "REVISION_SUPPORT", + { + revision: replace(support, transports=support.transports - {MCPTransport.stdio}) + for revision, support in capabilities.REVISION_SUPPORT.items() + }, + ) + assert required_conformance_nodes() == expected + + +@pytest.mark.parametrize("kind", ("operations", "extensions")) +def test_new_completed_capability_requires_mapped_tests(monkeypatch: pytest.MonkeyPatch, kind: str) -> None: + from dataclasses import replace + + from tests.integration._support.conformance import required_conformance_nodes + from litellm.proxy._experimental.mcp_server import capabilities + + support: Final = capabilities.REVISION_SUPPORT["2025-11-25"] + expanded: Final = ( + replace(support, operations=support.operations | {"resources/subscribe"}) + if kind == "operations" + else replace(support, extensions=support.extensions | {"tasks"}) + ) + monkeypatch.setattr(capabilities, "REVISION_SUPPORT", {**capabilities.REVISION_SUPPORT, "2025-11-25": expanded}) + with pytest.raises(AssertionError, match="capability"): + required_conformance_nodes() + + +@pytest.mark.parametrize("missing", ("scenario", "file", "worker")) +def test_required_execution_cannot_omit_part_of_baseline(tmp_path: Path, missing: str) -> None: + from tests.integration._support.conformance import required_conformance_nodes + + required: Final = required_conformance_nodes() + omitted: Final = ( + required[:1] + if missing == "scenario" + else tuple(node for node in required if "test_mcp_transports.py" in node) + if missing == "file" + else required[::4] + ) + remaining: Final = tuple(node for node in required if node not in omitted) + (tmp_path / "execution.json").write_text( + json.dumps({"collected": remaining, "passed": remaining, "skipped": [], "complete": True}) + ) + with pytest.raises(AssertionError, match="not collected"): + require_passes(tmp_path, required) + + +@pytest.mark.parametrize("invalid", ("revision", "test")) +def test_capability_mapping_must_reference_required_coverage(invalid: str) -> None: + from tests.integration._support.conformance import ( + CapabilityContract, + ConformanceBaseline, + require_capability_coverage, + ) + + baseline: Final = ConformanceBaseline( + revisions=("2025-11-25",), + ingress_transports=("http",), + upstream_transports=("http",), + official_scenarios=("tools-list",), + required_tests=("test_tools",), + capabilities=( + CapabilityContract( + revisions=("unknown" if invalid == "revision" else "2025-11-25",), + operations=("tools/list",), + extensions=(), + tests=("missing" if invalid == "test" else "test_tools",), + ), + ), + ) + with pytest.raises(AssertionError, match="capability maps"): + require_capability_coverage(baseline, ("test_tools[http]",), {}) + + +def test_capability_can_activate_after_its_required_contract_is_added() -> None: + from dataclasses import replace + + from tests.integration._support.conformance import ( + CapabilityContract, + ConformanceBaseline, + require_capability_coverage, + ) + from litellm.proxy._experimental.mcp_server.capabilities import REVISION_SUPPORT + + baseline: Final = ConformanceBaseline( + revisions=("candidate",), + ingress_transports=("http",), + upstream_transports=("http",), + official_scenarios=("tools-list",), + required_tests=("test_subscription",), + capabilities=( + CapabilityContract( + revisions=("candidate",), + operations=("resources/subscribe",), + extensions=("subscriptions",), + tests=("test_subscription",), + ), + ), + ) + support: Final = replace( + REVISION_SUPPORT["2025-11-25"], + operations=frozenset({"resources/subscribe"}), + extensions=frozenset({"subscriptions"}), + ) + assert require_capability_coverage(baseline, ("test_subscription",), {"candidate": support}) is None + assert ( + require_capability_coverage(baseline, ("test_subscription",), {"future": replace(support, completed=False)}) + is None + ) + with pytest.raises(AssertionError, match="coverage missing for revision"): + require_capability_coverage(baseline, ("test_subscription",), {"future": support})