fix(mcp): stdio 客户端改用读线程+队列，修复 Windows select 管道 WinError 10038

modules/mcp_client_manager/stdio_client.py

@@ -5,11 +5,12 @@ from __future__ import annotations
 import hashlib
 import json
 import os
-import select
+import queue
 import shutil
 import subprocess
 import threading
 import time
+from collections import deque
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, TYPE_CHECKING
@@ -17,6 +18,7 @@ from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, TYPE_CH
 import httpx
 
 from config import MCP_PROTOCOL_VERSION, MCP_DEFAULT_TIMEOUT_SECONDS
+from modules.mcp_client_manager.exceptions import MCPClientError
 from modules.mcp_server_registry import MCPServerRegistry
 
 if TYPE_CHECKING:
@@ -65,6 +67,11 @@ class _StdioMCPClient:
         self._initialized = False
         self._initialize_result: Dict[str, Any] = {}
         self._negotiated_protocol_version: Optional[str] = None
+        # stdout/stderr 后台排空线程状态。Windows 的 select 仅支持套接字、
+        # 不支持匿名管道（WinError 10038），因此统一用读线程 + 队列做跨平台超时读取。
+        self._stdout_queue: "queue.Queue[Optional[str]]" = queue.Queue()
+        self._stdout_eof = False
+        self._stderr_lines: "deque[str]" = deque(maxlen=80)
 
     def _next_id(self) -> int:
         self._seq += 1
@@ -263,10 +270,54 @@ class _StdioMCPClient:
             )
         except Exception as exc:
             raise MCPClientError(f"启动 stdio MCP 服务失败: {exc}") from exc
+        # 启动后台排空线程：stdout 按行入队（供 _read_message 带超时消费），
+        # stderr 持续排空（防止管道缓冲写满阻塞服务进程），仅保留末尾行用于诊断。
+        self._stdout_queue = queue.Queue()
+        self._stdout_eof = False
+        self._stderr_lines = deque(maxlen=80)
+        server_id = str(self.server.get("id") or "?")
+        threading.Thread(
+            target=self._drain_stdout_stream,
+            name=f"mcp-stdout-{server_id}",
+            daemon=True,
+        ).start()
+        threading.Thread(
+            target=self._drain_stderr_stream,
+            name=f"mcp-stderr-{server_id}",
+            daemon=True,
+        ).start()
         self._initialized = False
         self._initialize_result = {}
         self._negotiated_protocol_version = None
 
+    def _drain_stdout_stream(self) -> None:
+        """后台线程：阻塞按行读取 stdout 推入队列，EOF/异常时推 None 哨兵。"""
+        stream = self.process.stdout if self.process else None
+        try:
+            if stream:
+                for line in stream:
+                    self._stdout_queue.put(line)
+        except Exception:
+            pass
+        finally:
+            self._stdout_queue.put(None)
+
+    def _drain_stderr_stream(self) -> None:
+        """后台线程：持续排空 stderr，仅保留末尾若干行用于错误诊断。"""
+        stream = self.process.stderr if self.process else None
+        try:
+            if stream:
+                for line in stream:
+                    self._stderr_lines.append(line)
+        except Exception:
+            pass
+
+    def _process_exit_error(self, last_text_line: str) -> MCPClientError:
+        returncode = self.process.returncode if self.process else None
+        stderr_tail = "".join(self._stderr_lines)[-400:] if self._stderr_lines else ""
+        detail = (stderr_tail or last_text_line or "")[:400]
+        return MCPClientError(f"stdio 进程已退出（code={returncode}）{detail}")
+
     def _send(self, payload: Dict[str, Any]) -> None:
         if not self.process or not self.process.stdin:
             raise MCPClientError("stdio 进程未启动")
@@ -282,36 +333,26 @@ class _StdioMCPClient:
             raise MCPClientError("stdio 进程未启动")
         timeout = self.timeout_seconds if timeout_seconds is None else max(0.1, float(timeout_seconds))
         deadline = time.monotonic() + timeout
-        stdout = self.process.stdout
         last_text_line = ""
+        # 跨平台读取：不依赖 select（Windows 仅支持套接字），
+        # 消费后台读线程产出的队列，用 queue.get(timeout) 做超时控制。
         while True:
-            if self.process.poll() is not None:
-                stderr_text = ""
-                stdout_text = ""
-                try:
-                    if self.process.stderr:
-                        stderr_text = self.process.stderr.read() or ""
-                except Exception:
-                    pass
-                try:
-                    if self.process.stdout:
-                        stdout_text = self.process.stdout.read() or ""
-                except Exception:
-                    pass
-                detail = (stderr_text or stdout_text or last_text_line or "")[:400]
-                raise MCPClientError(f"stdio 进程已退出（code={self.process.returncode}）{detail}")
-
             remain = deadline - time.monotonic()
             if remain <= 0:
                 raise MCPClientError("读取 stdio MCP 响应超时")
-
-            readable, _, _ = select.select([stdout], [], [], min(0.3, remain))
-            if not readable:
+            try:
+                item = self._stdout_queue.get(timeout=min(0.3, remain))
+            except queue.Empty:
+                if self._stdout_eof and (not self.process or self.process.poll() is not None):
+                    raise self._process_exit_error(last_text_line)
                 continue
-            line = stdout.readline()
-            if not line:
+            if item is None:
+                # 读线程到达 EOF：服务进程退出或流被关闭
+                self._stdout_eof = True
+                if not self.process or self.process.poll() is not None:
+                    raise self._process_exit_error(last_text_line)
                 continue
-            line = line.strip()
+            line = item.strip()
             if not line:
                 continue
             try: