From b5ca19118e698e9f2a24a0f52fc49a326f840ff3 Mon Sep 17 00:00:00 2001 From: Yuanang Yang <178757933+nanami7777777@users.noreply.github.com> Date: Sat, 1 Aug 2026 11:25:37 +0530 Subject: [PATCH] fix(mcp): guard against duplicate spawns and stale connecting entries (#58862) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three fixes for concurrent MCP server spawn races in register_mcp_servers() and discover_mcp_tools(): 1. register_mcp_servers: add k not in _server_connecting guard to the new_servers filter. Without this, a concurrent second call sees the same servers as 'new' and spawns duplicate stdio subprocesses. 2. discover_mcp_tools: same _server_connecting guard in the new_server_names filter. This entry point is called from CLI, TUI, gateway, and cron — any two racing would double-spawn. 3. Stale _server_connecting cleanup on TimeoutError/InterruptedError. When _run_on_mcp_loop times out or is interrupted, _discover_all's gather may not have finished, leaving entries stranded in _server_connecting that block future reconnection attempts. The cleanup clears only entries added by this call (not external ones), logs a warning, and records connect errors. Salvage of #58879 by @nanami7777777 (superset of #58867 by @liuhao1024). Adapted to current main which has evolved significantly since July 5. Closes #58862 Closes #58867 Closes #58879 --- tests/tools/test_mcp_tool.py | 73 ++++++++++++++++++++++++++++++++++++ tools/mcp_tool.py | 35 +++++++++++++++-- 2 files changed, 105 insertions(+), 3 deletions(-) diff --git a/tests/tools/test_mcp_tool.py b/tests/tools/test_mcp_tool.py index 3ecbf9866682d..478693fcaa15b 100644 --- a/tests/tools/test_mcp_tool.py +++ b/tests/tools/test_mcp_tool.py @@ -2454,6 +2454,79 @@ class TestRegisterMcpServers: assert "mcp__my_server__tool1" in result _servers.pop("my_server", None) + def test_skips_servers_already_connecting(self): + """Servers in _server_connecting must not be spawned again (#58862).""" + from tools.mcp_tool import ( + register_mcp_servers, _servers, _server_connecting, _ensure_mcp_loop, + ) + + fake_config = {"my_srv": {"command": "npx", "args": ["test"]}} + + # Simulate a prior call that started connecting but hasn't finished + _server_connecting.add("my_srv") + connect_calls = [] + + async def fake_register(name, cfg): + connect_calls.append(name) + server = _make_mock_server(name) + server._registered_tool_names = [f"mcp_{name}_tool"] + _servers[name] = server + return [f"mcp_{name}_tool"] + + try: + with patch("tools.mcp_tool._MCP_AVAILABLE", True), \ + patch("tools.mcp_tool._discover_and_register_server", side_effect=fake_register), \ + patch("tools.mcp_tool._existing_tool_names", return_value=[]), \ + patch("tools.mcp_tool._connect_cooldown_active", return_value=False): + _ensure_mcp_loop() + result = register_mcp_servers(fake_config) + + # Should NOT have attempted to connect my_srv again + assert connect_calls == [], ( + f"Server already in _server_connecting should be skipped, " + f"but connect was called for: {connect_calls}" + ) + assert result == [] + finally: + _server_connecting.discard("my_srv") + _servers.pop("my_srv", None) + + def test_clears_stale_connecting_on_timeout(self): + """Stale entries in _server_connecting are cleaned up after timeout (#58862).""" + from tools.mcp_tool import ( + register_mcp_servers, _servers, _server_connecting, + _server_connect_errors, _ensure_mcp_loop, + ) + + fake_config = { + "srv_a": {"command": "npx", "args": ["a"]}, + "srv_b": {"command": "npx", "args": ["b"]}, + } + + # Simulate that srv_a is already connecting from another call + _server_connecting.add("srv_a") + + with patch("tools.mcp_tool._MCP_AVAILABLE", True), \ + patch("tools.mcp_tool._run_on_mcp_loop", side_effect=TimeoutError("timed out")), \ + patch("tools.mcp_tool._existing_tool_names", return_value=[]), \ + patch("tools.mcp_tool._connect_cooldown_active", return_value=False): + _ensure_mcp_loop() + + with pytest.raises(TimeoutError): + register_mcp_servers(fake_config) + + # After timeout, srv_b (which was in new_servers and added to _server_connecting) + # should have been cleaned up from _server_connecting. + # srv_a should remain since it was added externally and not part of new_servers. + assert "srv_b" not in _server_connecting, ( + "Stale server added during this call should have been removed from " + "_server_connecting after timeout" + ) + # Cleanup + _server_connecting.discard("srv_a") + _servers.pop("srv_a", None) + _servers.pop("srv_b", None) + # --------------------------------------------------------------------------- # Tests for parallel tool call support (port from openai/codex#17667) # --------------------------------------------------------------------------- diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index 83a827c8b8bfb..b3c356788a084 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -5969,13 +5969,17 @@ def register_mcp_servers(servers: Dict[str, dict]) -> List[str]: logger.debug("No explicit MCP servers provided") return [] - # Only attempt servers that aren't already connected and are enabled - # (enabled: false skips the server entirely without removing its config) + # Only attempt servers that aren't already connected (or currently + # connecting) and are enabled. Checking ``_server_connecting`` prevents + # duplicate subprocess spawns when ``discover_mcp_tools()`` is called + # from multiple entry-points before the first batch finishes (#58862). with _lock: + connecting = set(_server_connecting) new_servers = { k: v for k, v in servers.items() if k not in _servers + and k not in connecting and _parse_boolish(v.get("enabled", True), default=True) # Skip a server still serving its post-failure backoff. Without # this, a server that fails to connect (and is therefore never @@ -6061,6 +6065,28 @@ def register_mcp_servers(servers: Dict[str, dict]) -> List[str]: _set_interrupt(False) try: _run_on_mcp_loop(_discover_all, timeout=120) + except (TimeoutError, InterruptedError) as _e: + # When the outer timeout fires or the user interrupts, + # _discover_all's gather may not have finished, leaving + # entries stranded in _server_connecting. Those stale + # entries would block future reconnection attempts (#58862). + with _lock: + stale = [n for n in new_servers if n in _server_connecting] + if stale: + logger.warning( + "MCP discovery %s while %d server(s) were still " + "connecting; clearing stale connecting set: %s", + "timed out" if isinstance(_e, TimeoutError) else "interrupted", + len(stale), + ", ".join(stale), + ) + _server_connecting.difference_update(stale) + for _sn in stale: + _server_connect_errors.setdefault( + _sn, + f"Connection attempt {'timed out' if isinstance(_e, TimeoutError) else 'interrupted'} during discovery", + ) + raise finally: if _was_interrupted: _set_interrupt(True) @@ -6133,10 +6159,13 @@ def discover_mcp_tools() -> List[str]: try: with _lock: + connecting = set(_server_connecting) new_server_names = [ name for name, cfg in servers.items() - if name not in _servers and _parse_boolish(cfg.get("enabled", True), default=True) + if name not in _servers + and name not in connecting + and _parse_boolish(cfg.get("enabled", True), default=True) ] tool_names = register_mcp_servers(servers)