fix device MCP nav alert TTS handler

main/xiaozhi-server/core/providers/tools/device_mcp/mcp_handler.py

@@ -1,13 +1,17 @@
-"""设备端MCP客户端支持模块"""
+"""Device-side MCP client support."""
 
-import json
 import asyncio
+import json
 import re
+import uuid
 from concurrent.futures import Future
-from core.utils.util import get_vision_url, sanitize_tool_name
-from core.utils.auth import AuthToken
+from typing import TYPE_CHECKING, Any
+
 from config.logger import setup_logging
-from typing import TYPE_CHECKING
+from core.providers.tts.dto.dto import ContentType, SentenceType, TTSMessageDTO
+from core.utils.auth import AuthToken
+from core.utils.dialogue import Message
+from core.utils.util import get_vision_url, sanitize_tool_name
 
 if TYPE_CHECKING:
     from core.connection import ConnectionHandler
@@ -17,26 +21,24 @@ logger = setup_logging()
 
 
 class MCPClient:
-    """设备端MCP客户端，用于管理MCP状态和工具"""
+    """Tracks device MCP tools and pending call results."""
 
     def __init__(self):
-        self.tools = {}  # sanitized_name -> tool_data
-        self.name_mapping = {}
+        self.tools: dict[str, dict[str, Any]] = {}
+        self.name_mapping: dict[str, str] = {}
         self.ready = False
-        self.call_results = {}  # To store Futures for tool call responses
+        self.call_results: dict[int, Future] = {}
         self.next_id = 1
         self.lock = asyncio.Lock()
-        self._cached_available_tools = None  # Cache for get_available_tools
+        self._cached_available_tools = None
 
     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 = {
@@ -50,7 +52,7 @@ class MCPClient:
             }
             result.append({"type": "function", "function": function_def})
 
-        self._cached_available_tools = result  # Store the generated list in cache
+        self._cached_available_tools = result
         return result
 
     async def is_ready(self) -> bool:
@@ -66,9 +68,7 @@ class MCPClient:
             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
-            )
+            self._cached_available_tools = None
 
     async def get_next_id(self) -> int:
         async with self.lock:
@@ -76,189 +76,211 @@ class MCPClient:
             self.next_id += 1
             return current_id
 
-    async def register_call_result_future(self, id: int, future: Future):
+    async def register_call_result_future(self, call_id: int, future: Future):
         async with self.lock:
-            self.call_results[id] = future
+            self.call_results[call_id] = future
 
-    async def resolve_call_result(self, id: int, result: any):
+    async def resolve_call_result(self, call_id: int, result: Any):
         async with self.lock:
-            if id in self.call_results:
-                future = self.call_results.pop(id)
+            if call_id in self.call_results:
+                future = self.call_results.pop(call_id)
                 if not future.done():
                     future.set_result(result)
 
-    async def reject_call_result(self, id: int, exception: Exception):
+    async def reject_call_result(self, call_id: int, exception: Exception):
         async with self.lock:
-            if id in self.call_results:
-                future = self.call_results.pop(id)
+            if call_id in self.call_results:
+                future = self.call_results.pop(call_id)
                 if not future.done():
                     future.set_exception(exception)
 
-    async def cleanup_call_result(self, id: int):
+    async def cleanup_call_result(self, call_id: int):
         async with self.lock:
-            if id in self.call_results:
-                self.call_results.pop(id)
+            if call_id in self.call_results:
+                self.call_results.pop(call_id)
 
 
 async def send_mcp_message(conn: "ConnectionHandler", payload: dict):
-    """Helper to send MCP messages, encapsulating common logic."""
-    if not conn.features.get("mcp"):
-        logger.bind(tag=TAG).warning("客户端不支持MCP，无法发送MCP消息")
+    """Send an MCP message to the device if the connection supports MCP."""
+    if not conn.features or not conn.features.get("mcp"):
+        logger.bind(tag=TAG).warning("Client does not support MCP, skipping message")
         return
 
     message = json.dumps({"type": "mcp", "payload": payload})
-
     try:
         await conn.websocket.send(message)
