Compare protocol versions through a known-version registry

Protocol revisions are an enumerated set, not an ordered scalar: future
identifiers are not guaranteed to be date-shaped, and unrecognized peer
strings must not accidentally compare as newer ("zzz" sorts after every
date). Add KNOWN_PROTOCOL_VERSIONS plus is_version_at_least and
is_stateful_protocol_version helpers, and replace the lexicographic
string comparisons in OAuth resource-param gating and streamable HTTP
priming/close-callback gating. Unknown versions now gate conservatively
(treated as older than every known revision).

Author: maxisbey

src/mcp/client/auth/oauth2.py

@@ -49,6 +49,7 @@
     check_resource_allowed,
     resource_url_from_server_url,
 )
+from mcp.shared.version import is_version_at_least
 
 logger = logging.getLogger(__name__)
 
@@ -172,9 +173,7 @@ def should_include_resource_param(self, protocol_version: str | None = None) ->
         if not protocol_version:
             return False
 
-        # Check if protocol version is 2025-06-18 or later
-        # Version format is YYYY-MM-DD, so string comparison works
-        return protocol_version >= "2025-06-18"
+        return is_version_at_least(protocol_version, "2025-06-18")
 
     def prepare_token_auth(
         self, data: dict[str, str], headers: dict[str, str] | None = None

src/mcp/server/streamable_http.py

@@ -28,7 +28,7 @@
 from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
-from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
+from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS, is_version_at_least
 from mcp.types import (
     DEFAULT_NEGOTIATED_VERSION,
     INTERNAL_ERROR,
@@ -238,7 +238,7 @@ def _create_session_message(
         the stream is closed early because they didn't receive a priming event.
         """
         # Only provide close callbacks when client supports resumability
-        if self._event_store and protocol_version >= "2025-11-25":
+        if self._event_store and is_version_at_least(protocol_version, "2025-11-25"):
 
             async def close_stream_callback() -> None:
                 self.close_sse_stream(request_id)
@@ -271,7 +271,7 @@ async def _maybe_send_priming_event(
         if not self._event_store:
             return
         # Priming events have empty data which older clients cannot handle.
-        if protocol_version < "2025-11-25":
+        if not is_version_at_least(protocol_version, "2025-11-25"):
             return
         priming_event_id = await self._event_store.store_event(
             str(request_id),  # Convert RequestId to StreamId (str)

src/mcp/shared/version.py

@@ -1,3 +1,59 @@
+"""Protocol-version registry and comparison/classification helpers.
+
+Date-string protocol revisions happen to sort lexicographically, but versions
+are an enumerated set, not an ordered scalar: future identifiers are not
+guaranteed to be date-shaped, and unrecognized peer strings must compare
+conservatively instead of accidentally (e.g. "zzz" > "2025-11-25"). All
+ordering questions go through KNOWN_PROTOCOL_VERSIONS.
+"""
+
+from typing import Final
+
 from mcp.types import LATEST_PROTOCOL_VERSION
 
+KNOWN_PROTOCOL_VERSIONS: Final[tuple[str, ...]] = (
+    "2024-11-05",
+    "2025-03-26",
+    "2025-06-18",
+    "2025-11-25",
+    "2026-07-28",  # draft: recognized for ordering/classification, NOT negotiable
+)
+"""Every protocol revision this SDK knows about, oldest to newest."""
+
+DRAFT_PROTOCOL_VERSION: Final[str] = "2026-07-28"
+"""The in-progress spec revision.
+
+Recognized by the helpers in this module but absent from
+SUPPORTED_PROTOCOL_VERSIONS until the SDK actually implements it.
+"""
+
 SUPPORTED_PROTOCOL_VERSIONS: list[str] = ["2024-11-05", "2025-03-26", "2025-06-18", LATEST_PROTOCOL_VERSION]
+"""Protocol revisions this SDK can negotiate."""
+
+STATEFUL_PROTOCOL_VERSIONS: Final[frozenset[str]] = frozenset({"2024-11-05", "2025-03-26", "2025-06-18", "2025-11-25"})
+"""Revisions that negotiate via the initialize handshake.
+
+Closed by design: every revision after 2025-11-25 is stateless and negotiates
+per-request, never via initialize. Hardcoded - do not derive from
+SUPPORTED_PROTOCOL_VERSIONS. (Matches typescript-sdk's
+STATEFUL_PROTOCOL_VERSIONS / isStatefulProtocolVersion.)
+"""
+
+
+def is_stateful_protocol_version(version: str) -> bool:
+    """Return True if `version` negotiates via the initialize handshake."""
+    return version in STATEFUL_PROTOCOL_VERSIONS
+
+
+def is_version_at_least(version: str, minimum: str) -> bool:
+    """Return True if `version` is a known revision at least as new as `minimum`.
+
+    Unknown `version` strings return False (treat unrecognized peers
+    conservatively). `minimum` must be a member of KNOWN_PROTOCOL_VERSIONS;
+    passing anything else is programmer error and raises ValueError.
+    """
+    if minimum not in KNOWN_PROTOCOL_VERSIONS:
+        raise ValueError(f"minimum must be a known protocol version, got {minimum!r}")
+    if version not in KNOWN_PROTOCOL_VERSIONS:
+        return False
+    return KNOWN_PROTOCOL_VERSIONS.index(version) >= KNOWN_PROTOCOL_VERSIONS.index(minimum)

tests/client/test_auth.py

@@ -819,6 +819,25 @@ async def test_resource_param_included_with_protected_resource_metadata(self, oa
         assert "resource=" in content
 
 
+@pytest.mark.parametrize(
+    ("protocol_version", "expected"),
+    [
+        ("2025-03-26", False),
+        ("2025-06-18", True),
+        ("2025-11-25", True),
+        ("2026-07-28", True),
+        # Unrecognized strings gate conservatively, even ones sorting after 2025-06-18.
+        ("zzz", False),
+        ("9999-99-99", False),
+    ],
+)
+def test_should_include_resource_param_by_protocol_version(
+    oauth_provider: OAuthClientProvider, protocol_version: str, expected: bool
+) -> None:
+    """Resource param is included only for recognized versions >= 2025-06-18."""
+    assert oauth_provider.context.should_include_resource_param(protocol_version) is expected
+
+
 @pytest.mark.anyio
 async def test_validate_resource_rejects_mismatched_resource(
     client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage

tests/shared/test_streamable_http.py

@@ -1696,6 +1696,85 @@ async def test_close_sse_stream_callback_not_provided_for_old_protocol_version()
     assert session_msg_new.metadata.close_standalone_sse_stream is not None
 
 
+@pytest.mark.anyio
+async def test_priming_event_not_sent_for_unknown_protocol_version() -> None:
+    """_maybe_send_priming_event treats unrecognized version strings conservatively.
+
+    A garbage version must not be mistaken for a future one (lexicographically
+    "zzz" sorts after every date-shaped revision).
+    """
+    transport = StreamableHTTPServerTransport(
+        "/mcp",
+        event_store=SimpleEventStore(),
+    )
+
+    write_stream, read_stream = anyio.create_memory_object_stream[dict[str, Any]](1)
+
+    try:
+        await transport._maybe_send_priming_event("test-request-id", write_stream, "zzz")
+
+        # Nothing should have been written to the stream
+        assert write_stream.statistics().current_buffer_used == 0
+    finally:
+        await write_stream.aclose()
+        await read_stream.aclose()
+
+
+@pytest.mark.anyio
+async def test_close_sse_stream_callback_not_provided_for_unknown_protocol_version() -> None:
+    """close_sse_stream callbacks are withheld when the client's version is unrecognized."""
+    transport = StreamableHTTPServerTransport(
+        "/mcp",
+        event_store=SimpleEventStore(),
+    )
+
+    mock_message = JSONRPCRequest(jsonrpc="2.0", id="test-1", method="tools/list")
+    mock_request = MagicMock()
+
+    session_msg = transport._create_session_message(mock_message, mock_request, "test-request-id", "zzz")
+
+    assert session_msg.metadata is not None
+    assert isinstance(session_msg.metadata, ServerMessageMetadata)
+    assert session_msg.metadata.close_sse_stream is None
+    assert session_msg.metadata.close_standalone_sse_stream is None
+
+
+@pytest.mark.anyio
+async def test_initialize_with_unknown_protocol_version_gets_no_priming_event(
+    event_app: tuple[SimpleEventStore, Starlette],
+) -> None:
+    """A garbage protocolVersion in initialize params must not trigger priming.
+
+    The priming decision reads the raw body params before any validation, so an
+    unrecognized string must gate conservatively (old-client behavior), not
+    compare lexicographically past "2025-11-25".
+    """
+    event_store, app = event_app
+    init_request = {
+        "jsonrpc": "2.0",
+        "method": "initialize",
+        "params": {
+            "clientInfo": {"name": "test-client", "version": "1.0"},
+            "protocolVersion": "zzz",
+            "capabilities": {},
+        },
+        "id": "init-1",
+    }
+    async with make_client(app) as client:
+        response = await client.post(
+            "/mcp",
+            headers={
+                "Accept": "application/json, text/event-stream",
+                "Content-Type": "application/json",
+            },
+            json=init_request,
+        )
+        assert response.status_code == 200
+
+    # Priming events are stored with a None payload; none may exist for this client.
+    assert all(message is not None for _, _, message in event_store._events)
+
+
 @pytest.mark.anyio
 async def test_streamable_http_client_receives_priming_event(
     event_app: tuple[SimpleEventStore, Starlette],

tests/shared/test_version.py

@@ -0,0 +1,81 @@
+"""Tests for the protocol-version registry and comparison helpers."""
+
+import pytest
+
+from mcp.shared.version import (
+    DRAFT_PROTOCOL_VERSION,
+    KNOWN_PROTOCOL_VERSIONS,
+    STATEFUL_PROTOCOL_VERSIONS,
+    SUPPORTED_PROTOCOL_VERSIONS,
+    is_stateful_protocol_version,
+    is_version_at_least,
+)
+from mcp.types import LATEST_PROTOCOL_VERSION
+
+
+@pytest.mark.parametrize(
+    ("version", "minimum", "expected"),
+    [
+        # equal
+        ("2025-11-25", "2025-11-25", True),
+        ("2024-11-05", "2024-11-05", True),
+        # above
+        ("2025-11-25", "2025-06-18", True),
+        ("2026-07-28", "2024-11-05", True),
+        # below
+        ("2025-06-18", "2025-11-25", False),
+        ("2024-11-05", "2026-07-28", False),
+    ],
+)
+def test_is_version_at_least_ordering(version: str, minimum: str, expected: bool) -> None:
+    assert is_version_at_least(version, minimum) is expected
+
+
+@pytest.mark.parametrize("version", ["zzz", "", "2025-11-26", "draft", "9999-99-99"])
+def test_is_version_at_least_unknown_version_is_false(version: str) -> None:
+    """Unrecognized peer strings compare conservatively, never accidentally."""
+    assert is_version_at_least(version, "2024-11-05") is False
+
+
+def test_is_version_at_least_unknown_minimum_raises() -> None:
+    """An unknown minimum is programmer error, not peer input."""
+    with pytest.raises(ValueError, match="zzz"):
+        is_version_at_least("2025-11-25", "zzz")
+
+
+@pytest.mark.parametrize(
+    ("version", "minimum"), [(v, m) for v in KNOWN_PROTOCOL_VERSIONS for m in KNOWN_PROTOCOL_VERSIONS]
+)
+def test_is_version_at_least_matches_lexicographic_for_known_versions(version: str, minimum: str) -> None:
+    """Drop-in equivalence: for every known (date-shaped) revision pair the helper
+    agrees with the string comparison it replaced."""
+    assert is_version_at_least(version, minimum) is (version >= minimum)
+
+
+def test_draft_version_is_known_but_not_negotiable_and_not_stateful() -> None:
+    assert DRAFT_PROTOCOL_VERSION in KNOWN_PROTOCOL_VERSIONS
+    assert DRAFT_PROTOCOL_VERSION not in SUPPORTED_PROTOCOL_VERSIONS
+    assert not is_stateful_protocol_version(DRAFT_PROTOCOL_VERSION)
+
+
+def test_draft_version_is_at_least_every_released_version() -> None:
+    for released in SUPPORTED_PROTOCOL_VERSIONS:
+        assert is_version_at_least(DRAFT_PROTOCOL_VERSION, released)
+
+
+def test_every_supported_version_is_stateful() -> None:
+    for version in SUPPORTED_PROTOCOL_VERSIONS:
+        assert is_stateful_protocol_version(version)
+
+
+def test_supported_versions_are_a_strict_subset_of_known() -> None:
+    assert set(SUPPORTED_PROTOCOL_VERSIONS) < set(KNOWN_PROTOCOL_VERSIONS)
+
+
+def test_latest_version_is_stateful() -> None:
+    assert LATEST_PROTOCOL_VERSION in STATEFUL_PROTOCOL_VERSIONS
+
+
+def test_known_versions_are_strictly_ordered() -> None:
+    """The registry tuple is the ordering source of truth: ascending, no duplicates."""
+    assert list(KNOWN_PROTOCOL_VERSIONS) == sorted(set(KNOWN_PROTOCOL_VERSIONS))