audio_codebuddy/mcp-for-xiaozhi/verify_pipe.py

"""
verify_pipe.py — 真实联调自检（接真小智AI）

替代旧的 mock-W 版本：mcp_pipe 改成 in-process 后，没有跨进程桥接可测了。
现在直接 spawn mcp_pipe.py 进程（不传 audio_server 脚本，因为它被 import 了），
让它连真实 xiaozhi.me，等握手完成 + tools/list round-trip + ping 一次。
"""
from __future__ import annotations

import os
import subprocess
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
PY = os.path.join(HERE, ".venv", "Scripts", "python.exe")
env_path = os.path.join(HERE, ".env")


def main() -> int:
    if not os.path.exists(env_path):
        print("FAIL: .env 不存在")
        return 1
    if not os.path.exists(PY):
        print(f"FAIL: 找不到 {PY}")
        return 1

    # 加载 .env
    with open(env_path, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, v = line.split("=", 1)
            os.environ.setdefault(k.strip(), v.strip().strip("'\""))

    if "MCP_ENDPOINT" not in os.environ or "your_token_here" in os.environ.get("MCP_ENDPOINT", ""):
        print("FAIL: .env 里 MCP_ENDPOINT 没填")
        return 1

    print(f"[verify_pipe] launching mcp_pipe.py (real xiaozhi.me) ...")
    proc = subprocess.Popen(
        [PY, "mcp_pipe.py"],
        cwd=HERE,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        env=os.environ,
        bufsize=0,
    )
    deadline = time.time() + 70  # xiaozhi 30s 后才发 ping，留足余量
    saw_init = False
    saw_tools_list = False
    saw_ping = False
    try:
        while time.time() < deadline:
            line = proc.stdout.readline()
            if not line:
                break
            text = line.decode("utf-8", errors="replace").rstrip()
            if not text:
                continue
            print(text)
            if "response for initialize" in text:
                saw_init = True
            elif "response for tools/list" in text:
                saw_tools_list = True
            elif "response for ping" in text:
                saw_ping = True
                break  # 一个 ping 就够了
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=3)
        except subprocess.TimeoutExpired:
            proc.kill()

    if saw_init and saw_tools_list and saw_ping:
        print("\n[verify_pipe] PASS: initialize + tools/list + ping all round-tripped")
        return 0
    print(f"\n[verify_pipe] FAIL: init={saw_init} tools_list={saw_tools_list} ping={saw_ping}")
    return 1


if __name__ == "__main__":
    sys.exit(main())