-        logger.bind(tag=TAG).debug(f"成功发送MCP消息: {message}")
+        logger.bind(tag=TAG).debug(f"Sent MCP message: {message}")
     except Exception as e:
-        logger.bind(tag=TAG).error(f"发送MCP消息失败: {e}")
+        logger.bind(tag=TAG).error(f"Failed to send MCP message: {e}")
+
+
+async def _enqueue_device_tts(
+    conn: "ConnectionHandler", text: str, interrupt: bool = True
+):
+    if interrupt:
+        conn.client_abort = True
+        conn.clear_queues()
+        conn.clearSpeakStatus()
+
+    conn.client_abort = False
+    conn.sentence_id = uuid.uuid4().hex
+    conn.tts.store_tts_text(conn.sentence_id, text)
+    conn.tts.tts_text_queue.put(
+        TTSMessageDTO(
+            sentence_id=conn.sentence_id,
+            sentence_type=SentenceType.FIRST,
+            content_type=ContentType.ACTION,
+        )
+    )
+    conn.tts.tts_one_sentence(conn, ContentType.TEXT, content_detail=text)
+    conn.tts.tts_text_queue.put(
+        TTSMessageDTO(
+            sentence_id=conn.sentence_id,
+            sentence_type=SentenceType.LAST,
+            content_type=ContentType.ACTION,
+        )
+    )
+    conn.dialogue.put(Message(role="assistant", content=text, is_temporary=True))
 
 
 async def handle_mcp_message(
     conn: "ConnectionHandler", mcp_client: MCPClient, payload: dict
 ):
-    """处理MCP消息,包括初始化、工具列表和工具调用响应等"""
-    logger.bind(tag=TAG).debug(f"处理MCP消息: {str(payload)[:100]}")
+    logger.bind(tag=TAG).debug(f"Handling MCP payload: {str(payload)[:200]}")
 
     if not isinstance(payload, dict):
-        logger.bind(tag=TAG).error("MCP消息缺少payload字段或格式错误")
+        logger.bind(tag=TAG).error("Invalid MCP payload format")
         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:
             logger.bind(tag=TAG).debug(
-                f"收到工具调用响应，ID: {msg_id}, 结果: {result}"
+                f"Received MCP tool call result 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初始化响应")
+        if msg_id == 1:
+            logger.bind(tag=TAG).debug("Received MCP initialize response")
             server_info = result.get("serverInfo")
             if isinstance(server_info, dict):
-                name = server_info.get("name")
-                version = server_info.get("version")
                 logger.bind(tag=TAG).debug(
-                    f"客户端MCP服务器信息: name={name}, version={version}"
+                    "Device MCP server info: name=%s version=%s",
+                    server_info.get("name"),
+                    server_info.get("version"),
                 )
 
             await asyncio.sleep(1)
-            logger.bind(tag=TAG).debug("初始化完成，开始请求MCP工具列表")
             await send_mcp_tools_list_request(conn)
-
             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
+        if msg_id == 2:
+            logger.bind(tag=TAG).debug("Received MCP tools list response")
+            if not isinstance(result, dict) or "tools" not in result:
+                return
 
-                logger.bind(tag=TAG).info(
-                    f"客户端设备支持的工具数量: {len(tools_data)}"
-                )
+            tools_data = result["tools"]
+            if not isinstance(tools_data, list):
+                logger.bind(tag=TAG).error("Invalid MCP tools list format")
+                return
 
