INNER CODE UNIT · Python

_send

CursorTouch/Android-MCP · e2e_test.py:84

def _send(proc, obj):
    proc.stdin.write((json.dumps(obj) + "\n").encode())
    proc.stdin.flush()


def _recv(proc, timeout=60):
    deadline = time.time() + timeout
    while True:
        if time.time() > deadline:
            raise TimeoutError(f"No response within {timeout}s")
        line = proc.stdout.readline()
        if not line:
            raise EOFError("Server stdout closed")
        line = line.strip()
        if line:
            return json.loads(line)

View source record →