-
Notifications
You must be signed in to change notification settings - Fork 2.6k
Description
Initial Checks
- I confirm that I'm using the latest version of MCP Python SDK
- I confirm that I searched for my issue in https://github.com/modelcontextprotocol/python-sdk/issues before opening this issue
Description
Description
The MCP Python SDK lacks automatic periodic ping functionality for connection health detection, despite the MCP protocol recommending with a SHOULD requirement that "Implementations SHOULD periodically issue pings to detect connection health." This creates a significant reliability risk where failed connections, network issues, or connection-based attacks may go undetected.
Detailed Analysis
The MCP Python SDK only provides a manual send_ping()
method without any automatic periodic ping functionality:
Current Implementation
# src/mcp/client/session.py:176-181
async def send_ping(self) -> types.EmptyResult:
"""Send a ping request."""
return await self.send_request(
types.ClientRequest(types.PingRequest()),
types.EmptyResult,
)
Missing Automatic Ping
- ❌ No automatic ping loop: No built-in mechanism to periodically send pings
- ❌ No connection health monitoring: Cannot detect connection failures automatically
- ❌ No keep-alive mechanism: Connections may silently fail without detection
- ❌ Protocol compliance gap: Does not meet "SHOULD periodically issue pings" requirement
Impact
Without automatic ping mechanisms, applications cannot reliably detect:
- Connection failures: Failed connections remain undetected
- Network issues: Network problems go unnoticed
- Connection-based attacks: Attacks targeting connection stability cannot be identified
- Resource leaks: Failed connections are not properly cleaned up
Suggested Implementation
Location: src/mcp/client/session.py
and src/mcp/server/session.py
Add automatic ping functionality:
class ClientSession:
def __init__(self,
read_stream,
write_stream,
auto_ping: bool = True, # Enable automatic ping
ping_interval: float = 30.0, # Ping interval in seconds
...):
self._auto_ping = auto_ping
self._ping_interval = ping_interval
if self._auto_ping:
self._ping_task = asyncio.create_task(self._ping_loop())
async def _ping_loop(self):
"""Automatic ping loop for connection health monitoring"""
while True:
try:
await asyncio.sleep(self._ping_interval)
await self.send_ping()
except Exception:
# Handle ping failures (connection lost)
break
Add connection health callbacks:
class ClientSession:
def __init__(self,
on_connection_lost: Callable = None,
on_connection_restored: Callable = None,
...):
self._on_connection_lost = on_connection_lost
self._on_connection_restored = on_connection_restored
Example Usage
With automatic ping:
client = ClientSession(
read_stream=read_stream,
write_stream=write_stream,
auto_ping=True, # Enable automatic ping
ping_interval=30.0, # Ping every 30 seconds
on_connection_lost=lambda: print("Connection lost!"),
on_connection_restored=lambda: print("Connection restored!")
)
Supporting Material/References
- MCP protocol specification regarding periodic ping requirements
- Current implementation in
src/mcp/client/session.py:176-181
- Related issue: [Missing Configurable Ping Frequency and Rate Limiting]
Note that this MCP Python SDK Implementation Gap-5: Missing Automatic Ping Mechanism for Connection Health Monitoring is different from MCP Python SDK Implementation Gap-4: Missing Configurable Ping Frequency and Rate Limiting. The former (Gap-4) focuses on lacking rate limitation protection, while this one (Gap-5) focuses on lacking the automatic calling of ping itself.
Example Code
Python & MCP Python SDK
latest