fix(mcp): serialize keepalive with active RPCs

tests/tools/test_mcp_capability_gating.py

@@ -181,6 +181,71 @@ async def test_keepalive_uses_ping_legacy_fallback(self):
         task.session.send_ping.assert_awaited_once()
         task.session.list_tools.assert_not_called()
 
+    async def test_keepalive_skips_an_active_tool_rpc(self):
+        """An active RPC proves liveness; keepalive must not wait or overlap."""
+        task = MCPServerTask("test")
+        rpc_started = asyncio.Event()
+        release_rpc = asyncio.Event()
+        ping_started = asyncio.Event()
+
+        async def active_tool_rpc():
+            async with task._rpc_lock:
+                rpc_started.set()
+                await release_rpc.wait()
+
+        async def send_ping():
+            ping_started.set()
+
+        task.session = SimpleNamespace(
+            list_tools=AsyncMock(),
+            send_ping=AsyncMock(side_effect=send_ping),
+        )
+
+        active_task = asyncio.create_task(active_tool_rpc())
+        try:
+            await rpc_started.wait()
+            await asyncio.wait_for(task._keepalive_probe(), timeout=0.1)
+
+            assert not ping_started.is_set()
+            task.session.send_ping.assert_not_awaited()
+            task.session.list_tools.assert_not_awaited()
+        finally:
+            release_rpc.set()
+            await active_task
+
+    async def test_active_keepalive_serializes_a_user_rpc(self):
+        """A user RPC must wait while keepalive owns the shared session lock."""
+        task = MCPServerTask("test")
+        ping_started = asyncio.Event()
+        release_ping = asyncio.Event()
+        rpc_started = asyncio.Event()
+
+        async def send_ping():
+            ping_started.set()
+            await release_ping.wait()
+
+        async def user_rpc():
+            async with task._rpc_lock:
+                rpc_started.set()
+
+        task.session = SimpleNamespace(
+            list_tools=AsyncMock(),
+            send_ping=AsyncMock(side_effect=send_ping),
+        )
+
+        keepalive_task = asyncio.create_task(task._keepalive_probe())
+        await ping_started.wait()
+        rpc_task = asyncio.create_task(user_rpc())
+        try:
+            await asyncio.sleep(0)
+            assert not rpc_started.is_set()
+        finally:
+            release_ping.set()
+            await keepalive_task
+            await rpc_task
+
+        assert rpc_started.is_set()
+
 
 class TestKeepaliveInterval:
     """The keepalive cadence is configurable so servers with short session

tools/mcp_tool.py

@@ -1556,7 +1556,9 @@ def __init__(self, name: str):
         # handler calls list_tools while a normal tool call is in flight, the
         # stream can wedge and the user-visible tool call times out. Serialize
         # client-initiated RPCs per server. The lock is also applied to HTTP
-        # transports for conservative per-server ordering.
+        # transports for conservative per-server ordering. Keepalive also
+        # treats ownership as proof of liveness, so every future ClientSession
+        # request must acquire this lock to preserve that invariant.
         self._rpc_lock = asyncio.Lock()
         self._pending_refresh_tasks: set[asyncio.Task] = set()
         # contextvars snapshot of the agent task that's currently in
@@ -1834,32 +1836,56 @@ async def _keepalive_probe(self) -> None:
         transport connection so a server that gains ping support after a
         reconnect is re-probed with the cheap path.
 
+        When a user RPC owns ``_rpc_lock``, that RPC and its own timeout are the
+        liveness detector; the periodic probe is intentionally skipped.
+
         Raises on a genuine connection failure so the caller triggers a
         reconnect; returns normally when the session is alive.
         """
-        if not self._ping_unsupported:
-            try:
-                await asyncio.wait_for(self.session.send_ping(), timeout=30.0)
-                return
-            except Exception as exc:
-                # Only a "method not found" means ping is unsupported. Any
-                # other error (timeout, closed transport, session expired) is
-                # a real liveness failure — propagate so we reconnect.
-                if not _is_method_not_found_error(exc):
-                    raise
-                if not self._advertises_tools():
-                    # No ping, no tools → no cheaper probe to fall back to.
-                    raise
-                self._ping_unsupported = True
-                logger.info(
-                    "MCP server '%s': does not implement the optional 'ping' "
-                    "utility (-32601); using 'list_tools' for keepalive on "
-                    "this connection.",
-                    self.name,
-                )
+        # Session replacement is owned by the lifecycle coroutine that calls
+        # this probe; shutdown waits for that task before clearing the session.
+        # Keep one reference throughout the probe so ping and fallback cannot
+        # accidentally target different sessions.
+        session = self.session
+        if session is None:
+            raise RuntimeError("MCP keepalive requested without an active session")
+
+        # A ClientSession is one JSON-RPC stream. Sending a keepalive while a
+        # tools/call is in flight reproducibly wedges the original request on
+        # some SDK/server combinations even though the ping itself succeeds.
+        # The underlying SDK response-routing mechanism remains unidentified;
+        # this serialization is the durable workaround, not redundant locking.
+        # An active RPC already proves liveness, so skip this cycle rather than
+        # queueing a redundant probe behind a potentially long-running call.
+        if self._rpc_lock.locked():
+            return
+
+        # Every other client-initiated RPC uses this lock, so keepalive must too.
+        # The lock also closes the race with a call that starts after the check.
+        async with self._rpc_lock:
+            if not self._ping_unsupported:
+                try:
+                    await asyncio.wait_for(session.send_ping(), timeout=30.0)
+                    return
+                except Exception as exc:
+                    # Only a "method not found" means ping is unsupported. Any
+                    # other error (timeout, closed transport, session expired) is
+                    # a real liveness failure — propagate so we reconnect.
+                    if not _is_method_not_found_error(exc):
+                        raise
+                    if not self._advertises_tools():
+                        # No ping, no tools → no cheaper probe to fall back to.
+                        raise
+                    self._ping_unsupported = True
+                    logger.info(
+                        "MCP server '%s': does not implement the optional 'ping' "
+                        "utility (-32601); using 'list_tools' for keepalive on "
+                        "this connection.",
+                        self.name,
+                    )
 
-        # Fallback probe for servers without ping support.
-        await asyncio.wait_for(self.session.list_tools(), timeout=30.0)
+            # Fallback probe for servers without ping support.
+            await asyncio.wait_for(session.list_tools(), timeout=30.0)
 
     async def _wait_for_lifecycle_event(self) -> str:
         """Block until either _shutdown_event or _reconnect_event fires.