Add named protocol-version scalars and replace tuple indexing

shared/version.py gains four derived constants alongside the existing tuples:
LATEST_PROTOCOL_VERSION (now derived here instead of a duplicate literal in
types/_types.py), LATEST_HANDSHAKE_VERSION, LATEST_MODERN_VERSION, and
OLDEST_SUPPORTED_VERSION. Call sites that previously wrote
HANDSHAKE_PROTOCOL_VERSIONS[-1] / MODERN_PROTOCOL_VERSIONS[-1] / [0] now
import the named scalar so the meaning is explicit at the use site and a
future version bump is one edit.

This also fixes a quiet drift: a handful of tests were passing
LATEST_PROTOCOL_VERSION (now "2026-07-28") into InitializeRequest on the
legacy handshake path; those now use LATEST_HANDSHAKE_VERSION.

Author: maxisbey

src/mcp/client/session.py

@@ -27,7 +27,12 @@
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
 from mcp.shared.session import RequestResponder
 from mcp.shared.transport_context import TransportContext
-from mcp.shared.version import HANDSHAKE_PROTOCOL_VERSIONS, MODERN_PROTOCOL_VERSIONS
+from mcp.shared.version import (
+    HANDSHAKE_PROTOCOL_VERSIONS,
+    LATEST_HANDSHAKE_VERSION,
+    LATEST_MODERN_VERSION,
+    MODERN_PROTOCOL_VERSIONS,
+)
 from mcp.types import (
     CLIENT_CAPABILITIES_META_KEY,
     CLIENT_INFO_META_KEY,
@@ -333,7 +338,7 @@ async def initialize(self) -> types.InitializeResult:
         result = await self.send_request(
             types.InitializeRequest(
                 params=types.InitializeRequestParams(
-                    protocol_version=HANDSHAKE_PROTOCOL_VERSIONS[-1],
+                    protocol_version=LATEST_HANDSHAKE_VERSION,
                     capabilities=self._build_capabilities(),
                     client_info=self._client_info,
                 ),
@@ -413,7 +418,7 @@ async def probe(version: str) -> dict[str, Any]:
             return await self._dispatcher.send_raw_request("server/discover", params, opts)
 
         try:
-            raw = await probe(MODERN_PROTOCOL_VERSIONS[-1])
+            raw = await probe(LATEST_MODERN_VERSION)
         except MCPError as e:
             if e.code != UNSUPPORTED_PROTOCOL_VERSION:
                 raise

src/mcp/server/connection.py

@@ -31,7 +31,7 @@
 from mcp.shared.dispatcher import CallOptions, Outbound
 from mcp.shared.exceptions import MCPDeprecationWarning, NoBackChannelError
 from mcp.shared.peer import Meta, dump_params
-from mcp.shared.version import HANDSHAKE_PROTOCOL_VERSIONS
+from mcp.shared.version import LATEST_HANDSHAKE_VERSION
 from mcp.types import (
     ClientCapabilities,
     CreateMessageRequest,
@@ -192,14 +192,12 @@ def for_loop(
 
         Not born-ready: `initialized` is set later by the kernel when
         `notifications/initialized` arrives. `protocol_version` is seeded from
-        the transport hint (or `HANDSHAKE_PROTOCOL_VERSIONS[-1]`) so it's never `None`;
+        the transport hint (or `LATEST_HANDSHAKE_VERSION`) so it's never `None`;
         the handshake overwrites it once negotiated.
         """
         return cls(
             outbound,
-            protocol_version=protocol_version_hint
-            if protocol_version_hint is not None
-            else HANDSHAKE_PROTOCOL_VERSIONS[-1],
+            protocol_version=protocol_version_hint if protocol_version_hint is not None else LATEST_HANDSHAKE_VERSION,
             session_id=session_id,
         )
 

src/mcp/shared/version.py

@@ -35,6 +35,18 @@
 Kept as the union for v1.x compatibility.
 """
 
+LATEST_PROTOCOL_VERSION: Final[str] = KNOWN_PROTOCOL_VERSIONS[-1]
+"""Newest protocol revision this SDK speaks (any era)."""
+
+LATEST_HANDSHAKE_VERSION: Final[str] = HANDSHAKE_PROTOCOL_VERSIONS[-1]
+"""Newest revision reachable via the ``initialize`` handshake; the client's offer and server's counter-offer default."""
+
+LATEST_MODERN_VERSION: Final[str] = MODERN_PROTOCOL_VERSIONS[-1]
+"""Newest per-request-envelope revision; the ``server/discover`` probe default."""
+
+OLDEST_SUPPORTED_VERSION: Final[str] = HANDSHAKE_PROTOCOL_VERSIONS[0]
+"""Oldest revision this SDK still negotiates via the ``initialize`` handshake."""
+
 
 def is_version_at_least(version: str, minimum: str) -> bool:
     """Return True if `version` is a known revision at least as new as `minimum`.

src/mcp/types/_types.py

@@ -20,14 +20,9 @@
 from pydantic.alias_generators import to_camel
 from typing_extensions import NotRequired, TypedDict
 
+from mcp.shared.version import LATEST_PROTOCOL_VERSION as LATEST_PROTOCOL_VERSION
 from mcp.types.jsonrpc import RequestId
 
-LATEST_PROTOCOL_VERSION: Final[str] = "2026-07-28"
-"""The newest protocol version this SDK can negotiate.
-
-See https://modelcontextprotocol.io/specification/latest.
-"""
-
 DEFAULT_NEGOTIATED_VERSION: Final[str] = "2025-03-26"
 """The default negotiated version of the Model Context Protocol when no version is specified.
 

tests/client/test_session.py

@@ -18,7 +18,7 @@
 from mcp.shared.message import SessionMessage
 from mcp.shared.session import RequestResponder
 from mcp.shared.transport_context import TransportContext
-from mcp.shared.version import HANDSHAKE_PROTOCOL_VERSIONS
+from mcp.shared.version import HANDSHAKE_PROTOCOL_VERSIONS, LATEST_HANDSHAKE_VERSION
 from mcp.types import (
     CONNECTION_CLOSED,
     INTERNAL_ERROR,
@@ -87,7 +87,7 @@ async def mock_server():
         assert isinstance(request, InitializeRequest)
 
         result = InitializeResult(
-            protocol_version=HANDSHAKE_PROTOCOL_VERSIONS[-1],
+            protocol_version=LATEST_HANDSHAKE_VERSION,
             capabilities=ServerCapabilities(
                 logging=None,
                 resources=None,
@@ -140,7 +140,7 @@ async def message_handler(  # pragma: no cover
 
     # Assert the result
     assert isinstance(result, InitializeResult)
-    assert result.protocol_version == HANDSHAKE_PROTOCOL_VERSIONS[-1]
+    assert result.protocol_version == LATEST_HANDSHAKE_VERSION
     assert isinstance(result.capabilities, ServerCapabilities)
     assert result.server_info == Implementation(name="mock-server", version="0.1.0")
     assert result.instructions == "The server instructions."
@@ -171,7 +171,7 @@ async def mock_server():
         received_client_info = request.params.client_info
 
         result = InitializeResult(
-            protocol_version=HANDSHAKE_PROTOCOL_VERSIONS[-1],
+            protocol_version=LATEST_HANDSHAKE_VERSION,
             capabilities=ServerCapabilities(),
             server_info=Implementation(name="mock-server", version="0.1.0"),
         )
@@ -228,7 +228,7 @@ async def mock_server():
         received_client_info = request.params.client_info
 
         result = InitializeResult(
-            protocol_version=HANDSHAKE_PROTOCOL_VERSIONS[-1],
+            protocol_version=LATEST_HANDSHAKE_VERSION,
             capabilities=ServerCapabilities(),
             server_info=Implementation(name="mock-server", version="0.1.0"),
         )
@@ -278,7 +278,7 @@ async def mock_server():
         assert isinstance(request, InitializeRequest)
 
         # Verify client offers the newest handshake protocol version
-        assert request.params.protocol_version == HANDSHAKE_PROTOCOL_VERSIONS[-1]
+        assert request.params.protocol_version == LATEST_HANDSHAKE_VERSION
 
         # Server responds with a supported older version
         result = InitializeResult(
@@ -386,7 +386,7 @@ async def mock_server():
         received_capabilities = request.params.capabilities
 
         result = InitializeResult(
-            protocol_version=HANDSHAKE_PROTOCOL_VERSIONS[-1],
+            protocol_version=LATEST_HANDSHAKE_VERSION,
             capabilities=ServerCapabilities(),
             server_info=Implementation(name="mock-server", version="0.1.0"),
         )
@@ -457,7 +457,7 @@ async def mock_server():
         received_capabilities = request.params.capabilities
 
         result = InitializeResult(
-            protocol_version=HANDSHAKE_PROTOCOL_VERSIONS[-1],
+            protocol_version=LATEST_HANDSHAKE_VERSION,
             capabilities=ServerCapabilities(),
             server_info=Implementation(name="mock-server", version="0.1.0"),
         )
@@ -536,7 +536,7 @@ async def mock_server():
         received_capabilities = request.params.capabilities
 
         result = InitializeResult(
-            protocol_version=HANDSHAKE_PROTOCOL_VERSIONS[-1],
+            protocol_version=LATEST_HANDSHAKE_VERSION,
             capabilities=ServerCapabilities(),
             server_info=Implementation(name="mock-server", version="0.1.0"),
         )
@@ -604,7 +604,7 @@ async def mock_server():
         assert isinstance(request, InitializeRequest)
 
         result = InitializeResult(
-            protocol_version=HANDSHAKE_PROTOCOL_VERSIONS[-1],
+            protocol_version=LATEST_HANDSHAKE_VERSION,
             capabilities=expected_capabilities,
             server_info=expected_server_info,
             instructions=expected_instructions,
@@ -643,12 +643,12 @@ async def mock_server():
         assert result.server_info == expected_server_info
         assert result.capabilities == expected_capabilities
         assert result.instructions == expected_instructions
-        assert result.protocol_version == HANDSHAKE_PROTOCOL_VERSIONS[-1]
+        assert result.protocol_version == LATEST_HANDSHAKE_VERSION
         # Era-neutral accessors are populated from the InitializeResult.
         assert session.server_info == expected_server_info
         assert session.server_capabilities == expected_capabilities
         assert session.instructions == expected_instructions
-        assert session.protocol_version == HANDSHAKE_PROTOCOL_VERSIONS[-1]
+        assert session.protocol_version == LATEST_HANDSHAKE_VERSION
 
 
 @pytest.mark.anyio
@@ -671,7 +671,7 @@ async def mock_server():
         assert isinstance(request, InitializeRequest)
 
         result = InitializeResult(
-            protocol_version=HANDSHAKE_PROTOCOL_VERSIONS[-1],
+            protocol_version=LATEST_HANDSHAKE_VERSION,
             capabilities=ServerCapabilities(),
             server_info=Implementation(name="mock-server", version="0.1.0"),
         )
@@ -1354,7 +1354,7 @@ async def send_raw_request(
             self.calls.append((method, opts or {}))
             if method == "initialize":
                 return InitializeResult(
-                    protocol_version=HANDSHAKE_PROTOCOL_VERSIONS[-1],
+                    protocol_version=LATEST_HANDSHAKE_VERSION,
                     capabilities=ServerCapabilities(),
                     server_info=Implementation(name="mock-server", version="0.1.0"),
                 ).model_dump(by_alias=True, mode="json", exclude_none=True)

tests/interaction/_connect.py

@@ -31,7 +31,7 @@
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
-from mcp.shared.version import HANDSHAKE_PROTOCOL_VERSIONS, MODERN_PROTOCOL_VERSIONS
+from mcp.shared.version import LATEST_HANDSHAKE_VERSION, MODERN_PROTOCOL_VERSIONS
 from mcp.types import (
     ClientCapabilities,
     Implementation,
@@ -70,7 +70,7 @@ def __call__(
         message_handler: MessageHandlerFnT | None = None,
         client_info: Implementation | None = None,
         elicitation_callback: ElicitationFnT | None = None,
-        spec_version: str = HANDSHAKE_PROTOCOL_VERSIONS[-1],
+        spec_version: str = LATEST_HANDSHAKE_VERSION,
     ) -> AbstractAsyncContextManager[Client]: ...
 
 
@@ -85,7 +85,7 @@ async def connect_in_memory(
     message_handler: MessageHandlerFnT | None = None,
     client_info: Implementation | None = None,
     elicitation_callback: ElicitationFnT | None = None,
-    spec_version: str = HANDSHAKE_PROTOCOL_VERSIONS[-1],
+    spec_version: str = LATEST_HANDSHAKE_VERSION,
 ) -> AsyncIterator[Client]:
     """Yield a Client connected to the server over the in-memory transport.
 
@@ -122,7 +122,7 @@ async def connect_over_streamable_http(
     message_handler: MessageHandlerFnT | None = None,
     client_info: Implementation | None = None,
     elicitation_callback: ElicitationFnT | None = None,
-    spec_version: str = HANDSHAKE_PROTOCOL_VERSIONS[-1],
+    spec_version: str = LATEST_HANDSHAKE_VERSION,
 ) -> AsyncIterator[Client]:
     """Yield a Client connected to the server's streamable HTTP app, entirely in process.
 
@@ -276,7 +276,7 @@ def base_headers(*, session_id: str | None = None) -> dict[str, str]:
     headers = {
         "accept": "application/json, text/event-stream",
         "content-type": "application/json",
-        "mcp-protocol-version": HANDSHAKE_PROTOCOL_VERSIONS[-1],
+        "mcp-protocol-version": LATEST_HANDSHAKE_VERSION,
     }
     if session_id is not None:
         headers["mcp-session-id"] = session_id
@@ -286,7 +286,7 @@ def base_headers(*, session_id: str | None = None) -> dict[str, str]:
 def initialize_body(request_id: int = 1) -> dict[str, object]:
     """A wire-level initialize JSON-RPC request body, exactly as an SDK client would send it."""
     params = InitializeRequestParams(
-        protocol_version=HANDSHAKE_PROTOCOL_VERSIONS[-1],
+        protocol_version=LATEST_HANDSHAKE_VERSION,
         capabilities=ClientCapabilities(),
         client_info=Implementation(name="raw", version="0.0.0"),
     )
@@ -354,7 +354,7 @@ async def connect_over_sse(
     message_handler: MessageHandlerFnT | None = None,
     client_info: Implementation | None = None,
     elicitation_callback: ElicitationFnT | None = None,
-    spec_version: str = HANDSHAKE_PROTOCOL_VERSIONS[-1],
+    spec_version: str = LATEST_HANDSHAKE_VERSION,
 ) -> AsyncIterator[Client]:
     """Yield a Client connected to the server's legacy SSE transport, entirely in process."""
     app, _ = build_sse_app(server)

tests/interaction/transports/test_hosting_resume.py

@@ -21,7 +21,7 @@
 from mcp.client.streamable_http import streamable_http_client
 from mcp.server.mcpserver import Context, MCPServer
 from mcp.shared.message import ClientMessageMetadata
-from mcp.shared.version import HANDSHAKE_PROTOCOL_VERSIONS
+from mcp.shared.version import LATEST_HANDSHAKE_VERSION
 from mcp.types import (
     CallToolRequest,
     CallToolRequestParams,
@@ -431,7 +431,7 @@ async def collect(params: LoggingMessageNotificationParams) -> None:
                 # The session id is only observable via the manager (the client transport does not expose it).
                 (session_id,) = manager._server_instances
                 http.headers["mcp-session-id"] = session_id
-                http.headers["mcp-protocol-version"] = HANDSHAKE_PROTOCOL_VERSIONS[-1]
+                http.headers["mcp-protocol-version"] = LATEST_HANDSHAKE_VERSION
                 tg.cancel_scope.cancel()
 
         with anyio.fail_after(5):  # pragma: no branch

tests/issues/test_192_request_id.py

@@ -4,8 +4,8 @@
 from mcp.server.lowlevel import NotificationOptions, Server
 from mcp.server.models import InitializationOptions
 from mcp.shared.message import SessionMessage
+from mcp.shared.version import LATEST_HANDSHAKE_VERSION
 from mcp.types import (
-    LATEST_PROTOCOL_VERSION,
     ClientCapabilities,
     Implementation,
     InitializeRequestParams,
@@ -59,7 +59,7 @@ async def run_server():
             id="init-1",
             method="initialize",
             params=InitializeRequestParams(
-                protocol_version=LATEST_PROTOCOL_VERSION,
+                protocol_version=LATEST_HANDSHAKE_VERSION,
                 capabilities=ClientCapabilities(),
                 client_info=Implementation(name="test-client", version="1.0.0"),
             ).model_dump(by_alias=True, exclude_none=True),

tests/issues/test_552_windows_hang.py

@@ -9,7 +9,7 @@
 
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.stdio import stdio_client
-from mcp.shared.version import HANDSHAKE_PROTOCOL_VERSIONS
+from mcp.shared.version import LATEST_HANDSHAKE_VERSION
 from mcp.types import InitializeResult
 
 
@@ -33,7 +33,7 @@ async def test_initialize_succeeds_and_shutdown_returns_after_the_server_exits_m
             "jsonrpc": "2.0",
             "id": request["id"],
             "result": {{
-                "protocolVersion": {json.dumps(HANDSHAKE_PROTOCOL_VERSIONS[-1])},
+                "protocolVersion": {json.dumps(LATEST_HANDSHAKE_VERSION)},
                 "capabilities": {{}},
                 "serverInfo": {{"name": "test-server", "version": "1.0"}}
             }}

tests/server/test_cancel_handling.py

@@ -7,8 +7,8 @@
 from mcp.server import Server, ServerRequestContext
 from mcp.shared.exceptions import MCPError
 from mcp.shared.message import SessionMessage
+from mcp.shared.version import LATEST_HANDSHAKE_VERSION
 from mcp.types import (
-    LATEST_PROTOCOL_VERSION,
     CallToolRequest,
     CallToolRequestParams,
     CallToolResult,
@@ -138,7 +138,7 @@ async def run_server():
         id=1,
         method="initialize",
         params=InitializeRequestParams(
-            protocol_version=LATEST_PROTOCOL_VERSION,
+            protocol_version=LATEST_HANDSHAKE_VERSION,
             capabilities=ClientCapabilities(),
             client_info=Implementation(name="test", version="1.0"),
         ).model_dump(by_alias=True, mode="json", exclude_none=True),
@@ -212,7 +212,7 @@ async def run_server():
         id=1,
         method="initialize",
         params=InitializeRequestParams(
-            protocol_version=LATEST_PROTOCOL_VERSION,
+            protocol_version=LATEST_HANDSHAKE_VERSION,
             capabilities=ClientCapabilities(),
             client_info=Implementation(name="test", version="1.0"),
         ).model_dump(by_alias=True, mode="json", exclude_none=True),