-                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"客户端工具 #{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).debug(f"有更多工具，nextCursor: {next_cursor}")
-                    await send_mcp_tools_list_continue_request(conn, next_cursor)
-                else:
-                    await mcp_client.set_ready(True)
-                    logger.bind(tag=TAG).debug("所有工具已获取，MCP客户端准备就绪")
-
-                    # 刷新工具缓存，确保MCP工具被包含在函数列表中
-                    if hasattr(conn, "func_handler") and conn.func_handler:
-                        conn.func_handler.tool_manager.refresh_tools()
-                        conn.func_handler.current_support_functions()
+            logger.bind(tag=TAG).info(
+                f"Device reported {len(tools_data)} MCP tools"
+            )
+
+            for tool in tools_data:
+                if not isinstance(tool, dict):
+                    continue
+
+                schema = tool.get("inputSchema", {})
+                if not isinstance(schema, dict):
+                    schema = {}
+
+                new_tool = {
+                    "name": tool.get("name", ""),
+                    "description": tool.get("description", ""),
+                    "inputSchema": {
+                        "type": schema.get("type", "object"),
+                        "properties": schema.get("properties", {}),
+                        "required": [
+                            item
+                            for item in schema.get("required", [])
+                            if isinstance(item, str)
+                        ],
+                    },
+                }
+                await mcp_client.add_tool(new_tool)
+
+            for tool_data in mcp_client.tools.values():
+                description = tool_data.get("description")
+                if not isinstance(description, str):
+                    continue
+                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:
+                await send_mcp_tools_list_continue_request(conn, next_cursor)
+            else:
+                await mcp_client.set_ready(True)
+                if hasattr(conn, "func_handler") and conn.func_handler:
+                    conn.func_handler.tool_manager.refresh_tools()
+                    conn.func_handler.current_support_functions()
             return
 
-    # Handle method calls (requests from the client)
     elif "method" in payload:
         method = payload["method"]
-        logger.bind(tag=TAG).info(f"收到MCP客户端请求: {method}")
+        logger.bind(tag=TAG).info(f"Received MCP request from device: {method}")
+
+        if method == "nav.alert.speak":
+            params = payload.get("params", {})
+            if not isinstance(params, dict):
+                logger.bind(tag=TAG).warning("nav.alert.speak params are invalid")
+                return
+
+            text = params.get("text", "")
+            interrupt = bool(params.get("interrupt", True))
+            if not isinstance(text, str) or not text.strip():
+                logger.bind(tag=TAG).warning("nav.alert.speak missing text")
+                return
+
+            await _enqueue_device_tts(conn, text.strip(), interrupt)
+            return
 
     elif "error" in payload:
         error_data = payload["error"]
-        error_msg = error_data.get("message", "未知错误")
-        logger.bind(tag=TAG).error(f"收到MCP错误响应: {error_msg}")
+        error_msg = error_data.get("message", "Unknown MCP error")
+        logger.bind(tag=TAG).error(f"Received MCP error response: {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}")
+                msg_id, Exception(f"MCP error: {error_msg}")
             )
 
 
 async def send_mcp_initialize_message(conn: "ConnectionHandler"):
-    """发送MCP初始化消息"""
-
     vision_url = get_vision_url(conn.config)
-
-    # 密钥生成token
     auth = AuthToken(conn.config["server"]["auth_key"])
     token = auth.generate_token(conn.headers.get("device-id"))
 
-    vision = {
-        "url": vision_url,
-        "token": token,
-    }
-
     payload = {
         "jsonrpc": "2.0",
-        "id": 1,  # mcpInitializeID
+        "id": 1,
         "method": "initialize",
         "params": {
             "protocolVersion": "2024-11-05",
             "capabilities": {
                 "roots": {"listChanged": True},
                 "sampling": {},
-                "vision": vision,
+                "vision": {
+                    "url": vision_url,
+                    "token": token,
+                },
             },
             "clientInfo": {
                 "name": "XiaozhiClient",
@@ -266,30 +288,30 @@ async def send_mcp_initialize_message(conn: "ConnectionHandler"):
             },
         },
     }
-    logger.bind(tag=TAG).debug("发送MCP初始化消息")
+    logger.bind(tag=TAG).debug("Sending MCP initialize message")
     await send_mcp_message(conn, payload)
 
 
 async def send_mcp_tools_list_request(conn: "ConnectionHandler"):
-    """发送MCP工具列表请求"""
     payload = {
         "jsonrpc": "2.0",
-        "id": 2,  # mcpToolsListID
+        "id": 2,
         "method": "tools/list",
     }
-    logger.bind(tag=TAG).debug("发送MCP工具列表请求")
+    logger.bind(tag=TAG).debug("Sending MCP tools list request")
     await send_mcp_message(conn, payload)
 
 
