mirror of
https://github.com/zhayujie/chatgpt-on-wechat.git
synced 2026-07-19 12:47:25 +08:00
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
This commit is contained in:
@@ -7,7 +7,7 @@ without any external MCP SDK dependency.
|
|||||||
|
|
||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import select
|
import queue
|
||||||
import subprocess
|
import subprocess
|
||||||
import threading
|
import threading
|
||||||
import urllib.request
|
import urllib.request
|
||||||
@@ -34,6 +34,8 @@ class McpClient:
|
|||||||
self.config = config
|
self.config = config
|
||||||
self.name: str = config.get("name", "unknown")
|
self.name: str = config.get("name", "unknown")
|
||||||
raw_transport: str = config.get("type", "stdio")
|
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
|
# Normalize streamable-http aliases to a single internal key
|
||||||
self.transport: str = (
|
self.transport: str = (
|
||||||
"streamable-http"
|
"streamable-http"
|
||||||
@@ -43,6 +45,7 @@ class McpClient:
|
|||||||
|
|
||||||
# stdio state
|
# stdio state
|
||||||
self._proc: Optional[subprocess.Popen] = None
|
self._proc: Optional[subprocess.Popen] = None
|
||||||
|
self._read_queue: queue.Queue = queue.Queue()
|
||||||
|
|
||||||
# SSE state
|
# SSE state
|
||||||
self._sse_url: Optional[str] = None
|
self._sse_url: Optional[str] = None
|
||||||
@@ -172,6 +175,9 @@ class McpClient:
|
|||||||
threading.Thread(
|
threading.Thread(
|
||||||
target=self._drain_stderr, daemon=True, name=f"mcp-stderr-{self.name}"
|
target=self._drain_stderr, daemon=True, name=f"mcp-stderr-{self.name}"
|
||||||
).start()
|
).start()
|
||||||
|
threading.Thread(
|
||||||
|
target=self._drain_stdout, daemon=True, name=f"mcp-stdout-{self.name}"
|
||||||
|
).start()
|
||||||
|
|
||||||
return self._handshake()
|
return self._handshake()
|
||||||
|
|
||||||
@@ -179,14 +185,35 @@ class McpClient:
|
|||||||
for line in self._proc.stderr:
|
for line in self._proc.stderr:
|
||||||
line = line.strip()
|
line = line.strip()
|
||||||
if line:
|
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:
|
def _drain_stdout(self):
|
||||||
"""Read one line from stdio stdout with a hard timeout."""
|
"""Background thread: read lines from stdout and put them into the queue."""
|
||||||
ready, _, _ = select.select([self._proc.stdout], [], [], timeout)
|
try:
|
||||||
if not ready:
|
for line in self._proc.stdout:
|
||||||
raise TimeoutError(f"[MCP:{self.name}] stdio read timed out after {timeout}s")
|
self._read_queue.put(line)
|
||||||
return self._proc.stdout.readline()
|
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:
|
def _stdio_send(self, message: dict) -> dict:
|
||||||
"""Send a JSON-RPC message over stdio and read the response."""
|
"""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.write(raw)
|
||||||
self._proc.stdin.flush()
|
self._proc.stdin.flush()
|
||||||
|
|
||||||
|
expected_id = message.get("id")
|
||||||
while True:
|
while True:
|
||||||
line = self._readline_with_timeout()
|
line = self._readline_with_timeout()
|
||||||
if not line:
|
if not line:
|
||||||
@@ -208,6 +236,14 @@ class McpClient:
|
|||||||
if "id" not in data:
|
if "id" not in data:
|
||||||
logger.debug(f"[MCP:{self.name}] notification skipped: {data.get('method', '?')}")
|
logger.debug(f"[MCP:{self.name}] notification skipped: {data.get('method', '?')}")
|
||||||
continue
|
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
|
return data
|
||||||
|
|
||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
|
|||||||
Reference in New Issue
Block a user