audio_codebuddy/mcp-for-xiaozhi/verify_local.py

"""
本地验证脚本：直接调用 audio_server 里注册的 MCP 工具（不经过 WebSocket）
============================================================

用法（在 mcp-for-xiaozhi 目录下）：
    python verify_local.py

要求：
    - 后端 audiobook 服务跑在 AUDIOBOOK_API_BASE（默认 127.0.0.1:3000）
    - 已安装依赖：pip install -r requirements.txt

它会：
    1. import audio_server 注册到 FastMCP 实例的内部 tool 列表
    2. 通过 mcp.ClientSession 用 stdio 启动 audio_server，列工具 + 调用若干工具
"""
from __future__ import annotations

import asyncio
import os
import sys

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


async def main() -> int:
    here = os.path.dirname(os.path.abspath(__file__))
    server_path = os.path.join(here, "audio_server.py")
    api_base = os.environ.get("AUDIOBOOK_API_BASE", "http://127.0.0.1:3000")
    print(f"[verify] api_base = {api_base}")

    params = StdioServerParameters(
        command=sys.executable,
        args=[server_path],
        env={**os.environ, "MCP_LOG_LEVEL": "WARNING"},
    )

    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools = await session.list_tools()
            print(f"[verify] list_tools -> {len(tools.tools)} tools")
            for t in tools.tools:
                print(f"  - {t.name}: {t.description[:60] if t.description else ''}")

            # 用例 1：搜书
            r = await session.call_tool("search_audiobooks", {"keywords": "三国"})
            print("\n[verify] search_audiobooks('三国') ->")
            print((r.content[0].text if r.content else "")[:400])

            # 用例 2：分类
            r = await session.call_tool("list_categories", {})
            print("\n[verify] list_categories ->")
            print((r.content[0].text if r.content else "")[:400])

            # 用例 3：详情（用第一个搜索结果的 id）
            try:
                import json
                search_text = (await session.call_tool("search_audiobooks", {"keywords": "三国"})).content[0].text
                payload = json.loads(search_text)
                first_id = (payload.get("items") or [{}])[0].get("id")
                if first_id:
                    r = await session.call_tool("get_book_details", {"book_id": int(first_id)})
                    print(f"\n[verify] get_book_details({first_id}) ->")
                    print((r.content[0].text if r.content else "")[:400])
            except Exception as e:  # noqa: BLE001
                print(f"[verify] get_book_details skipped: {e}")

    print("\n[verify] OK")
    return 0


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