diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 6dc0b25d..71c13ec7 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -104,7 +104,8 @@ wakeup_words: - "喵喵同学" - "小滨小滨" - "小冰小冰" - +# MCP接入点地址 +mcp_endpoint: 你的接入点 websocket地址 # 插件的基础配置 plugins: # 获取天气插件的配置，这里填写你的api_key diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 9bce2ca4..cdb20ac5 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -77,7 +77,7 @@ async def handleTextMessage(conn, message): 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}") + conn.logger.bind(tag=TAG).info(f"收到mcp消息：{message[:100]}") if "payload" in msg_json: asyncio.create_task( handle_mcp_message(conn, conn.mcp_client, msg_json["payload"]) diff --git a/main/xiaozhi-server/core/providers/tools/base/tool_types.py b/main/xiaozhi-server/core/providers/tools/base/tool_types.py index 0f86d99e..91466fad 100644 --- a/main/xiaozhi-server/core/providers/tools/base/tool_types.py +++ b/main/xiaozhi-server/core/providers/tools/base/tool_types.py @@ -14,6 +14,7 @@ class ToolType(Enum): SERVER_MCP = "server_mcp" # 服务端MCP DEVICE_IOT = "device_iot" # 设备端IoT DEVICE_MCP = "device_mcp" # 设备端MCP + MCP_ENDPOINT = "mcp_endpoint" # MCP接入点 @dataclass diff --git a/main/xiaozhi-server/core/providers/tools/device_mcp/mcp_handler.py b/main/xiaozhi-server/core/providers/tools/device_mcp/mcp_handler.py index 42a69c0f..d5f42833 100644 --- a/main/xiaozhi-server/core/providers/tools/device_mcp/mcp_handler.py +++ b/main/xiaozhi-server/core/providers/tools/device_mcp/mcp_handler.py @@ -113,7 +113,7 @@ async def send_mcp_message(conn, payload: dict): async def handle_mcp_message(conn, mcp_client: MCPClient, payload: dict): """处理MCP消息,包括初始化、工具列表和工具调用响应等""" - logger.bind(tag=TAG).info(f"处理MCP消息: {payload}") + logger.bind(tag=TAG).info(f"处理MCP消息: {str(payload)[:100]}") if not isinstance(payload, dict): logger.bind(tag=TAG).error("MCP消息缺少payload字段或格式错误") diff --git a/main/xiaozhi-server/core/providers/tools/mcp_endpoint/__init__.py b/main/xiaozhi-server/core/providers/tools/mcp_endpoint/__init__.py new file mode 100644 index 00000000..de2d9b1c --- /dev/null +++ b/main/xiaozhi-server/core/providers/tools/mcp_endpoint/__init__.py @@ -0,0 +1,21 @@ +"""MCP接入点工具模块""" + +from .mcp_endpoint_executor import MCPEndpointExecutor +from .mcp_endpoint_client import MCPEndpointClient +from .mcp_endpoint_handler import ( + connect_mcp_endpoint, + send_mcp_endpoint_initialize, + send_mcp_endpoint_notification, + send_mcp_endpoint_tools_list, + call_mcp_endpoint_tool, +) + +__all__ = [ + "MCPEndpointExecutor", + "MCPEndpointClient", + "connect_mcp_endpoint", + "send_mcp_endpoint_initialize", + "send_mcp_endpoint_notification", + "send_mcp_endpoint_tools_list", + "call_mcp_endpoint_tool", +] diff --git a/main/xiaozhi-server/core/providers/tools/mcp_endpoint/mcp_endpoint_client.py b/main/xiaozhi-server/core/providers/tools/mcp_endpoint/mcp_endpoint_client.py new file mode 100644 index 00000000..da8241d5 --- /dev/null +++ b/main/xiaozhi-server/core/providers/tools/mcp_endpoint/mcp_endpoint_client.py @@ -0,0 +1,111 @@ +"""MCP接入点客户端定义""" + +import asyncio +from concurrent.futures import Future +from core.utils.util import sanitize_tool_name +from config.logger import setup_logging + +TAG = __name__ +logger = setup_logging() + + +class MCPEndpointClient: + """MCP接入点客户端，用于管理MCP接入点状态和工具""" + + def __init__(self): + self.tools = {} # sanitized_name -> tool_data + self.name_mapping = {} + self.ready = False + self.call_results = {} # To store Futures for tool call responses + self.next_id = 1 + self.lock = asyncio.Lock() + self._cached_available_tools = None # Cache for get_available_tools + self.websocket = None # WebSocket连接 + + def has_tool(self, name: str) -> bool: + return name in self.tools + + def get_available_tools(self) -> list: + # Check if the cache is valid + if self._cached_available_tools is not None: + return self._cached_available_tools + + # If cache is not valid, regenerate the list + result = [] + for tool_name, tool_data in self.tools.items(): + function_def = { + "name": tool_name, + "description": tool_data["description"], + "parameters": { + "type": tool_data["inputSchema"].get("type", "object"), + "properties": tool_data["inputSchema"].get("properties", {}), + "required": tool_data["inputSchema"].get("required", []), + }, + } + result.append({"type": "function", "function": function_def}) + + self._cached_available_tools = result # Store the generated list in cache + 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: + sanitized_name = sanitize_tool_name(tool_data["name"]) + self.tools[sanitized_name] = tool_data + self.name_mapping[sanitized_name] = tool_data["name"] + self._cached_available_tools = ( + None # Invalidate the cache when a tool is added + ) + + 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) + + def set_websocket(self, websocket): + """设置WebSocket连接""" + self.websocket = websocket + + async def send_message(self, message: str): + """发送消息到MCP接入点""" + if self.websocket: + await self.websocket.send(message) + else: + raise RuntimeError("WebSocket连接未建立") + + async def close(self): + """关闭WebSocket连接""" + if self.websocket: + await self.websocket.close() + self.websocket = None diff --git a/main/xiaozhi-server/core/providers/tools/mcp_endpoint/mcp_endpoint_executor.py b/main/xiaozhi-server/core/providers/tools/mcp_endpoint/mcp_endpoint_executor.py new file mode 100644 index 00000000..f2c9bac1 --- /dev/null +++ b/main/xiaozhi-server/core/providers/tools/mcp_endpoint/mcp_endpoint_executor.py @@ -0,0 +1,97 @@ +"""MCP接入点工具执行器""" + +from typing import Dict, Any +from ..base import ToolType, ToolDefinition, ToolExecutor +from plugins_func.register import Action, ActionResponse +from .mcp_endpoint_handler import call_mcp_endpoint_tool + + +class MCPEndpointExecutor(ToolExecutor): + """MCP接入点工具执行器""" + + def __init__(self, conn): + self.conn = conn + + async def execute( + self, conn, tool_name: str, arguments: Dict[str, Any] + ) -> ActionResponse: + """执行MCP接入点工具""" + if not hasattr(conn, "mcp_endpoint_client") or not conn.mcp_endpoint_client: + return ActionResponse( + action=Action.ERROR, + response="MCP接入点客户端未初始化", + ) + + if not await conn.mcp_endpoint_client.is_ready(): + return ActionResponse( + action=Action.ERROR, + response="MCP接入点客户端未准备就绪", + ) + + try: + # 转换参数为JSON字符串 + import json + + args_str = json.dumps(arguments) if arguments else "{}" + + # 调用MCP接入点工具 + result = await call_mcp_endpoint_tool( + conn.mcp_endpoint_client, tool_name, args_str + ) + + resultJson = None + if isinstance(result, str): + try: + resultJson = json.loads(result) + except Exception as e: + pass + + # 视觉大模型不经过二次LLM处理 + if ( + resultJson is not None + and isinstance(resultJson, dict) + and "action" in resultJson + ): + return ActionResponse( + action=Action[resultJson["action"]], + response=resultJson.get("response", ""), + ) + + return ActionResponse(action=Action.REQLLM, result=str(result)) + + except ValueError as e: + return ActionResponse(action=Action.NOTFOUND, response=str(e)) + except Exception as e: + return ActionResponse(action=Action.ERROR, response=str(e)) + + def get_tools(self) -> Dict[str, ToolDefinition]: + """获取所有MCP接入点工具""" + if ( + not hasattr(self.conn, "mcp_endpoint_client") + or not self.conn.mcp_endpoint_client + ): + return {} + + tools = {} + mcp_tools = self.conn.mcp_endpoint_client.get_available_tools() + + for tool in mcp_tools: + func_def = tool.get("function", {}) + tool_name = func_def.get("name", "") + + if tool_name: + tools[tool_name] = ToolDefinition( + name=tool_name, description=tool, tool_type=ToolType.MCP_ENDPOINT + ) + + return tools + + def has_tool(self, tool_name: str) -> bool: + """检查是否有指定的MCP接入点工具""" + if ( + not hasattr(self.conn, "mcp_endpoint_client") + or not self.conn.mcp_endpoint_client + ): + return False + + return self.conn.mcp_endpoint_client.has_tool(tool_name) diff --git a/main/xiaozhi-server/core/providers/tools/mcp_endpoint/mcp_endpoint_handler.py b/main/xiaozhi-server/core/providers/tools/mcp_endpoint/mcp_endpoint_handler.py new file mode 100644 index 00000000..bf3253f0 --- /dev/null +++ b/main/xiaozhi-server/core/providers/tools/mcp_endpoint/mcp_endpoint_handler.py @@ -0,0 +1,365 @@ +"""MCP接入点处理器""" + +import json +import asyncio +import re +import websockets +from concurrent.futures import Future +from core.utils.util import sanitize_tool_name +from config.logger import setup_logging +from .mcp_endpoint_client import MCPEndpointClient + +TAG = __name__ +logger = setup_logging() + + +async def connect_mcp_endpoint(mcp_endpoint_url: str) -> MCPEndpointClient: + """连接到MCP接入点""" + if not mcp_endpoint_url: + logger.bind(tag=TAG).warning("MCP接入点URL为空，跳过连接") + return None + + try: + logger.bind(tag=TAG).info(f"正在连接到MCP接入点: {mcp_endpoint_url}") + websocket = await websockets.connect(mcp_endpoint_url) + + mcp_client = MCPEndpointClient() + mcp_client.set_websocket(websocket) + + # 启动消息监听器 + asyncio.create_task(_message_listener(mcp_client)) + + # 发送初始化消息 + await send_mcp_endpoint_initialize(mcp_client) + + # 发送初始化完成通知 + await send_mcp_endpoint_notification(mcp_client, "notifications/initialized") + + # 获取工具列表 + await send_mcp_endpoint_tools_list(mcp_client) + + logger.bind(tag=TAG).info("MCP接入点连接成功") + return mcp_client + + except Exception as e: + logger.bind(tag=TAG).error(f"连接MCP接入点失败: {e}") + return None + + +async def _message_listener(mcp_client: MCPEndpointClient): + """监听MCP接入点消息""" + try: + async for message in mcp_client.websocket: + await handle_mcp_endpoint_message(mcp_client, message) + except websockets.exceptions.ConnectionClosed: + logger.bind(tag=TAG).info("MCP接入点连接已关闭") + except Exception as e: + logger.bind(tag=TAG).error(f"MCP接入点消息监听器错误: {e}") + finally: + await mcp_client.set_ready(False) + + +async def handle_mcp_endpoint_message(mcp_client: MCPEndpointClient, message: str): + """处理MCP接入点消息""" + try: + payload = json.loads(message) + logger.bind(tag=TAG).debug(f"收到MCP接入点消息: {payload}") + + if not isinstance(payload, dict): + logger.bind(tag=TAG).error("MCP接入点消息格式错误") + return + + # Handle result + if "result" in payload: + result = payload["result"] + # 安全地获取消息ID，如果为None则使用0 + msg_id_raw = payload.get("id") + msg_id = int(msg_id_raw) if msg_id_raw is not None else 0 + + # Check for tool call response first + if msg_id in mcp_client.call_results: + logger.bind(tag=TAG).debug( + f"收到工具调用响应，ID: {msg_id}, 结果: {result}" + ) + await mcp_client.resolve_call_result(msg_id, result) + return + + if msg_id == 1: # mcpInitializeID + 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") + logger.bind(tag=TAG).info( + f"MCP接入点服务器信息: name={name}, version={version}" + ) + return + + elif msg_id == 2: # mcpToolsListID + logger.bind(tag=TAG).debug("收到MCP接入点工具列表响应") + if isinstance(result, dict) and "tools" in result: + tools_data = result["tools"] + if not isinstance(tools_data, list): + logger.bind(tag=TAG).error("工具列表格式错误") + return + + logger.bind(tag=TAG).info( + f"MCP接入点支持的工具数量: {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) + logger.bind(tag=TAG).debug(f"MCP接入点工具 #{i+1}: {name}") + + # 替换所有工具描述中的工具名称 + for tool_data in mcp_client.tools.values(): + if "description" in tool_data: + description = tool_data["description"] + # 遍历所有工具名称进行替换 + for ( + sanitized_name, + original_name, + ) in mcp_client.name_mapping.items(): + description = description.replace( + original_name, sanitized_name + ) + tool_data["description"] = description + + next_cursor = result.get("nextCursor", "") + if next_cursor: + logger.bind(tag=TAG).info( + f"有更多工具，nextCursor: {next_cursor}" + ) + await send_mcp_endpoint_tools_list_continue( + mcp_client, next_cursor + ) + else: + await mcp_client.set_ready(True) + logger.bind(tag=TAG).info( + "所有MCP接入点工具已获取，客户端准备就绪" + ) + return + + # Handle method calls (requests from the endpoint) + elif "method" in payload: + method = payload["method"] + logger.bind(tag=TAG).info(f"收到MCP接入点请求: {method}") + + elif "error" in payload: + error_data = payload["error"] + error_msg = error_data.get("message", "未知错误") + logger.bind(tag=TAG).error(f"收到MCP接入点错误响应: {error_msg}") + + # 安全地获取消息ID，如果为None则使用0 + msg_id_raw = payload.get("id") + msg_id = int(msg_id_raw) if msg_id_raw is not None else 0 + + if msg_id in mcp_client.call_results: + await mcp_client.reject_call_result( + msg_id, Exception(f"MCP接入点错误: {error_msg}") + ) + + except json.JSONDecodeError as e: + logger.bind(tag=TAG).error(f"MCP接入点消息JSON解析失败: {e}") + except Exception as e: + logger.bind(tag=TAG).error(f"处理MCP接入点消息时出错: {e}") + import traceback + + logger.bind(tag=TAG).error(f"错误详情: {traceback.format_exc()}") + + +async def send_mcp_endpoint_initialize(mcp_client: MCPEndpointClient): + """发送MCP接入点初始化消息""" + payload = { + "jsonrpc": "2.0", + "id": 1, # mcpInitializeID + "method": "initialize", + "params": { + "protocolVersion": "2024-11-05", + "capabilities": { + "roots": {"listChanged": True}, + "sampling": {}, + }, + "clientInfo": { + "name": "XiaozhiMCPEndpointClient", + "version": "1.0.0", + }, + }, + } + message = json.dumps(payload) + logger.bind(tag=TAG).info("发送MCP接入点初始化消息") + await mcp_client.send_message(message) + + +async def send_mcp_endpoint_notification(mcp_client: MCPEndpointClient, method: str): + """发送MCP接入点通知消息""" + payload = { + "jsonrpc": "2.0", + "method": method, + "params": {}, + } + message = json.dumps(payload) + logger.bind(tag=TAG).debug(f"发送MCP接入点通知: {method}") + await mcp_client.send_message(message) + + +async def send_mcp_endpoint_tools_list(mcp_client: MCPEndpointClient): + """发送MCP接入点工具列表请求""" + payload = { + "jsonrpc": "2.0", + "id": 2, # mcpToolsListID + "method": "tools/list", + } + message = json.dumps(payload) + logger.bind(tag=TAG).debug("发送MCP接入点工具列表请求") + await mcp_client.send_message(message) + + +async def send_mcp_endpoint_tools_list_continue( + mcp_client: MCPEndpointClient, cursor: str +): + """发送带有cursor的MCP接入点工具列表请求""" + payload = { + "jsonrpc": "2.0", + "id": 2, # mcpToolsListID (same ID for continuation) + "method": "tools/list", + "params": {"cursor": cursor}, + } + message = json.dumps(payload) + logger.bind(tag=TAG).info(f"发送带cursor的MCP接入点工具列表请求: {cursor}") + await mcp_client.send_message(message) + + +async def call_mcp_endpoint_tool( + mcp_client: MCPEndpointClient, tool_name: str, args: str = "{}", timeout: int = 30 +): + """ + 调用指定的MCP接入点工具，并等待响应 + """ + if not await mcp_client.is_ready(): + raise RuntimeError("MCP接入点客户端尚未准备就绪") + + if not 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) + + # 处理参数 + try: + if isinstance(args, str): + # 确保字符串是有效的JSON + if not args.strip(): + arguments = {} + else: + try: + # 尝试直接解析 + arguments = json.loads(args) + except json.JSONDecodeError: + # 如果解析失败，尝试合并多个JSON对象 + try: + # 使用正则表达式匹配所有JSON对象 + json_objects = re.findall(r"\{[^{}]*\}", args) + if len(json_objects) > 1: + # 合并所有JSON对象 + merged_dict = {} + for json_str in json_objects: + try: + obj = json.loads(json_str) + if isinstance(obj, dict): + merged_dict.update(obj) + except json.JSONDecodeError: + continue + if merged_dict: + arguments = merged_dict + else: + raise ValueError(f"无法解析任何有效的JSON对象: {args}") + else: + raise ValueError(f"参数JSON解析失败: {args}") + except Exception as e: + logger.bind(tag=TAG).error( + f"参数JSON解析失败: {str(e)}, 原始参数: {args}" + ) + raise ValueError(f"参数JSON解析失败: {str(e)}") + elif isinstance(args, dict): + arguments = args + else: + raise ValueError(f"参数类型错误，期望字符串或字典，实际类型: {type(args)}") + + # 确保参数是字典类型 + if not isinstance(arguments, dict): + raise ValueError(f"参数必须是字典类型，实际类型: {type(arguments)}") + + except Exception as e: + if not isinstance(e, ValueError): + raise ValueError(f"参数处理失败: {str(e)}") + raise e + + actual_name = mcp_client.name_mapping.get(tool_name, tool_name) + payload = { + "jsonrpc": "2.0", + "id": tool_call_id, + "method": "tools/call", + "params": {"name": actual_name, "arguments": arguments}, + } + + message = json.dumps(payload) + logger.bind(tag=TAG).info(f"发送MCP接入点工具调用请求: {actual_name}，参数: {args}") + await mcp_client.send_message(message) + + try: + # Wait for response or timeout + raw_result = await asyncio.wait_for(result_future, timeout=timeout) + logger.bind(tag=TAG).info( + f"MCP接入点工具调用 {actual_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]: + # 直接返回文本内容，不进行JSON解析 + return content[0]["text"] + # 如果结果不是预期的格式，将其转换为字符串 + return str(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/providers/tools/unified_tool_handler.py b/main/xiaozhi-server/core/providers/tools/unified_tool_handler.py index 8e3b20b5..e65aa791 100644 --- a/main/xiaozhi-server/core/providers/tools/unified_tool_handler.py +++ b/main/xiaozhi-server/core/providers/tools/unified_tool_handler.py @@ -1,4 +1,4 @@ -"""统一工具处理器，替换原有的FunctionHandler""" +"""统一工具处理器""" import json from typing import Dict, List, Any, Optional @@ -12,6 +12,7 @@ from .server_plugins import ServerPluginExecutor from .server_mcp import ServerMCPExecutor from .device_iot import DeviceIoTExecutor from .device_mcp import DeviceMCPExecutor +from .mcp_endpoint import MCPEndpointExecutor class UnifiedToolHandler: @@ -30,6 +31,7 @@ class UnifiedToolHandler: self.server_mcp_executor = ServerMCPExecutor(conn) self.device_iot_executor = DeviceIoTExecutor(conn) self.device_mcp_executor = DeviceMCPExecutor(conn) + self.mcp_endpoint_executor = MCPEndpointExecutor(conn) # 注册执行器 self.tool_manager.register_executor( @@ -44,6 +46,9 @@ class UnifiedToolHandler: self.tool_manager.register_executor( ToolType.DEVICE_MCP, self.device_mcp_executor ) + self.tool_manager.register_executor( + ToolType.MCP_ENDPOINT, self.mcp_endpoint_executor + ) # 初始化标志 self.finish_init = False @@ -57,6 +62,9 @@ class UnifiedToolHandler: # 初始化服务端MCP await self.server_mcp_executor.initialize() + # 初始化MCP接入点 + await self._initialize_mcp_endpoint() + # 初始化Home Assistant（如果需要） self._initialize_home_assistant() @@ -66,6 +74,28 @@ class UnifiedToolHandler: except Exception as e: self.logger.error(f"统一工具处理器初始化失败: {e}") + async def _initialize_mcp_endpoint(self): + """初始化MCP接入点""" + try: + from .mcp_endpoint import connect_mcp_endpoint + + # 从配置中获取MCP接入点URL + mcp_endpoint_url = self.config.get("mcp_endpoint", "") + + if mcp_endpoint_url and "你的" not in mcp_endpoint_url: + self.logger.info(f"正在初始化MCP接入点: {mcp_endpoint_url}") + mcp_endpoint_client = await connect_mcp_endpoint(mcp_endpoint_url) + + if mcp_endpoint_client: + # 将MCP接入点客户端保存到连接对象中 + self.conn.mcp_endpoint_client = mcp_endpoint_client + self.logger.info("MCP接入点初始化成功") + else: + self.logger.warning("MCP接入点初始化失败") + + except Exception as e: + self.logger.error(f"初始化MCP接入点失败: {e}") + def _initialize_home_assistant(self): """初始化Home Assistant提示词""" try: @@ -183,6 +213,14 @@ class UnifiedToolHandler: """清理资源""" try: await self.server_mcp_executor.cleanup() + + # 清理MCP接入点连接 + if ( + hasattr(self.conn, "mcp_endpoint_client") + and self.conn.mcp_endpoint_client + ): + await self.conn.mcp_endpoint_client.close() + self.logger.info("工具处理器清理完成") except Exception as e: self.logger.error(f"工具处理器清理失败: {e}")