From 639a3eac1e467b832c80ffe8a260221e32be9849 Mon Sep 17 00:00:00 2001
From: liusk <16952718+ai-trip@user.noreply.gitee.com>
Date: Thu, 4 Jun 2026 09:10:48 +0800
Subject: [PATCH] fix(mcp): replace select.select with queue.Queue for
 cross-platform stdio I/O

- Use reader thread + queue.Queue instead of select.select() which does not
  work with pipes on Windows (only sockets)
- Make MCP server timeout configurable via mcp.json (default 120s)
- Validate JSON-RPC response id to skip stale responses from timed-out calls
- Log MCP server stderr at WARNING level instead of DEBUG for visibility
---
 agent/tools/mcp/mcp_client.py | 52 +++++++++++++++++++++++++++++------
 1 file changed, 44 insertions(+), 8 deletions(-)

diff --git a/agent/tools/mcp/mcp_client.py b/agent/tools/mcp/mcp_client.py
index be93c716..5cf86174 100644
--- a/agent/tools/mcp/mcp_client.py
+++ b/agent/tools/mcp/mcp_client.py
@@ -7,7 +7,7 @@ without any external MCP SDK dependency.
 
 import json
 import os
-import select
+import queue
 import subprocess
 import threading
 import urllib.request
@@ -34,6 +34,8 @@ class McpClient:
         self.config = config
         self.name: str = config.get("name", "unknown")
         raw_transport: str = config.get("type", "stdio")
+        # Per-server timeout for tool calls (default 120s, suitable for data queries)
+        self._timeout: int = int(config.get("timeout", 120))
         # Normalize streamable-http aliases to a single internal key
         self.transport: str = (
             "streamable-http"
@@ -43,6 +45,7 @@ class McpClient:
 
         # stdio state
         self._proc: Optional[subprocess.Popen] = None
+        self._read_queue: queue.Queue = queue.Queue()
 
         # SSE state
         self._sse_url: Optional[str] = None
@@ -172,6 +175,9 @@ class McpClient:
         threading.Thread(
             target=self._drain_stderr, daemon=True, name=f"mcp-stderr-{self.name}"
         ).start()
+        threading.Thread(
+            target=self._drain_stdout, daemon=True, name=f"mcp-stdout-{self.name}"
+        ).start()
 
         return self._handshake()
 
@@ -179,14 +185,35 @@ class McpClient:
         for line in self._proc.stderr:
             line = line.strip()
             if line:
-                logger.debug(f"[MCP:{self.name}] stderr: {line}")
+                logger.warning(f"[MCP:{self.name}] stderr: {line}")
 
-    def _readline_with_timeout(self, timeout: int = 30) -> str:
-        """Read one line from stdio stdout with a hard timeout."""
-        ready, _, _ = select.select([self._proc.stdout], [], [], timeout)
-        if not ready:
-            raise TimeoutError(f"[MCP:{self.name}] stdio read timed out after {timeout}s")
-        return self._proc.stdout.readline()
+    def _drain_stdout(self):
+        """Background thread: read lines from stdout and put them into the queue."""
+        try:
+            for line in self._proc.stdout:
+                self._read_queue.put(line)
+        except Exception:
+            pass
+        finally:
+            try:
+                self._read_queue.put("")
+            except Exception:
+                pass
+
+    def _readline_with_timeout(self, timeout: Optional[int] = None) -> str:
+        """Read one line from stdio stdout with a hard timeout (cross-platform).
+
+        Uses the per-server timeout from mcp.json config when no explicit
+        timeout is provided.
+        """
+        effective = timeout if timeout is not None else self._timeout
+        try:
+            line = self._read_queue.get(timeout=effective)
+        except queue.Empty:
+            raise TimeoutError(f"[MCP:{self.name}] stdio read timed out after {effective}s")
+        if not line:
+            raise IOError(f"[MCP:{self.name}] stdio process closed unexpectedly")
+        return line
 
     def _stdio_send(self, message: dict) -> dict:
         """Send a JSON-RPC message over stdio and read the response."""
@@ -194,6 +221,7 @@ class McpClient:
         self._proc.stdin.write(raw)
         self._proc.stdin.flush()
 
+        expected_id = message.get("id")
         while True:
             line = self._readline_with_timeout()
             if not line:
@@ -208,6 +236,14 @@ class McpClient:
             if "id" not in data:
                 logger.debug(f"[MCP:{self.name}] notification skipped: {data.get('method', '?')}")
                 continue
+            # Verify response id matches request id to avoid consuming a stale
+            # response left over from a previously failed/timed-out request.
+            if data.get("id") != expected_id:
+                logger.warning(
+                    f"[MCP:{self.name}] Stale response id={data.get('id')} "
+                    f"(expected {expected_id}), skipping"
+                )
+                continue
             return data
 
     # ------------------------------------------------------------------