diff --git a/pyproject.toml b/pyproject.toml index df8dd4405..c33ba8034 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,7 +70,7 @@ dependencies = [ "chromadb>=1.0.0,<2.0.0", "qdrant-client (>=1.15.1,<2.0.0)", "pyseekdb==1.1.0.post3", - "langbot-plugin==0.4.7", + "langbot-plugin==0.4.8", "asyncpg>=0.30.0", "line-bot-sdk>=3.19.0", "matrix-nio>=0.25.2", diff --git a/src/langbot/pkg/provider/tools/loaders/mcp.py b/src/langbot/pkg/provider/tools/loaders/mcp.py index e059a756d..9fb3907ab 100644 --- a/src/langbot/pkg/provider/tools/loaders/mcp.py +++ b/src/langbot/pkg/provider/tools/loaders/mcp.py @@ -185,6 +185,16 @@ class MCPSessionStatus(enum.Enum): ERROR = 'error' +class _TransportReconnect(Exception): + """Internal signal: the Box stdio WS transport dropped but the managed + process is still alive. Triggers a lightweight transport reconnect that + reuses the live process, instead of a full process rebuild. + + Reconnect attempts are NOT counted toward the fatal retry budget, so a + long-lived session can survive arbitrarily many transient drops. + """ + + class RuntimeMCPSession: """运行时 MCP 会话""" @@ -254,6 +264,9 @@ class RuntimeMCPSession: self._lifecycle_task = None self._shutdown_event = asyncio.Event() self._ready_event = asyncio.Event() + # Set transiently when a WS transport drop should NOT stop the managed + # process (it will be re-attached on the next initialize()). + self._preserve_managed_process = False self._box_stdio_runtime = BoxStdioSessionRuntime(self) self.box_config = self._box_stdio_runtime.config @@ -399,6 +412,28 @@ class RuntimeMCPSession: task.cancel() for task in done: if task is monitor_task and not self._shutdown_event.is_set(): + # The monitor completed. This is EITHER the managed + # process actually exiting OR just the WS transport + # dropping while the process stays alive in the Box + # runtime. Re-check the real process state so a + # transient transport drop reconnects (reusing the live + # process) instead of tearing the process down and + # running a full rebuild+backoff cycle. + process_still_running = False + try: + process_still_running = await self._box_stdio_runtime._managed_process_is_running() + except Exception: + process_still_running = False + if process_still_running: + self.ap.logger.info( + f'MCP server {self.server_name}: transport dropped but ' + f'managed process is still running; reconnecting transport' + ) + self.error_phase = MCPSessionErrorPhase.RELAY_CONNECT + # Preserve the live process across the finally-block + # cleanup: only the WS transport should be torn down. + self._preserve_managed_process = True + raise _TransportReconnect('Box managed process transport dropped; reconnecting') self.error_phase = MCPSessionErrorPhase.RUNTIME raise Exception('Box managed process exited unexpectedly') else: @@ -424,14 +459,37 @@ class RuntimeMCPSession: except Exception as e: self.ap.logger.error(f'Error cleaning up MCP session {self.server_name}: {e}\n{traceback.format_exc()}') finally: - await self._cleanup_box_stdio_session() + # On a transport-only reconnect the managed process is healthy + # and will be re-attached on the next initialize(); do NOT stop + # it. Any other exit path fully tears the session down. + if getattr(self, '_preserve_managed_process', False): + self._preserve_managed_process = False + else: + await self._cleanup_box_stdio_session() async def _lifecycle_loop_with_retry(self): """Wrap _lifecycle_loop with retry and exponential backoff.""" - for attempt in range(self._MAX_RETRIES + 1): + attempt = 0 + while attempt <= self._MAX_RETRIES: try: await self._lifecycle_loop() return # Normal shutdown, don't retry + except _TransportReconnect as e: + # Transient WS transport drop while the managed process is still + # alive. Reconnect promptly WITHOUT consuming the fatal retry + # budget and WITHOUT stopping the process — initialize() will + # re-attach to the live process. This is what lets a long-lived + # stdio MCP survive repeated brief event-loop stalls / pings. + if self._shutdown_event.is_set(): + return + self.ap.logger.info( + f'MCP session {self.server_name}: reconnecting transport ({self._describe_exception(e)})' + ) + self.status = MCPSessionStatus.CONNECTING + self.error_message = None + self.error_phase = None + await asyncio.sleep(1) + continue except Exception as e: self.retry_count = attempt + 1 if self._shutdown_event.is_set(): @@ -460,6 +518,7 @@ class RuntimeMCPSession: self.error_message = None self.error_phase = None await asyncio.sleep(delay) + attempt += 1 @staticmethod def _describe_exception(exc: BaseException) -> str: diff --git a/src/langbot/pkg/provider/tools/loaders/mcp_stdio.py b/src/langbot/pkg/provider/tools/loaders/mcp_stdio.py index dcfbb9132..4bab3712f 100644 --- a/src/langbot/pkg/provider/tools/loaders/mcp_stdio.py +++ b/src/langbot/pkg/provider/tools/loaders/mcp_stdio.py @@ -173,25 +173,36 @@ class BoxStdioSessionRuntime: stderr_preview = (result.stderr or '')[:500] raise Exception(f'Dependency install failed (exit code {result.exit_code}): {stderr_preview}') - try: - process_workspace = ( - self._build_workspace(host_path=host_path, workdir=process_cwd, mount_path=process_cwd) - if host_path - else workspace + # Reuse an already-running managed process instead of rebuilding it. + # The Box runtime keeps the managed process alive across a transient + # WebSocket transport drop, so on a reconnect we only need to re-attach + # the WS below. Rebuilding here would needlessly stop a healthy process + # and re-run the (slow, network-touching) dependency bootstrap. + if not await self._managed_process_is_running(): + try: + process_workspace = ( + self._build_workspace(host_path=host_path, workdir=process_cwd, mount_path=process_cwd) + if host_path + else workspace + ) + payload = process_workspace.build_process_payload( + self.server_config['command'], + self.server_config.get('args', []), + env=self.server_config.get('env', {}), + cwd=process_cwd, + ) + if install_cmd: + payload = self._wrap_process_payload_with_python_env(payload, process_cwd) + payload['process_id'] = self.process_id + await workspace.box_service.start_managed_process(workspace.session_id, payload) + except Exception: + self.owner.error_phase = MCPSessionErrorPhase.PROCESS_START + raise + else: + self.ap.logger.info( + f'MCP server {self.server_name}: reusing live managed process ' + f'process_id={self.process_id} (transport reconnect)' ) - payload = process_workspace.build_process_payload( - self.server_config['command'], - self.server_config.get('args', []), - env=self.server_config.get('env', {}), - cwd=process_cwd, - ) - if install_cmd: - payload = self._wrap_process_payload_with_python_env(payload, process_cwd) - payload['process_id'] = self.process_id - await workspace.box_service.start_managed_process(workspace.session_id, payload) - except Exception: - self.owner.error_phase = MCPSessionErrorPhase.PROCESS_START - raise try: websocket_url = workspace.get_managed_process_websocket_url(self.process_id) @@ -236,6 +247,23 @@ class BoxStdioSessionRuntime: return await asyncio.sleep(self.owner._MONITOR_POLL_INTERVAL) + async def _managed_process_is_running(self) -> bool: + """Return True if this server's managed process exists and is running. + + Used to decide whether initialize() must (re)start the process or can + simply re-attach the WebSocket transport to a process the Box runtime + kept alive across a transient transport drop. + """ + from langbot_plugin.box.models import BoxManagedProcessStatus + + workspace = self._build_workspace() + try: + info = await workspace.get_managed_process(self.process_id) + except Exception: + return False + status = info.get('status', '') if isinstance(info, dict) else getattr(info, 'status', '') + return status in (BoxManagedProcessStatus.RUNNING.value, BoxManagedProcessStatus.RUNNING) + async def _stage_host_path_to_shared_workspace(self, host_path: str) -> str: source_path = normalize_host_path(host_path) if not source_path: diff --git a/uv.lock b/uv.lock index 3c2627117..7ecac56bd 100644 --- a/uv.lock +++ b/uv.lock @@ -2123,7 +2123,7 @@ requires-dist = [ { name = "ebooklib", specifier = ">=0.18" }, { name = "gewechat-client", specifier = ">=0.1.5" }, { name = "html2text", specifier = ">=2024.2.26" }, - { name = "langbot-plugin", specifier = "==0.4.7" }, + { name = "langbot-plugin", specifier = "==0.4.8" }, { name = "langchain", specifier = ">=1.3.9" }, { name = "langchain-core", specifier = ">=1.3.3" }, { name = "langchain-text-splitters", specifier = ">=1.1.2" }, @@ -2187,7 +2187,7 @@ dev = [ [[package]] name = "langbot-plugin" -version = "0.4.7" +version = "0.4.8" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "aiofiles" }, @@ -2208,9 +2208,9 @@ dependencies = [ { name = "watchdog" }, { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/00/6c/850a92cc05583ab71e6ac527ff50fa5f3db43cc16e86f4f6d5ac684da9a8/langbot_plugin-0.4.7.tar.gz", hash = "sha256:16b24d79fc55c0a6b15d901b1134de9e298e2cadab9a9b25c9231f731333ebe9", size = 333867, upload-time = "2026-07-02T16:49:46.258Z" } +sdist = { url = "https://files.pythonhosted.org/packages/7f/f5/1b417a48d329f959a261eaf9a37b26197f18081d2b6f0ef3e4bbab687b81/langbot_plugin-0.4.8.tar.gz", hash = "sha256:e9614be248acf352acb9430d9e154e0abae981575de16812f68282c92a4e42d8", size = 334076, upload-time = "2026-07-03T04:15:28.233Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/c6/f5/a00a929ffb4dcbd958e5f5249516f19ecc720f25f072c9bfda8bbb56851c/langbot_plugin-0.4.7-py3-none-any.whl", hash = "sha256:1364f80fcf448f4503e87d1367af60438abd2cf16e78be6807032cbdb4710a56", size = 220999, upload-time = "2026-07-02T16:49:44.868Z" }, + { url = "https://files.pythonhosted.org/packages/40/21/46bfa911ab27d6856a3b4363805c4bb8a657500cbc12e1e719d0b3b349dc/langbot_plugin-0.4.8-py3-none-any.whl", hash = "sha256:87c12d9d5579b0ab4dc43c9e559b4d15d7387db8c8b52d476cf27a3de74aedf5", size = 221244, upload-time = "2026-07-03T04:15:27.087Z" }, ] [[package]]