From a136e818d7ec05db0a9b4be5c11357719e30f7fb Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Sat, 1 Aug 2026 10:45:29 +0800 Subject: [PATCH] =?UTF-8?q?fix(mcp):=20stdio=20=E5=AE=A2=E6=88=B7=E7=AB=AF?= =?UTF-8?q?=E6=94=B9=E7=94=A8=E8=AF=BB=E7=BA=BF=E7=A8=8B+=E9=98=9F?= =?UTF-8?q?=E5=88=97=EF=BC=8C=E4=BF=AE=E5=A4=8D=20Windows=20select=20?= =?UTF-8?q?=E7=AE=A1=E9=81=93=20WinError=2010038?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- modules/mcp_client_manager/stdio_client.py | 89 ++++++++++++++++------ 1 file changed, 65 insertions(+), 24 deletions(-) diff --git a/modules/mcp_client_manager/stdio_client.py b/modules/mcp_client_manager/stdio_client.py index 9a83207a..b49fc4f7 100644 --- a/modules/mcp_client_manager/stdio_client.py +++ b/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: