Class McpClient

tests/test_daemon_smoke.py:351–452

class McpClient:
    def __init__(self, binary, env, stderr_path):
        self.stderr_path = stderr_path
        self.stderr_stream = stderr_path.open("w", encoding="utf-8")
        self.process = subprocess.Popen(
            [str(binary)],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=self.stderr_stream,
            text=True,
            bufsize=1,
            env=env,
        )
        self.condition = threading.Condition()
        self.responses = {}
        self.other_output = []
        self.stdout_closed = False
        self.reader = threading.Thread(target=self._read_stdout, daemon=True)
        self.reader.start()

    def _read_stdout(self):
        assert self.process.stdout is not None
        for raw_line in self.process.stdout:
            line = raw_line.strip()
            if not line:
                continue
            try:
                value = json.loads(line)
            except json.JSONDecodeError:
                value = None
            with self.condition:
                if isinstance(value, dict) and "id" in value:
                    self.responses[value["id"]] = value
                else:
                    self.other_output.append(line)
                self.condition.notify_all()
        with self.condition:
            self.stdout_closed = True
            self.condition.notify_all()

    def send(self, value):
        check(self.process.stdin is not None, "client stdin is already closed")
        payload = json.dumps(value, separators=(",", ":"))
        try:
            self.process.stdin.write(payload + "\n")
            self.process.stdin.flush()
        except (BrokenPipeError, OSError) as exc:
            raise SmokeFailure("thin client closed while sending: " + str(exc)) from exc

    def wait_response(self, request_id, timeout=START_TIMEOUT):
        deadline = time.monotonic() + timeout
        with self.condition:
            while request_id not in self.responses:
                remaining = deadline - time.monotonic()
                if remaining <= 0 or self.stdout_closed:
                    break
                self.condition.wait(remaining)
            if request_id in self.responses: