From 4b212753fb677504c29f8a5b7368657ead2a0d4f Mon Sep 17 00:00:00 2001 From: myifeng Date: Thu, 29 May 2025 13:42:31 +0800 Subject: [PATCH] =?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(