From 19526878b3299382da9a3f131eaaac3df85012c0 Mon Sep 17 00:00:00 2001 From: myifeng Date: Wed, 28 May 2025 11:49:18 +0800 Subject: [PATCH 1/7] =?UTF-8?q?=E6=8E=A5=E6=94=B6hello=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E4=B8=AD=E7=9A=84features=E4=BF=A1=E6=81=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 2 ++ main/xiaozhi-server/core/handle/helloHandle.py | 4 ++++ 2 files changed, 6 insertions(+) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 11ae6836..87c96ffe 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -142,6 +142,8 @@ class ConnectionHandler: ) # 在原来第一道关闭的基础上加60秒，进行二道关闭 self.audio_format = "opus" + # {"mcp":true} 表示启用MCP功能 + self.features = None async def handle_connection(self, ws): try: diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index 6fd40401..5dfc6b62 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -31,6 +31,10 @@ async def handleHelloMessage(conn, msg_json): if conn.asr is not None: conn.asr.set_audio_format(format) conn.welcome_msg["audio_params"] = audio_params + features = msg_json.get("features") + if features: + conn.logger.bind(tag=TAG).info(f"客户端特性: {features}") + conn.features = features await conn.websocket.send(json.dumps(conn.welcome_msg)) From d62b27814c4f1559b70441d27d14e426dced2236 Mon Sep 17 00:00:00 2001 From: myifeng Date: Wed, 28 May 2025 14:09:13 +0800 Subject: [PATCH 2/7] =?UTF-8?q?=E6=89=93=E5=8D=B0cmp=E7=B1=BB=E5=9E=8B?= =?UTF-8?q?=E6=B6=88=E6=81=AF=EF=BC=8C=E5=90=8E=E6=9C=9F=E9=80=82=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/handle/textHandle.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index da6ede47..33596b50 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -72,6 +72,8 @@ async def handleTextMessage(conn, message): asyncio.create_task(handleIotDescriptors(conn, msg_json["descriptors"])) if "states" in msg_json: asyncio.create_task(handleIotStatus(conn, msg_json["states"])) + elif msg_json["type"] == "mcp": + conn.logger.bind(tag=TAG).info(f"收到mcp消息：{message}") elif msg_json["type"] == "server": # 记录日志时过滤敏感信息 conn.logger.bind(tag=TAG).info( @@ -151,5 +153,7 @@ async def handleTextMessage(conn, message): # 重启服务器 elif msg_json["action"] == "restart": await conn.handle_restart(msg_json) + else: + conn.logger.bind(tag=TAG).error(f"收到未知类型消息：{message}") except json.JSONDecodeError: await conn.websocket.send(message) From d7f3b3caf52e16b08e33d19bbc254761b888d863 Mon Sep 17 00:00:00 2001 From: myifeng Date: Wed, 28 May 2025 18:32:15 +0800 Subject: [PATCH 3/7] =?UTF-8?q?=E6=8E=A5=E6=94=B6=E5=B9=B6=E6=89=93?= =?UTF-8?q?=E5=8D=B0MCP=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../xiaozhi-server/core/handle/helloHandle.py | 21 ++++++++++++++++ main/xiaozhi-server/core/handle/textHandle.py | 25 +++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index 5dfc6b62..ac97c566 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -35,6 +35,27 @@ async def handleHelloMessage(conn, msg_json): if features: conn.logger.bind(tag=TAG).info(f"客户端特性: {features}") conn.features = features + 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" + } + })) + # 发送mcp消息，获取tools列表 + await conn.websocket.send(json.dumps( { + "type": "mcp", + "payload": { + "jsonrpc": "2.0", + "method": "tools/list", + "id": 10001, + "params": { + } + } + })) await conn.websocket.send(json.dumps(conn.welcome_msg)) diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 33596b50..06964105 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -74,6 +74,31 @@ async def handleTextMessage(conn, message): asyncio.create_task(handleIotStatus(conn, msg_json["states"])) 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}") + elif msg_json["type"] == "server": # 记录日志时过滤敏感信息 conn.logger.bind(tag=TAG).info( From 8ea48f987565ee200ffb9be8a89ff6fb58912ff2 Mon Sep 17 00:00:00 2001 From: myifeng Date: Wed, 28 May 2025 19:28:36 +0800 Subject: [PATCH 4/7] =?UTF-8?q?=E7=8B=AC=E7=AB=8B=E5=A4=84=E7=90=86MCP=20M?= =?UTF-8?q?essage?= 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( From 4b212753fb677504c29f8a5b7368657ead2a0d4f Mon Sep 17 00:00:00 2001 From: myifeng Date: Thu, 29 May 2025 13:42:31 +0800 Subject: [PATCH 5/7] =?UTF-8?q?=E5=AF=B9=E6=8E=A5=E5=B0=8F=E6=99=BAMCP?= =?UTF-8?q?=E5=8D=8F=E8=AE=AE=E6=8E=A7=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 15 + .../xiaozhi-server/core/handle/helloHandle.py | 7 +- main/xiaozhi-server/core/handle/mcpHandle.py | 334 +++++++++++++----- main/xiaozhi-server/core/handle/textHandle.py | 4 +- 4 files changed, 262 insertions(+), 98 deletions(-) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 87c96ffe..6193ca99 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -10,6 +10,7 @@ import threading import traceback import subprocess import websockets +from core.handle.mcpHandle import call_mcp_tool from core.utils.util import ( extract_json_from_string, initialize_modules, @@ -560,6 +561,12 @@ class ConnectionHandler: functions = None if self.intent_type == "function_call" and hasattr(self, "func_handler"): functions = self.func_handler.get_functions() + if self.mcp_client is not None: + mcp_tools = self.mcp_client.get_available_tools() + if mcp_tools is not None and len(mcp_tools) > 0: + if functions is None: + functions = [] + functions.extend(mcp_tools) response_message = [] try: @@ -681,6 +688,14 @@ class ConnectionHandler: # 处理MCP工具调用 if self.mcp_manager.is_mcp_tool(function_name): result = self._handle_mcp_tool_call(function_call_data) + elif self.mcp_client is not None and self.mcp_client.has_tool(function_name): + # 如果是MCP工具调用 + self.logger.bind(tag=TAG).debug( + f"调用MCP工具: {function_name}, 参数: {function_arguments}" + ) + result = asyncio.run_coroutine_threadsafe(call_mcp_tool(self, self.mcp_client, function_name, function_arguments), self.loop).result() + self.logger.bind(tag=TAG).debug(f"MCP工具调用结果: {result}") + result = ActionResponse(action=Action.REQLLM, result=result, response="") else: # 处理系统函数 result = self.func_handler.handle_llm_function_call( diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index 75275c22..a38a33ce 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -4,7 +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.mcpHandle import MCPClient, send_mcp_initialize_message, send_mcp_tools_list_request 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,10 +38,11 @@ async def handleHelloMessage(conn, msg_json): conn.features = features if features.get("mcp"): conn.logger.bind(tag=TAG).info("客户端支持MCP") + conn.mcp_client = MCPClient() # 发送初始化 - asyncio.create_task(send_mcp_initialized(conn)) + asyncio.create_task(send_mcp_initialize_message(conn)) # 发送mcp消息，获取tools列表 - asyncio.create_task(send_mcp_tools_list(conn)) + asyncio.create_task(send_mcp_tools_list_request(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 index 426f5ada..2b574e92 100644 --- a/main/xiaozhi-server/core/handle/mcpHandle.py +++ b/main/xiaozhi-server/core/handle/mcpHandle.py @@ -1,78 +1,88 @@ - - import json +import asyncio +from concurrent.futures import Future TAG = __name__ -async def handleMCPMessage(conn, payload): - """处理MCP消息""" +class MCPClient: + """MCPClient，用于管理MCP状态和工具""" + def __init__(self): + self.tools = [] + self.ready = False + self.call_results = {} # To store Futures for tool call responses + self.next_id = 1 + self.lock = asyncio.Lock() + + async def has_tool(self, name: str) -> bool: + async with self.lock: + for tool in self.tools: + if tool["name"] == name: + return True + return False + + def get_available_tools(self) -> list: + # async with self.lock: + result = [] + for tool in self.tools: + function_def = { + "name": tool["name"], + "description": tool["description"], + "parameters": { + "type": tool["inputSchema"].get("type", "object"), + "properties": tool["inputSchema"].get("properties", {}), + "required": tool["inputSchema"].get("required", []) + } + } + result.append({"type": "function", "function": function_def}) + return result + + async def is_ready(self) -> bool: + async with self.lock: + return self.ready + + async def set_ready(self, status: bool): + async with self.lock: + self.ready = status + + async def add_tool(self, tool_data: dict): + async with self.lock: + self.tools.append(tool_data) + + async def get_next_id(self) -> int: + async with self.lock: + current_id = self.next_id + self.next_id += 1 + return current_id + + async def register_call_result_future(self, id: int, future: Future): + async with self.lock: + self.call_results[id] = future + + async def resolve_call_result(self, id: int, result: any): + async with self.lock: + if id in self.call_results: + future = self.call_results.pop(id) + if not future.done(): + future.set_result(result) + + async def reject_call_result(self, id: int, exception: Exception): + async with self.lock: + if id in self.call_results: + future = self.call_results.pop(id) + if not future.done(): + future.set_exception(exception) + + async def cleanup_call_result(self, id: int): + async with self.lock: + if id in self.call_results: + self.call_results.pop(id) + +async def send_mcp_message(conn, payload: dict): + """Helper to send MCP messages, encapsulating common logic.""" if not conn.features.get("mcp"): - conn.logger.bind(tag=TAG).warning("客户端不支持MCP，忽略消息") + conn.logger.bind(tag=TAG).warning("客户端不支持MCP，无法发送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 @@ -80,50 +90,188 @@ async def send_mcp_response(conn, payload): try: await conn.websocket.send(message) - conn.logger.bind(tag=__name__).info(f"成功发送MCP消息: {message}") + conn.logger.bind(tag=TAG).info(f"成功发送MCP消息: {message}") except Exception as e: - conn.logger.bind(tag=__name__).error(f"发送MCP消息失败: {e}") + conn.logger.bind(tag=TAG).error(f"发送MCP消息失败: {e}") -async def send_mcp_initialized(conn): - """发送MCP初始化通知""" - if not conn.features.get("mcp"): - conn.logger.bind(tag=__name__).warning("客户端不支持MCP，无法发送初始化通知") +async def handle_mcp_message(conn, mcp_client: MCPClient, payload: dict): + """处理MCP消息,包括初始化、工具列表和工具调用响应等""" + conn.logger.bind(tag=TAG).info(f"处理MCP消息: {payload}") + + if not isinstance(payload, dict): + conn.logger.bind(tag=TAG).error("MCP消息缺少payload字段或格式错误") return + # Handle result + if "result" in payload: + result = payload["result"] + msg_id = int(payload.get("id", 0)) + + # Check for tool call response first + if msg_id in mcp_client.call_results: + conn.logger.bind(tag=TAG).info(f"收到工具调用响应，ID: {msg_id}, 结果: {result}") + await mcp_client.resolve_call_result(msg_id, result) + return + + if msg_id == 1: # mcpInitializeID + conn.logger.bind(tag=TAG).debug("收到MCP初始化响应") + server_info = result.get("serverInfo") + if isinstance(server_info, dict): + name = server_info.get("name") + version = server_info.get("version") + conn.logger.bind(tag=TAG).info(f"客户端MCP服务器信息: name={name}, version={version}") + await send_mcp_tools_list_request(conn) # After initialization, request tool list + return + + elif msg_id == 2: # mcpToolsListID + conn.logger.bind(tag=TAG).debug("收到MCP工具列表响应") + if isinstance(result, dict) and "tools" in result: + tools_data = result["tools"] + if not isinstance(tools_data, list): + conn.logger.bind(tag=TAG).error("工具列表格式错误") + return + + conn.logger.bind(tag=TAG).info(f"客户端设备支持的工具数量: {len(tools_data)}") + + for i, tool in enumerate(tools_data): + if not isinstance(tool, dict): + continue + + name = tool.get("name", "") + description = tool.get("description", "") + input_schema = {"type": "object", "properties": {}, "required": []} + + if "inputSchema" in tool and isinstance(tool["inputSchema"], dict): + schema = tool["inputSchema"] + input_schema["type"] = schema.get("type", "object") + input_schema["properties"] = schema.get("properties", {}) + input_schema["required"] = [ + s for s in schema.get("required", []) if isinstance(s, str) + ] + + new_tool = { + "name": name, + "description": description, + "inputSchema": input_schema, + } + await mcp_client.add_tool(new_tool) + conn.logger.bind(tag=TAG).debug(f"客户端工具 #{i+1}: {name}") + + next_cursor = result.get("nextCursor", "") + if next_cursor: + conn.logger.bind(tag=TAG).info(f"有更多工具，nextCursor: {next_cursor}") + await send_mcp_tools_list_continue_request(conn, next_cursor) + else: + await mcp_client.set_ready(True) + conn.logger.bind(tag=TAG).info("所有工具已获取，MCP客户端准备就绪") + tool_result = await call_mcp_tool(conn, mcp_client, "self.get_device_status", {}) + print(f"Tool call result: {tool_result}") + return + + # Handle method calls (requests from the client) + elif "method" in payload: + method = payload["method"] + conn.logger.bind(tag=TAG).info(f"收到MCP客户端请求: {method}") + + elif "error" in payload: + error_data = payload["error"] + error_msg = error_data.get("message", "未知错误") + conn.logger.bind(tag=TAG).error(f"收到MCP错误响应: {error_msg}") + + msg_id = int(payload.get("id", 0)) + if msg_id in mcp_client.call_results: + await mcp_client.reject_call_result(msg_id, Exception(f"MCP错误: {error_msg}")) + + +# --- Outgoing MCP Messages --- + +async def send_mcp_initialize_message(conn): + """发送MCP初始化消息""" payload = { "jsonrpc": "2.0", - "method": "notifications/initialized" + "id": 1, # mcpInitializeID + "method": "initialize", + "params": { + "protocolVersion": "2024-11-05", + "capabilities": { + "roots": {"listChanged": True}, + "sampling": {}, + }, + "clientInfo": { + "name": "XiaozhiClient", + "version": "1.0.0", + }, + }, } + conn.logger.bind(tag=TAG).info("发送MCP初始化消息") + await send_mcp_message(conn, payload) - 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 - +async def send_mcp_tools_list_request(conn): + """发送MCP工具列表请求""" payload = { "jsonrpc": "2.0", + "id": 2, # mcpToolsListID "method": "tools/list", - "params": {} } + conn.logger.bind(tag=TAG).debug("发送MCP工具列表请求") + await send_mcp_message(conn, payload) - await send_mcp_response(conn, payload) +async def send_mcp_tools_list_continue_request(conn, cursor: str): + """发送带有cursor的MCP工具列表请求""" + payload = { + "jsonrpc": "2.0", + "id": 2, # mcpToolsListID (same ID for continuation) + "method": "tools/list", + "params": {"cursor": cursor}, + } + conn.logger.bind(tag=TAG).info(f"发送带cursor的MCP工具列表请求: {cursor}") + await send_mcp_message(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 +async def call_mcp_tool(conn, mcp_client: MCPClient, tool_name: str, args: dict, timeout: int = 30): + """ + 调用指定的工具，并等待响应 + """ + if not await mcp_client.is_ready(): + raise RuntimeError("MCP客户端尚未准备就绪") + + if not await mcp_client.has_tool(tool_name): + raise ValueError(f"工具 {tool_name} 不存在") + + tool_call_id = await mcp_client.get_next_id() + result_future = asyncio.Future() + await mcp_client.register_call_result_future(tool_call_id, result_future) payload = { "jsonrpc": "2.0", + "id": tool_call_id, "method": "tools/call", "params": { "name": tool_name, - "arguments": params - } + "arguments": args, + }, } - await send_mcp_response(conn, payload) \ No newline at end of file + conn.logger.bind(tag=TAG).info(f"发送客户端mcp工具调用请求: {tool_name}，参数: {args}") + await send_mcp_message(conn, payload) + + try: + # Wait for response or timeout + raw_result = await asyncio.wait_for(result_future, timeout=timeout) + conn.logger.bind(tag=TAG).info(f"客户端mcp工具调用 {tool_name} 成功，原始结果: {raw_result}") + + if isinstance(raw_result, dict): + if raw_result.get("isError") is True: + error_msg = raw_result.get("error", "工具调用返回错误，但未提供具体错误信息") + raise RuntimeError(f"工具调用错误: {error_msg}") + + content = raw_result.get("content") + if isinstance(content, list) and len(content) > 0: + if isinstance(content[0], dict) and "text" in content[0]: + return content[0]["text"] + return raw_result + except asyncio.TimeoutError: + await mcp_client.cleanup_call_result(tool_call_id) + raise TimeoutError("工具调用请求超时") + except Exception as e: + await mcp_client.cleanup_call_result(tool_call_id) + raise e diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 5babe4a2..37e5e60c 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -1,7 +1,7 @@ import json from core.handle.abortHandle import handleAbortMessage from core.handle.helloHandle import handleHelloMessage -from core.handle.mcpHandle import handleMCPMessage +from core.handle.mcpHandle import handle_mcp_message 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 @@ -76,7 +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: - asyncio.create_task(handleMCPMessage(conn, msg_json["payload"])) + asyncio.create_task(handle_mcp_message(conn, conn.mcp_client, msg_json["payload"])) elif msg_json["type"] == "server": # 记录日志时过滤敏感信息 conn.logger.bind(tag=TAG).info( From 69520391f2a0df9aa88bdb9a6852daffdc78495a Mon Sep 17 00:00:00 2001 From: myifeng Date: Thu, 29 May 2025 15:41:18 +0800 Subject: [PATCH 6/7] =?UTF-8?q?=E5=8F=82=E6=95=B0=E4=B8=BAJSON?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/handle/mcpHandle.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/main/xiaozhi-server/core/handle/mcpHandle.py b/main/xiaozhi-server/core/handle/mcpHandle.py index 2b574e92..838fb6bd 100644 --- a/main/xiaozhi-server/core/handle/mcpHandle.py +++ b/main/xiaozhi-server/core/handle/mcpHandle.py @@ -109,7 +109,7 @@ async def handle_mcp_message(conn, mcp_client: MCPClient, payload: dict): # Check for tool call response first if msg_id in mcp_client.call_results: - conn.logger.bind(tag=TAG).info(f"收到工具调用响应，ID: {msg_id}, 结果: {result}") + conn.logger.bind(tag=TAG).debug(f"收到工具调用响应，ID: {msg_id}, 结果: {result}") await mcp_client.resolve_call_result(msg_id, result) return @@ -227,7 +227,7 @@ async def send_mcp_tools_list_continue_request(conn, cursor: str): conn.logger.bind(tag=TAG).info(f"发送带cursor的MCP工具列表请求: {cursor}") await send_mcp_message(conn, payload) -async def call_mcp_tool(conn, mcp_client: MCPClient, tool_name: str, args: dict, timeout: int = 30): +async def call_mcp_tool(conn, mcp_client: MCPClient, tool_name: str, args: str = '{}', timeout: int = 30): """ 调用指定的工具，并等待响应 """ @@ -247,7 +247,7 @@ async def call_mcp_tool(conn, mcp_client: MCPClient, tool_name: str, args: dict, "method": "tools/call", "params": { "name": tool_name, - "arguments": args, + "arguments": json.loads(args) if isinstance(args, str) else args }, } From c487f69676cadd81b970409fe7f65377aa2b1501 Mon Sep 17 00:00:00 2001 From: myifeng Date: Thu, 29 May 2025 17:55:01 +0800 Subject: [PATCH 7/7] =?UTF-8?q?FIX=20=E4=B8=BA=E7=A9=BA=E5=88=A4=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../plugins_func/functions/handle_speaker_or_screen.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/main/xiaozhi-server/plugins_func/functions/handle_speaker_or_screen.py b/main/xiaozhi-server/plugins_func/functions/handle_speaker_or_screen.py index f89f30b0..8547c985 100644 --- a/main/xiaozhi-server/plugins_func/functions/handle_speaker_or_screen.py +++ b/main/xiaozhi-server/plugins_func/functions/handle_speaker_or_screen.py @@ -102,11 +102,11 @@ handle_device_function_desc = { } -@register_function( - "handle_speaker_volume_or_screen_brightness", - handle_device_function_desc, - ToolType.IOT_CTL, -) +# @register_function( +# "handle_speaker_volume_or_screen_brightness", +# handle_device_function_desc, +# ToolType.IOT_CTL, +# ) def handle_speaker_volume_or_screen_brightness( conn, device_type: str, action: str, value: int = None ):