From 8ea48f987565ee200ffb9be8a89ff6fb58912ff2 Mon Sep 17 00:00:00 2001 From: myifeng Date: Wed, 28 May 2025 19:28:36 +0800 Subject: [PATCH] =?UTF-8?q?=E7=8B=AC=E7=AB=8B=E5=A4=84=E7=90=86MCP=20Messa?= =?UTF-8?q?ge?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../xiaozhi-server/core/handle/helloHandle.py | 20 +-- main/xiaozhi-server/core/handle/mcpHandle.py | 129 ++++++++++++++++++ main/xiaozhi-server/core/handle/textHandle.py | 26 +--- 3 files changed, 134 insertions(+), 41 deletions(-) create mode 100644 main/xiaozhi-server/core/handle/mcpHandle.py diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index ac97c566..75275c22 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -4,6 +4,7 @@ import json import random import shutil import asyncio +from core.handle.mcpHandle import send_mcp_initialized, send_mcp_tools_list from core.handle.sendAudioHandle import send_stt_message from core.utils.util import remove_punctuation_and_length from core.providers.tts.dto.dto import ContentType, InterfaceType @@ -38,24 +39,9 @@ async def handleHelloMessage(conn, msg_json): if features.get("mcp"): conn.logger.bind(tag=TAG).info("客户端支持MCP") # 发送初始化 - await conn.websocket.send(json.dumps({ - "type": "mcp", - "payload": { - "jsonrpc": "2.0", - "method": "notifications/initialized" - } - })) + asyncio.create_task(send_mcp_initialized(conn)) # 发送mcp消息，获取tools列表 - await conn.websocket.send(json.dumps( { - "type": "mcp", - "payload": { - "jsonrpc": "2.0", - "method": "tools/list", - "id": 10001, - "params": { - } - } - })) + asyncio.create_task(send_mcp_tools_list(conn)) await conn.websocket.send(json.dumps(conn.welcome_msg)) diff --git a/main/xiaozhi-server/core/handle/mcpHandle.py b/main/xiaozhi-server/core/handle/mcpHandle.py new file mode 100644 index 00000000..426f5ada --- /dev/null +++ b/main/xiaozhi-server/core/handle/mcpHandle.py @@ -0,0 +1,129 @@ + + +import json + +TAG = __name__ + +async def handleMCPMessage(conn, payload): + """处理MCP消息""" + if not conn.features.get("mcp"): + conn.logger.bind(tag=TAG).warning("客户端不支持MCP，忽略消息") + return + + if payload.get("jsonrpc") != "2.0": + conn.logger.bind(tag=TAG).error("MCP消息格式错误，缺少jsonrpc字段") + return + + if "id" not in payload or not isinstance(payload["id"], int): + # MCP消息必须包含id字段，且id必须是整数 + conn.logger.bind(tag=TAG).error("MCP消息格式错误，缺少id字段") + return + + # payload必须包含result,而result中包含tools、content、isError（默认False） + if "result" not in payload: + conn.logger.bind(tag=TAG).error("MCP消息格式错误，缺少result字段") + return + # tools是一个列表，包含所有可用的工具,tools中每个工具必须包含name和description字段和inputSchema字段，如果有tools则处理 + if "tools" in payload["result"]: + tools = payload["result"]["tools"] + if not isinstance(tools, list): + conn.logger.bind(tag=TAG).error("MCP消息格式错误，tools字段不是列表") + return + for tool in tools: + if not isinstance(tool, dict) or "name" not in tool or "description" not in tool: + conn.logger.bind(tag=TAG).error("MCP消息格式错误，tools中的工具缺少name或description字段") + return + # 处理工具注册逻辑 + + conn.logger.bind(tag=TAG).info(f"注册了{len(tools)}个MCP工具") + # content是一个数组，每个元素包含type和text,包含消息内容，是回复mcp的内容，如果有则进行处理 + if "content" in payload["result"]: + content = payload["result"]["content"] + if not isinstance(content, list): + conn.logger.bind(tag=TAG).error("MCP消息格式错误，content字段不是列表") + return + for item in content: + if not isinstance(item, dict) or "type" not in item or "text" not in item: + conn.logger.bind(tag=TAG).error("MCP消息格式错误，content中的元素缺少type或text字段") + return + if item["type"] == "text": + # 处理文本消息 + pass # TODO: Implement text message handling + +async def send_mcp_response(conn, payload): + """发送MCP响应""" + if not conn.features.get("mcp"): + conn.logger.bind(tag=__name__).warning("客户端不支持MCP，无法发送MCP消息") + return + + # payload中id必须是整数，必须递增，如果传递了id，z则使用id,否则需要生成一个id + id = payload.get("id") + if id is not None and not isinstance(id, int): + conn.logger.bind(tag=__name__).error("MCP消息ID必须是整数") + return + if id is None: + # 生成一个新的ID，通常是递增的， mcp_next_id可能为空,需要考虑并发 + if not hasattr(conn, 'mcp_next_id'): + conn.mcp_next_id = 1 + id = conn.mcp_next_id + conn.mcp_next_id += 1 + elif not isinstance(id, int): + conn.logger.bind(tag=__name__).error("MCP消息ID必须是整数") + return + + payload["id"] = id + # MCP消息格式为JSON字符串 + message = json.dumps({ + "type": "mcp", + "payload": payload + }) + + try: + await conn.websocket.send(message) + conn.logger.bind(tag=__name__).info(f"成功发送MCP消息: {message}") + except Exception as e: + conn.logger.bind(tag=__name__).error(f"发送MCP消息失败: {e}") + +async def send_mcp_initialized(conn): + """发送MCP初始化通知""" + if not conn.features.get("mcp"): + conn.logger.bind(tag=__name__).warning("客户端不支持MCP，无法发送初始化通知") + return + + payload = { + "jsonrpc": "2.0", + "method": "notifications/initialized" + } + + await send_mcp_response(conn, payload) + +async def send_mcp_tools_list(conn): + """发送MCP工具列表""" + if not conn.features.get("mcp"): + conn.logger.bind(tag=__name__).warning("客户端不支持MCP，无法发送工具列表") + return + + payload = { + "jsonrpc": "2.0", + "method": "tools/list", + "params": {} + } + + await send_mcp_response(conn, payload) + +async def send_mcp_tool_call(conn, tool_name, params): + """发送MCP工具调用请求""" + if not conn.features.get("mcp"): + conn.logger.bind(tag=__name__).warning("客户端不支持MCP，无法发送工具调用请求") + return + + payload = { + "jsonrpc": "2.0", + "method": "tools/call", + "params": { + "name": tool_name, + "arguments": params + } + } + + await send_mcp_response(conn, payload) \ No newline at end of file diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 06964105..5babe4a2 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -1,6 +1,7 @@ import json from core.handle.abortHandle import handleAbortMessage from core.handle.helloHandle import handleHelloMessage +from core.handle.mcpHandle import handleMCPMessage from core.utils.util import remove_punctuation_and_length, filter_sensitive_info from core.handle.receiveAudioHandle import startToChat, handleAudioMessage from core.handle.sendAudioHandle import send_stt_message, send_tts_message @@ -75,30 +76,7 @@ async def handleTextMessage(conn, message): elif msg_json["type"] == "mcp": conn.logger.bind(tag=TAG).info(f"收到mcp消息：{message}") if "payload" in msg_json: - payload = msg_json["payload"] - id = payload.get("id", None) - if id is None: - conn.logger.bind(tag=TAG).error("MCP消息缺少ID") - return - # result: 包含工具列表和上下文 - # TODO: 处理MCP消息时，进行不同的处理 - if "jsonrpc" not in payload or payload["jsonrpc"] != "2.0": - conn.logger.bind(tag=TAG).error("MCP消息缺少jsonrpc版本") - return - if "result" in payload: - result = payload["result"] - if "tools" in result: - # 初始化处理工具列表 - tools = result["tools"] - conn.features["mcp"] = True - # 设置工具列表到连接对象，方便后续使用 - conn.features["mcp_tools"] = tools - conn.logger.bind(tag=TAG).info(f"收到MCP工具列表：{tools}") - elif "context" in result: - # 处理上下文,上下文根据id进行管理，上报的id对应Server下发时的id - context = result["context"] - conn.logger.bind(tag=TAG).info(f"收到MCP上下文：{context}, ID: {id}") - + asyncio.create_task(handleMCPMessage(conn, msg_json["payload"])) elif msg_json["type"] == "server": # 记录日志时过滤敏感信息 conn.logger.bind(tag=TAG).info(