Expose the negotiated protocol version on ServerSession

Handlers had no way to read which protocol revision the initialize
handshake settled on. Add ServerSession.protocol_version, backed by the
connection state: None before initialization and always None on
stateless connections, where no handshake reaches the session and the
per-request version lives in the MCP-Protocol-Version header.

Author: maxisbey

src/mcp/server/connection.py

@@ -83,6 +83,9 @@ class Connection:
     """The full `initialize` request params; `None` before initialization."""
 
     protocol_version: str | None
+    """The protocol version negotiated during `initialize`; `None` before
+    initialization, and always `None` on stateless connections (no handshake
+    reaches them). Handlers read this as `ServerSession.protocol_version`."""
 
     initialized: anyio.Event
     """Set when `notifications/initialized` arrives (matches TS `oninitialized`);

src/mcp/server/session.py

@@ -50,6 +50,17 @@ def client_params(self) -> types.InitializeRequestParams | None:
         """The client's `initialize` request params; `None` before initialization."""
         return self._connection.client_params
 
+    @property
+    def protocol_version(self) -> str | None:
+        """The protocol version negotiated during `initialize`.
+
+        `None` before initialization completes, and always `None` on stateless
+        connections (no handshake reaches them; on streamable HTTP the
+        per-request version is the `MCP-Protocol-Version` header, available
+        via `ctx.request.headers`).
+        """
+        return self._connection.protocol_version
+
     async def send_request(
         self,
         request: types.ServerRequest,

tests/client/test_client.py

@@ -113,6 +113,12 @@ async def test_client_is_initialized(app: MCPServer):
         assert client.initialize_result.server_info.name == "test"
 
 
+async def test_client_initialize_result_exposes_negotiated_protocol_version(app: MCPServer):
+    """The negotiated protocol version is readable after initialization."""
+    async with Client(app) as client:
+        assert client.initialize_result.protocol_version == types.LATEST_PROTOCOL_VERSION
+
+
 async def test_client_with_simple_server(simple_server: Server):
     """Test that from_server works with a basic Server instance."""
     async with Client(simple_server) as client:

tests/server/test_session.py

@@ -11,6 +11,7 @@
 import pytest
 
 from mcp import types
+from mcp.server import Server, ServerRequestContext
 from mcp.server.connection import Connection
 from mcp.server.session import ServerSession
 from mcp.shared.dispatcher import CallOptions
@@ -26,6 +27,8 @@
     SamplingToolsCapability,
 )
 
+from .test_runner import connected_runner
+
 
 class StubDispatcher:
     """Records `send_raw_request` / `notify` calls and returns a canned result."""
@@ -158,3 +161,63 @@ def test_check_client_capability_delegates_to_connection():
     session = _make_session(dispatcher, capabilities=ClientCapabilities(sampling=SamplingCapability()))
     assert session.check_client_capability(ClientCapabilities(sampling=SamplingCapability())) is True
     assert session.check_client_capability(ClientCapabilities(experimental={"x": {}})) is False
+
+
+def _runner_server(seen_versions: list[str | None]) -> Server[dict[str, Any]]:
+    """A lowlevel Server whose tools/list handler records `ctx.session.protocol_version`."""
+
+    async def list_tools(
+        ctx: ServerRequestContext[dict[str, Any], Any], params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        seen_versions.append(ctx.session.protocol_version)
+        return types.ListToolsResult(tools=[])
+
+    return Server(name="test-server", version="0.0.1", on_list_tools=list_tools)
+
+
+def _init_params(protocol_version: str) -> dict[str, Any]:
+    return InitializeRequestParams(
+        protocol_version=protocol_version,
+        capabilities=ClientCapabilities(),
+        client_info=Implementation(name="test-client", version="1.0"),
+    ).model_dump(by_alias=True, exclude_none=True)
+
+
+@pytest.mark.anyio
+async def test_protocol_version_is_none_before_initialize():
+    async with connected_runner(_runner_server([]), initialized=False) as (_client, runner):
+        assert runner.session.protocol_version is None
+
+
+@pytest.mark.anyio
+async def test_protocol_version_is_negotiated_version_after_initialize():
+    """A supported requested version is echoed back and readable on the session,
+    both directly and from inside a handler via `ctx.session`."""
+    seen: list[str | None] = []
+    async with connected_runner(_runner_server(seen), initialized=False) as (client, runner):
+        result = await client.send_raw_request("initialize", _init_params("2025-03-26"))
+        assert result["protocolVersion"] == "2025-03-26"
+        assert runner.session.protocol_version == "2025-03-26"
+        await client.send_raw_request("tools/list", None)
+        assert seen == ["2025-03-26"]
+
+
+@pytest.mark.anyio
+async def test_protocol_version_reads_latest_when_requested_version_unsupported():
+    """An unsupported requested version negotiates down to LATEST_PROTOCOL_VERSION."""
+    async with connected_runner(_runner_server([]), initialized=False) as (client, runner):
+        result = await client.send_raw_request("initialize", _init_params("1999-01-01"))
+        assert result["protocolVersion"] == LATEST_PROTOCOL_VERSION
+        assert runner.session.protocol_version == LATEST_PROTOCOL_VERSION
+
+
+@pytest.mark.anyio
+async def test_protocol_version_is_none_on_stateless_connection():
+    """Stateless connections never see a handshake: requests flow, but the
+    negotiated version legitimately stays None."""
+    seen: list[str | None] = []
+    async with connected_runner(_runner_server(seen), initialized=False, stateless=True) as (client, runner):
+        result = await client.send_raw_request("tools/list", None)
+        assert result == {"tools": []}
+        assert seen == [None]
+        assert runner.session.protocol_version is None