-async def send_mcp_tools_list_continue_request(conn: "ConnectionHandler", cursor: str):
-    """发送带有cursor的MCP工具列表请求"""
+async def send_mcp_tools_list_continue_request(
+    conn: "ConnectionHandler", cursor: str
+):
     payload = {
         "jsonrpc": "2.0",
-        "id": 2,  # mcpToolsListID (same ID for continuation)
+        "id": 2,
         "method": "tools/list",
         "params": {"cursor": cursor},
     }
-    logger.bind(tag=TAG).info(f"发送带cursor的MCP工具列表请求: {cursor}")
+    logger.bind(tag=TAG).info(f"Sending MCP tools list continuation: {cursor}")
     await send_mcp_message(conn, payload)
 
 
@@ -300,36 +322,27 @@ async def call_mcp_tool(
     args: str = "{}",
     timeout: int = 30,
 ):
-    """
-    调用指定的工具，并等待响应
-    """
     if not await mcp_client.is_ready():
-        raise RuntimeError("MCP客户端尚未准备就绪")
+        raise RuntimeError("MCP client is not ready")
 
     if not mcp_client.has_tool(tool_name):
-        raise ValueError(f"工具 {tool_name} 不存在")
+        raise ValueError(f"Tool {tool_name} does not exist")
 
     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:
@@ -341,27 +354,32 @@ async def call_mcp_tool(
                             if merged_dict:
                                 arguments = merged_dict
                             else:
-                                raise ValueError(f"无法解析任何有效的JSON对象: {args}")
+                                raise ValueError(
+                                    f"Unable to parse any valid JSON object: {args}"
+                                )
                         else:
-                            raise ValueError(f"参数JSON解析失败: {args}")
+                            raise ValueError(f"Failed to parse JSON arguments: {args}")
                     except Exception as e:
                         logger.bind(tag=TAG).error(
-                            f"参数JSON解析失败: {str(e)}, 原始参数: {args}"
+                            f"Failed to parse tool arguments: {e}, raw={args}"
                         )
-                        raise ValueError(f"参数JSON解析失败: {str(e)}")
+                        raise ValueError(f"Failed to parse tool arguments: {e}")
         elif isinstance(args, dict):
             arguments = args
         else:
-            raise ValueError(f"参数类型错误，期望字符串或字典，实际类型: {type(args)}")
+            raise ValueError(
+                f"Invalid arguments type: expected str or dict, got {type(args)}"
+            )
 
-        # 确保参数是字典类型
         if not isinstance(arguments, dict):
-            raise ValueError(f"参数必须是字典类型，实际类型: {type(arguments)}")
+            raise ValueError(
+                f"Arguments must be a dict, got {type(arguments)}"
+            )
 
     except Exception as e:
         if not isinstance(e, ValueError):
-            raise ValueError(f"参数处理失败: {str(e)}")
-        raise e
+            raise ValueError(f"Failed to process tool arguments: {e}")
+        raise
 
     actual_name = mcp_client.name_mapping.get(tool_name, tool_name)
     payload = {
@@ -371,33 +389,34 @@ async def call_mcp_tool(
         "params": {"name": actual_name, "arguments": arguments},
     }
 
-    logger.bind(tag=TAG).info(f"发送客户端mcp工具调用请求: {actual_name}，参数: {args}")
+    logger.bind(tag=TAG).info(
+        f"Sending device MCP tool call: {actual_name}, args={arguments}"
+    )
     await send_mcp_message(conn, payload)
 
     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}"
+            f"Device MCP tool call succeeded: {actual_name}, result={raw_result}"
         )
 
         if isinstance(raw_result, dict):
             if raw_result.get("isError") is True:
                 error_msg = raw_result.get(
-                    "error", "工具调用返回错误，但未提供具体错误信息"
+                    "error",
+                    "Tool call returned an error without details",
                 )
-                raise RuntimeError(f"工具调用错误: {error_msg}")
+                raise RuntimeError(f"Tool call error: {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:
+        raise TimeoutError("Tool call request timed out")
+    except Exception:
         await mcp_client.cleanup_call_result(tool_call_id)
-        raise e
+        raise