audio_codebuddy/mcp-for-xiaozhi/mcp_server_local.py

"""
mcp_server_local.py — 本地小智服务端直连 MCP Server
===================================================

运行一个 WebSocket MCP 服务器，本地 xiaozhi 服务端直接连接（不走云端 xz-mcp-broker）。

启动后，把地址 ws://192.168.31.155:8765 填到智控台的 MCP 接入点即可。
"""
from __future__ import annotations

import asyncio
import json
import logging
import os
import sys

import websockets
from websockets.asyncio.server import serve

import audio_server

logger = logging.getLogger("MCP_LOCAL")


async def handle_ws(ws):
    """处理一个 xiaozhi MCP 客户端连接"""
    logger.info("xiaozhi 客户端已连接")
    try:
        async for raw in ws:
            if isinstance(raw, bytes):
                raw = raw.decode("utf-8", errors="replace")
            logger.info("<- %s", raw[:200])
            try:
                req = json.loads(raw)
            except Exception as e:
                logger.warning("parse error: %s", e)
                continue
            rid = req.get("id")
            method = req.get("method")
            if rid is None and method and method.startswith("notifications/"):
                continue
            resp = audio_server.handle_request(req)
            if resp is None:
                continue
            line = json.dumps(resp, ensure_ascii=False, separators=(",", ":")) + "\n"
            logger.info("-> %s", line[:200])
            await ws.send(line)
    except websockets.ConnectionClosed:
        logger.info("客户端断开")


async def main():
    port = int(os.environ.get("MCP_LOCAL_PORT", "8765"))
    logging.basicConfig(level="INFO", format="%(asctime)s - %(levelname)s - %(message)s")
    logger.info(f"MCP Local Server 启动在 ws://0.0.0.0:{port}")
    logger.info(f"API base = {audio_server.API_BASE}")
    async with serve(handle_ws, "0.0.0.0", port):
        await asyncio.Future()


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