fix: reject changed duplicate initialize

Author: he-yufeng

src/mcp/server/runner.py

@@ -398,7 +398,12 @@ def _negotiate_initialize(params: Mapping[str, Any] | None) -> tuple[InitializeR
 
     def _handle_initialize(self, params: Mapping[str, Any] | None) -> InitializeResult:
         """Build the `initialize` result; state commits later in `_on_request`."""
-        _, negotiated = self._negotiate_initialize(params)
+        init, negotiated = self._negotiate_initialize(params)
+        if self.connection.client_params is not None and init != self.connection.client_params:
+            raise MCPError(
+                code=INVALID_PARAMS,
+                message="Session already initialized with different parameters",
+            )
         opts = self.init_options if self.init_options is not None else self.server.create_initialization_options()
         return InitializeResult(
             protocol_version=negotiated,

tests/server/test_runner.py

@@ -172,6 +172,25 @@ async def test_runner_handles_initialize_and_populates_connection(server: SrvT):
     assert runner.connection.initialize_accepted is True
 
 
+@pytest.mark.anyio
+async def test_runner_rejects_changed_duplicate_initialize(server: SrvT):
+    async with connected_runner(server, initialized=False) as (client, runner):
+        first_params = _initialize_params()
+        await client.send_raw_request("initialize", first_params)
+        assert runner.connection.client_params is not None
+        assert runner.connection.client_params.client_info.name == "test-client"
+
+        second_params = _initialize_params()
+        second_params["clientInfo"] = {"name": "second-client", "version": "1.0"}
+        with pytest.raises(MCPError) as exc:
+            await client.send_raw_request("initialize", second_params)
+
+        assert exc.value.error.code == INVALID_PARAMS
+        assert "already initialized" in exc.value.error.message
+        assert runner.connection.client_params is not None
+        assert runner.connection.client_params.client_info.name == "test-client"
+
+
 @pytest.mark.anyio
 async def test_runner_initialize_opens_gate_but_event_fires_only_after_initialized_notification(server: SrvT):
     """`initialize` commits the gate flag and peer info, but the public