Merge branch 'newsnow' into add_local_tts_change_voice

main/xiaozhi-server/core/providers/asr/doubao_stream.py

@@ -93,9 +93,7 @@ class ASRProvider(ASRProviderBase):
 
                     # 检查初始化响应
                     if "code" in result and result["code"] != 1000:
-                        error_msg = f"ASR服务初始化失败: {result.get('payload_msg', {}).get('message', '未知错误')}"
-                        if "payload_msg" in result:
-                            error_msg += f"\n详细错误信息: {json.dumps(result['payload_msg'], ensure_ascii=False)}"
+                        error_msg = f"ASR服务初始化失败: {result.get('payload_msg', {}).get('error', '未知错误')}"
                         logger.bind(tag=TAG).error(error_msg)
                         raise Exception(error_msg)
 
@@ -256,7 +254,6 @@ class ASRProvider(ASRProviderBase):
             "X-Api-Access-Key": self.access_token,
             "X-Api-Resource-Id": "volc.bigasr.sauc.duration",
             "X-Api-Connect-Id": str(uuid.uuid4()),
-            "Host": "openspeech.bytedance.com",
         }
 
     def generate_header(
@@ -309,9 +306,14 @@ class ASRProvider(ASRProviderBase):
 
             # 如果是错误响应
             if message_type == 0x0F:  # SERVER_ERROR_RESPONSE
-                code = int.from_bytes(header[4:8], "big", signed=False)
-                error_msg = res[8:].decode("utf-8")
-                return {"code": code, "error": error_msg}
+                code = int.from_bytes(res[4:8], "big", signed=False)
+                msg_length = int.from_bytes(res[8:12], "big", signed=False)
+                error_msg = json.loads(res[12:].decode("utf-8"))
+                return {
+                    "code": code,
+                    "msg_length": msg_length,
+                    "payload_msg": error_msg,
+                }
 
             # 获取JSON数据（跳过12字节头部）
             try:

main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py

@@ -76,6 +76,14 @@ class IntentProvider(IntentProviderBase):
             '返回: {"function_call": {"name": "get_battery_level", "arguments": {"response_success": "当前电池电量为{value}%", "response_failure": "无法获取Battery的当前电量百分比"}}}\n'
             "```\n"
             "```\n"
+            "用户: 当前屏幕亮度是多少？\n"
+            '返回: {"function_call": {"name": "self_screen_get_brightness"}}\n'
+            "```\n"
+            "```\n"
+            "用户: 设置屏幕亮度为50%\n"
+            '返回: {"function_call": {"name": "self_screen_set_brightness", "arguments": {"brightness": 50}}}\n'
+            "```\n"
+            "```\n"
             "用户: 我想结束对话\n"
             '返回: {"function_call": {"name": "handle_exit_intent", "arguments": {"say_goodbye": "goodbye"}}}\n'
             "```\n"
@@ -87,6 +95,10 @@ class IntentProvider(IntentProviderBase):
             "1. 只返回JSON格式，不要包含任何其他文字\n"
             '2. 如果没有找到匹配的函数，返回{"function_call": {"name": "continue_chat"}}\n'
             "3. 确保返回的JSON格式正确，包含所有必要的字段\n"
+            "特殊说明：\n"
+            "- 当用户单次输入包含多个指令时（如'打开灯并且调高音量'）\n"
+            "- 请返回多个function_call组成的JSON数组\n"
+            "- 示例：{'function_calls': [{name:'light_on'}, {name:'volume_up'}]}"
         )
         return prompt
 
@@ -164,7 +176,11 @@ class IntentProvider(IntentProviderBase):
         music_file_names = music_config["music_file_names"]
         prompt_music = f"{self.promot}\n<musicNames>{music_file_names}\n</musicNames>"
 
-        devices = conn.config["plugins"]["home_assistant"].get("devices", [])
+        home_assistant_cfg = conn.config["plugins"].get("home_assistant")
+        if home_assistant_cfg:
+            devices = home_assistant_cfg.get("devices", [])
+        else:
+            devices = []
         if len(devices) > 0:
             hass_prompt = "\n下面是我家智能设备列表（位置，设备名，entity_id），可以通过homeassistant控制\n"
             for device in devices:

main/xiaozhi-server/core/providers/llm/coze/coze.py

@@ -23,7 +23,9 @@ class LLMProvider(LLMProviderBase):
         self.bot_id = str(config.get("bot_id"))
         self.user_id = str(config.get("user_id"))
         self.session_conversation_map = {}  # 存储session_id和conversation_id的映射
-        check_model_key("CozeLLM", self.personal_access_token)
+        model_key_msg = check_model_key("CozeLLM", self.personal_access_token)
+        if model_key_msg:
+            logger.bind(tag=TAG).error(model_key_msg)
 
     def response(self, session_id, dialogue, **kwargs):
         coze_api_token = self.personal_access_token

main/xiaozhi-server/core/providers/llm/dify/dify.py

@@ -15,7 +15,9 @@ class LLMProvider(LLMProviderBase):
         self.mode = config.get("mode", "chat-messages")
         self.base_url = config.get("base_url", "https://api.dify.ai/v1").rstrip("/")
         self.session_conversation_map = {}  # 存储session_id和conversation_id的映射
-        check_model_key("DifyLLM", self.api_key)
+        model_key_msg = check_model_key("DifyLLM", self.api_key)
+        if model_key_msg:
+            logger.bind(tag=TAG).error(model_key_msg)
 
     def response(self, session_id, dialogue, **kwargs):
         try:

main/xiaozhi-server/core/providers/llm/fastgpt/fastgpt.py

@@ -14,7 +14,9 @@ class LLMProvider(LLMProviderBase):
         self.base_url = config.get("base_url")
         self.detail = config.get("detail", False)
         self.variables = config.get("variables", {})
-        check_model_key("FastGPTLLM", self.api_key)
+        model_key_msg = check_model_key("FastGPTLLM", self.api_key)
+        if model_key_msg:
+            logger.bind(tag=TAG).error(model_key_msg)
 
     def response(self, session_id, dialogue, **kwargs):
         try:

main/xiaozhi-server/core/providers/llm/gemini/gemini.py

@@ -73,8 +73,9 @@ class LLMProvider(LLMProviderBase):
         http_proxy = cfg.get("http_proxy")
         https_proxy = cfg.get("https_proxy")
 
-        if not check_model_key("LLM", self.api_key):
-            raise ValueError("无效的Gemini API Key，请检查是否配置正确")
+        model_key_msg = check_model_key("LLM", self.api_key)
+        if model_key_msg:
+            log.bind(tag=TAG).error(model_key_msg)
 
         if http_proxy or https_proxy:
             log.bind(tag=TAG).info(

main/xiaozhi-server/core/providers/llm/openai/openai.py

@@ -1,3 +1,4 @@
+import httpx
 import openai
 from openai.types import CompletionUsage
 from config.logger import setup_logging
@@ -16,26 +17,36 @@ class LLMProvider(LLMProviderBase):
             self.base_url = config.get("base_url")
         else:
             self.base_url = config.get("url")
+        # 增加timeout的配置项，单位为秒
+        timeout = config.get("timeout", 300)
+        self.timeout = int(timeout) if timeout else 300
 
         param_defaults = {
             "max_tokens": (500, int),
             "temperature": (0.7, lambda x: round(float(x), 1)),
             "top_p": (1.0, lambda x: round(float(x), 1)),
-            "frequency_penalty": (0, lambda x: round(float(x), 1))
+            "frequency_penalty": (0, lambda x: round(float(x), 1)),
         }
 
         for param, (default, converter) in param_defaults.items():
             value = config.get(param)
             try:
-                setattr(self, param, converter(value) if value not in (None, "") else default)
+                setattr(
+                    self,
+                    param,
+                    converter(value) if value not in (None, "") else default,
+                )
             except (ValueError, TypeError):
                 setattr(self, param, default)
 
         logger.debug(
-            f"意图识别参数初始化: {self.temperature}, {self.max_tokens}, {self.top_p}, {self.frequency_penalty}")
+            f"意图识别参数初始化: {self.temperature}, {self.max_tokens}, {self.top_p}, {self.frequency_penalty}"
+        )
 
-        check_model_key("LLM", self.api_key)
-        self.client = openai.OpenAI(api_key=self.api_key, base_url=self.base_url)
+        model_key_msg = check_model_key("LLM", self.api_key)
+        if model_key_msg:
+            logger.bind(tag=TAG).error(model_key_msg)
+        self.client = openai.OpenAI(api_key=self.api_key, base_url=self.base_url, timeout=httpx.Timeout(self.timeout))
 
     def response(self, session_id, dialogue, **kwargs):
         try:
@@ -46,7 +57,9 @@ class LLMProvider(LLMProviderBase):
                 max_tokens=kwargs.get("max_tokens", self.max_tokens),
                 temperature=kwargs.get("temperature", self.temperature),
                 top_p=kwargs.get("top_p", self.top_p),
-                frequency_penalty=kwargs.get("frequency_penalty", self.frequency_penalty),
+                frequency_penalty=kwargs.get(
+                    "frequency_penalty", self.frequency_penalty
+                ),
             )
 
             is_active = True
@@ -84,12 +97,14 @@ class LLMProvider(LLMProviderBase):
             for chunk in stream:
                 # 检查是否存在有效的choice且content不为空
                 if getattr(chunk, "choices", None):
-                    yield chunk.choices[0].delta.content, chunk.choices[0].delta.tool_calls
+                    yield chunk.choices[0].delta.content, chunk.choices[
+                        0
+                    ].delta.tool_calls
                 # 存在 CompletionUsage 消息时，生成 Token 消耗 log
-                elif isinstance(getattr(chunk, 'usage', None), CompletionUsage):
-                    usage_info = getattr(chunk, 'usage', None)
+                elif isinstance(getattr(chunk, "usage", None), CompletionUsage):
+                    usage_info = getattr(chunk, "usage", None)
                     logger.bind(tag=TAG).info(
-                        f"Token 消耗：输入 {getattr(usage_info, 'prompt_tokens', '未知')}，" 
+                        f"Token 消耗：输入 {getattr(usage_info, 'prompt_tokens', '未知')}，"
                         f"输出 {getattr(usage_info, 'completion_tokens', '未知')}，"
                         f"共计 {getattr(usage_info, 'total_tokens', '未知')}"
                     )

main/xiaozhi-server/core/providers/memory/base.py

@@ -12,10 +12,6 @@ class MemoryProviderBase(ABC):
 
     def set_llm(self, llm):
         self.llm = llm
-        # 获取模型名称和类型信息
-        model_name = getattr(llm, "model_name", str(llm.__class__.__name__))
-        # 记录更详细的日志
-        logger.bind(tag=TAG).info(f"记忆总结设置LLM: {model_name}")
 
     @abstractmethod
     async def save_memory(self, msgs):

main/xiaozhi-server/core/providers/memory/mem0ai/mem0ai.py

@@ -12,12 +12,14 @@ class MemoryProvider(MemoryProviderBase):
         super().__init__(config)
         self.api_key = config.get("api_key", "")
         self.api_version = config.get("api_version", "v1.1")
-        have_key = check_model_key("Mem0ai", self.api_key)
-        if not have_key:
+        model_key_msg = check_model_key("Mem0ai", self.api_key)
+        if model_key_msg:
+            logger.bind(tag=TAG).error(model_key_msg)
             self.use_mem0 = False
             return
         else:
             self.use_mem0 = True
+
         try:
             self.client = MemoryClient(api_key=self.api_key)
             logger.bind(tag=TAG).info("成功连接到 Mem0ai 服务")

main/xiaozhi-server/core/providers/memory/mem_local_short/mem_local_short.py

@@ -5,6 +5,7 @@ import os
 import yaml
 from config.config_loader import get_project_dir
 from config.manage_api_client import save_mem_local_short
+from core.utils.util import check_model_key
 
 
 short_term_memory_prompt = """
@@ -145,6 +146,10 @@ class MemoryProvider(MemoryProviderBase):
         # 打印使用的模型信息
         model_info = getattr(self.llm, "model_name", str(self.llm.__class__.__name__))
         logger.bind(tag=TAG).debug(f"使用记忆保存模型: {model_info}")
+        api_key = getattr(self.llm, "api_key", None)
+        memory_key_msg = check_model_key("记忆总结专用LLM", api_key)
+        if memory_key_msg:
+            logger.bind(tag=TAG).error(memory_key_msg)
         if self.llm is None:
             logger.bind(tag=TAG).error("LLM is not set for memory provider")
             return None

main/xiaozhi-server/core/providers/tools/__init__.py

@@ -0,0 +1 @@
+ 

main/xiaozhi-server/core/providers/tools/base/__init__.py

@@ -0,0 +1,6 @@
+"""基础工具定义模块"""
+
+from .tool_types import ToolType, ToolDefinition
+from .tool_executor import ToolExecutor
+
+__all__ = ["ToolType", "ToolDefinition", "ToolExecutor"]

main/xiaozhi-server/core/providers/tools/base/tool_executor.py

@@ -0,0 +1,27 @@
+"""工具执行器基类定义"""
+
+from abc import ABC, abstractmethod
+from typing import Dict, Any
+from .tool_types import ToolDefinition
+from plugins_func.register import ActionResponse
+
+
+class ToolExecutor(ABC):
+    """工具执行器抽象基类"""
+
+    @abstractmethod
+    async def execute(
+        self, conn, tool_name: str, arguments: Dict[str, Any]
+    ) -> ActionResponse:
+        """执行工具调用"""
+        pass
+
+    @abstractmethod
+    def get_tools(self) -> Dict[str, ToolDefinition]:
+        """获取该执行器管理的所有工具"""
+        pass
+
+    @abstractmethod
+    def has_tool(self, tool_name: str) -> bool:
+        """检查是否有指定工具"""
+        pass

main/xiaozhi-server/core/providers/tools/base/tool_types.py

@@ -0,0 +1,27 @@
+"""工具系统的类型定义"""
+
+from enum import Enum
+
+from dataclasses import dataclass
+from typing import Any, Dict, Optional
+from plugins_func.register import Action
+
+
+class ToolType(Enum):
+    """工具类型枚举"""
+
+    SERVER_PLUGIN = "server_plugin"  # 服务端插件
+    SERVER_MCP = "server_mcp"  # 服务端MCP
+    DEVICE_IOT = "device_iot"  # 设备端IoT
+    DEVICE_MCP = "device_mcp"  # 设备端MCP
+    MCP_ENDPOINT = "mcp_endpoint"  # MCP接入点
+
+
+@dataclass
+class ToolDefinition:
+    """工具定义"""
+
+    name: str  # 工具名称
+    description: Dict[str, Any]  # 工具描述（OpenAI函数调用格式）
+    tool_type: ToolType  # 工具类型
+    parameters: Optional[Dict[str, Any]] = None  # 额外参数

main/xiaozhi-server/core/providers/tools/device_iot/__init__.py

@@ -0,0 +1,12 @@
+"""设备端IoT工具模块"""
+
+from .iot_descriptor import IotDescriptor
+from .iot_handler import handleIotDescriptors, handleIotStatus
+from .iot_executor import DeviceIoTExecutor
+
+__all__ = [
+    "IotDescriptor",
+    "handleIotDescriptors",
+    "handleIotStatus",
+    "DeviceIoTExecutor",
+]

main/xiaozhi-server/core/providers/tools/device_iot/iot_descriptor.py

@@ -0,0 +1,46 @@
+"""IoT设备描述符定义"""
+
+from config.logger import setup_logging
+
+TAG = __name__
+logger = setup_logging()
+
+
+class IotDescriptor:
+    """IoT设备描述符"""
+
+    def __init__(self, name, description, properties, methods):
+        self.name = name
+        self.description = description
+        self.properties = []
+        self.methods = []
+
+        # 根据描述创建属性
+        if properties is not None:
+            for key, value in properties.items():
+                property_item = {}
+                property_item["name"] = key
+                property_item["description"] = value["description"]
+                if value["type"] == "number":
+                    property_item["value"] = 0
+                elif value["type"] == "boolean":
+                    property_item["value"] = False
+                else:
+                    property_item["value"] = ""
+                self.properties.append(property_item)
+
+        # 根据描述创建方法
+        if methods is not None:
+            for key, value in methods.items():
+                method = {}
+                method["description"] = value["description"]
+                method["name"] = key
+                # 检查方法是否有参数
+                if "parameters" in value:
+                    method["parameters"] = {}
+                    for k, v in value["parameters"].items():
+                        method["parameters"][k] = {
+                            "description": v["description"],
+                            "type": v["type"],
+                        }
+                self.methods.append(method)

main/xiaozhi-server/core/providers/tools/device_iot/iot_executor.py

@@ -0,0 +1,238 @@
+"""设备端IoT工具执行器"""
+
+import json
+import asyncio
+from typing import Dict, Any
+from ..base import ToolType, ToolDefinition, ToolExecutor
+from plugins_func.register import Action, ActionResponse
+
+
+class DeviceIoTExecutor(ToolExecutor):
+    """设备端IoT工具执行器"""
+
+    def __init__(self, conn):
+        self.conn = conn
+        self.iot_tools: Dict[str, ToolDefinition] = {}
+
+    async def execute(
+        self, conn, tool_name: str, arguments: Dict[str, Any]
+    ) -> ActionResponse:
+        """执行设备端IoT工具"""
+        if not self.has_tool(tool_name):
+            return ActionResponse(
+                action=Action.NOTFOUND, response=f"IoT工具 {tool_name} 不存在"
+            )
+
+        try:
+            # 解析工具名称，获取设备名和操作类型
+            if tool_name.startswith("get_"):
+                # 查询操作：get_devicename_property
+                parts = tool_name.split("_", 2)
+                if len(parts) >= 3:
+                    device_name = parts[1]
+                    property_name = parts[2]
+
+                    value = await self._get_iot_status(device_name, property_name)
+                    if value is not None:
+                        # 处理响应模板
+                        response_success = arguments.get(
+                            "response_success", "查询成功：{value}"
+                        )
+                        response = response_success.replace("{value}", str(value))
+
+                        return ActionResponse(
+                            action=Action.RESPONSE,
+                            response=response,
+                        )
+                    else:
+                        response_failure = arguments.get(
+                            "response_failure", f"无法获取{device_name}的状态"
+                        )
+                        return ActionResponse(
+                            action=Action.ERROR, response=response_failure
+                        )
+            else:
+                # 控制操作：devicename_method
+                parts = tool_name.split("_", 1)
+                if len(parts) >= 2:
+                    device_name = parts[0]
+                    method_name = parts[1]
+
+                    # 提取控制参数（排除响应参数）
+                    control_params = {
+                        k: v
+                        for k, v in arguments.items()
+                        if k not in ["response_success", "response_failure"]
+                    }
+
+                    # 发送IoT控制命令
+                    await self._send_iot_command(
+                        device_name, method_name, control_params
+                    )
+
+                    # 等待状态更新
+                    await asyncio.sleep(0.1)
+
+                    response_success = arguments.get("response_success", "操作成功")
+
+                    # 处理响应中的占位符
+                    for param_name, param_value in control_params.items():
+                        placeholder = "{" + param_name + "}"
+                        if placeholder in response_success:
+                            response_success = response_success.replace(
+                                placeholder, str(param_value)
+                            )
+                        if "{value}" in response_success:
+                            response_success = response_success.replace(
+                                "{value}", str(param_value)
+                            )
+                            break
+
+                    return ActionResponse(
+                        action=Action.REQLLM,
+                        result=response_success,
+                    )
+
+            return ActionResponse(action=Action.ERROR, response="无法解析IoT工具名称")
+
+        except Exception as e:
+            response_failure = arguments.get("response_failure", "操作失败")
+            return ActionResponse(action=Action.ERROR, response=response_failure)
+
+    async def _get_iot_status(self, device_name: str, property_name: str):
+        """获取IoT设备状态"""
+        for key, value in self.conn.iot_descriptors.items():
+            if key.lower() == device_name.lower():
+                for property_item in value.properties:
+                    if property_item["name"].lower() == property_name.lower():
+                        return property_item["value"]
+        return None
+
+    async def _send_iot_command(
+        self, device_name: str, method_name: str, parameters: Dict[str, Any]
+    ):
+        """发送IoT控制命令"""
+        for key, value in self.conn.iot_descriptors.items():
+            if key.lower() == device_name.lower():
+                for method in value.methods:
+                    if method["name"].lower() == method_name.lower():
+                        command = {
+                            "name": key,
+                            "method": method["name"],
+                        }
+
+                        if parameters:
+                            command["parameters"] = parameters
+
+                        send_message = json.dumps(
+                            {"type": "iot", "commands": [command]}
+                        )
+                        await self.conn.websocket.send(send_message)
+                        return
+
+        raise Exception(f"未找到设备{device_name}的方法{method_name}")
+
+    def register_iot_tools(self, descriptors: list):
+        """注册IoT工具"""
+        for descriptor in descriptors:
+            device_name = descriptor["name"]
+            device_desc = descriptor["description"]
+
+            # 注册查询工具
+            if "properties" in descriptor:
+                for prop_name, prop_info in descriptor["properties"].items():
+                    tool_name = f"get_{device_name.lower()}_{prop_name.lower()}"
+
+                    tool_desc = {
+                        "type": "function",
+                        "function": {
+                            "name": tool_name,
+                            "description": f"查询{device_desc}的{prop_info['description']}",
+                            "parameters": {
+                                "type": "object",
+                                "properties": {
+                                    "response_success": {
+                                        "type": "string",
+                                        "description": f"查询成功时的友好回复，必须使用{{value}}作为占位符表示查询到的值",
+                                    },
+                                    "response_failure": {
+                                        "type": "string",
+                                        "description": f"查询失败时的友好回复",
+                                    },
+                                },
+                                "required": ["response_success", "response_failure"],
+                            },
+                        },
+                    }
+
+                    self.iot_tools[tool_name] = ToolDefinition(
+                        name=tool_name,
+                        description=tool_desc,
+                        tool_type=ToolType.DEVICE_IOT,
+                    )
+
+            # 注册控制工具
+            if "methods" in descriptor:
+                for method_name, method_info in descriptor["methods"].items():
+                    tool_name = f"{device_name.lower()}_{method_name.lower()}"
+
+                    # 构建参数
+                    parameters = {}
+                    required_params = []
+
+                    # 添加方法的原始参数
+                    if "parameters" in method_info:
+                        parameters.update(
+                            {
+                                param_name: {
+                                    "type": param_info["type"],
+                                    "description": param_info["description"],
+                                }
+                                for param_name, param_info in method_info[
+                                    "parameters"
+                                ].items()
+                            }
+                        )
+                        required_params.extend(method_info["parameters"].keys())
+
+                    # 添加响应参数
+                    parameters.update(
+                        {
+                            "response_success": {
+                                "type": "string",
+                                "description": "操作成功时的友好回复",
+                            },
+                            "response_failure": {
+                                "type": "string",
+                                "description": "操作失败时的友好回复",
+                            },
+                        }
+                    )
+                    required_params.extend(["response_success", "response_failure"])
+
+                    tool_desc = {
+                        "type": "function",
+                        "function": {
+                            "name": tool_name,
+                            "description": f"{device_desc} - {method_info['description']}",
+                            "parameters": {
+                                "type": "object",
+                                "properties": parameters,
+                                "required": required_params,
+                            },
+                        },
+                    }
+
+                    self.iot_tools[tool_name] = ToolDefinition(
+                        name=tool_name,
+                        description=tool_desc,
+                        tool_type=ToolType.DEVICE_IOT,
+                    )
+
+    def get_tools(self) -> Dict[str, ToolDefinition]:
+        """获取所有设备端IoT工具"""
+        return self.iot_tools.copy()
+
+    def has_tool(self, tool_name: str) -> bool:
+        """检查是否有指定的设备端IoT工具"""
+        return tool_name in self.iot_tools

main/xiaozhi-server/core/providers/tools/device_iot/iot_handler.py

@@ -0,0 +1,83 @@
+"""IoT设备支持模块，提供IoT设备描述符和状态处理"""
+
+import asyncio
+from config.logger import setup_logging
+from .iot_descriptor import IotDescriptor
+
+TAG = __name__
+logger = setup_logging()
+
+
+async def handleIotDescriptors(conn, descriptors):
+    """处理物联网描述"""
+    wait_max_time = 5
+    while (
+        not hasattr(conn, "func_handler")
+        or conn.func_handler is None
+        or not conn.func_handler.finish_init
+    ):
+        await asyncio.sleep(1)
+        wait_max_time -= 1
+        if wait_max_time <= 0:
+            logger.bind(tag=TAG).debug("连接对象没有func_handler")
+            return
+
+    functions_changed = False
+
+    for descriptor in descriptors:
+        # 如果descriptor没有properties和methods，则直接跳过
+        if "properties" not in descriptor and "methods" not in descriptor:
+            continue
+
+        # 处理缺失properties的情况
+        if "properties" not in descriptor:
+            descriptor["properties"] = {}
+            # 从methods中提取所有参数作为properties
+            if "methods" in descriptor:
+                for method_name, method_info in descriptor["methods"].items():
+                    if "parameters" in method_info:
+                        for param_name, param_info in method_info["parameters"].items():
+                            # 将参数信息转换为属性信息
+                            descriptor["properties"][param_name] = {
+                                "description": param_info["description"],
+                                "type": param_info["type"],
+                            }
+
+        # 创建IOT设备描述符
+        iot_descriptor = IotDescriptor(
+            descriptor["name"],
+            descriptor["description"],
+            descriptor["properties"],
+            descriptor["methods"],
+        )
+        conn.iot_descriptors[descriptor["name"]] = iot_descriptor
+        functions_changed = True
+
+    # 如果注册了新函数，更新function描述列表
+    if functions_changed and hasattr(conn, "func_handler"):
+        # 注册IoT工具到统一工具处理器
+        await conn.func_handler.register_iot_tools(descriptors)
+
+        conn.func_handler.current_support_functions()
+
+
+async def handleIotStatus(conn, states):
+    """处理物联网状态"""
+    for state in states:
+        for key, value in conn.iot_descriptors.items():
+            if key == state["name"]:
+                for property_item in value.properties:
+                    for k, v in state["state"].items():
+                        if property_item["name"] == k:
+                            if type(v) != type(property_item["value"]):
+                                logger.bind(tag=TAG).error(
+                                    f"属性{property_item['name']}的值类型不匹配"
+                                )
+                                break
+                            else:
+                                property_item["value"] = v
+                                logger.bind(tag=TAG).info(
+                                    f"物联网状态更新: {key} , {property_item['name']} = {v}"
+                                )
+                            break
+                break

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

@@ -0,0 +1,21 @@
+"""设备端MCP工具模块"""
+
+from .mcp_client import MCPClient
+from .mcp_handler import (
+    send_mcp_message,
+    handle_mcp_message,
+    send_mcp_initialize_message,
+    send_mcp_tools_list_request,
+    call_mcp_tool,
+)
+from .mcp_executor import DeviceMCPExecutor
+
+__all__ = [
+    "MCPClient",
+    "send_mcp_message",
+    "handle_mcp_message",
+    "send_mcp_initialize_message",
+    "send_mcp_tools_list_request",
+    "call_mcp_tool",
+    "DeviceMCPExecutor",
+]

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

@@ -0,0 +1,93 @@
+"""设备端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 MCPClient:
+    """设备端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
+
+    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)

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

@@ -0,0 +1,89 @@
+"""设备端MCP工具执行器"""
+
+from typing import Dict, Any
+from ..base import ToolType, ToolDefinition, ToolExecutor
+from plugins_func.register import Action, ActionResponse
+from .mcp_handler import call_mcp_tool
+
+
+class DeviceMCPExecutor(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_client") or not conn.mcp_client:
+            return ActionResponse(
+                action=Action.ERROR,
+                response="设备端MCP客户端未初始化",
+            )
+
+        if not await conn.mcp_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_tool(conn, conn.mcp_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_client") or not self.conn.mcp_client:
+            return {}
+
+        tools = {}
+        mcp_tools = self.conn.mcp_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.DEVICE_MCP
+                )
+
+        return tools
+
+    def has_tool(self, tool_name: str) -> bool:
+        """检查是否有指定的设备端MCP工具"""
+        if not hasattr(self.conn, "mcp_client") or not self.conn.mcp_client:
+            return False
+
+        return self.conn.mcp_client.has_tool(tool_name)

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

@@ -0,0 +1,388 @@
+"""设备端MCP客户端支持模块"""
+
+import json
+import asyncio
+import re
+from concurrent.futures import Future
+from core.utils.util import get_vision_url, sanitize_tool_name
+from core.utils.auth import AuthToken
+from config.logger import setup_logging
+
+TAG = __name__
+logger = setup_logging()
+
+
+class MCPClient:
+    """设备端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
+
+    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)
+
+
+async def send_mcp_message(conn, payload: dict):
+    """Helper to send MCP messages, encapsulating common logic."""
+    if not conn.features.get("mcp"):
+        logger.bind(tag=TAG).warning("客户端不支持MCP，无法发送MCP消息")
+        return
+
+    message = json.dumps({"type": "mcp", "payload": payload})
+
+    try:
+        await conn.websocket.send(message)
+        logger.bind(tag=TAG).info(f"成功发送MCP消息: {message}")
+    except Exception as e:
+        logger.bind(tag=TAG).error(f"发送MCP消息失败: {e}")
+
+
+async def handle_mcp_message(conn, mcp_client: MCPClient, payload: dict):
+    """处理MCP消息,包括初始化、工具列表和工具调用响应等"""
+    logger.bind(tag=TAG).info(f"处理MCP消息: {str(payload)[:100]}")
+
+    if not isinstance(payload, dict):
+        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:
+            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"客户端设备支持的工具数量: {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"客户端工具 #{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_tools_list_continue_request(conn, next_cursor)
+                else:
+                    await mcp_client.set_ready(True)
+                    logger.bind(tag=TAG).info("所有工具已获取，MCP客户端准备就绪")
+
+                    # 刷新工具缓存，确保MCP工具被包含在函数列表中
+                    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}")
+
+    elif "error" in payload:
+        error_data = payload["error"]
+        error_msg = error_data.get("message", "未知错误")
+        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}")
+            )
+
+
+async def send_mcp_initialize_message(conn):
+    """发送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
+        "method": "initialize",
+        "params": {
+            "protocolVersion": "2024-11-05",
+            "capabilities": {
+                "roots": {"listChanged": True},
+                "sampling": {},
+                "vision": vision,
+            },
+            "clientInfo": {
+                "name": "XiaozhiClient",
+                "version": "1.0.0",
+            },
+        },
+    }
+    logger.bind(tag=TAG).info("发送MCP初始化消息")
+    await send_mcp_message(conn, payload)
+
+
+async def send_mcp_tools_list_request(conn):
+    """发送MCP工具列表请求"""
+    payload = {
+        "jsonrpc": "2.0",
+        "id": 2,  # mcpToolsListID
+        "method": "tools/list",
+    }
+    logger.bind(tag=TAG).debug("发送MCP工具列表请求")
+    await send_mcp_message(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},
+    }
+    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: str = "{}", timeout: int = 30
+):
+    """
+    调用指定的工具，并等待响应
+    """
+    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},
+    }
+
+    logger.bind(tag=TAG).info(f"发送客户端mcp工具调用请求: {actual_name}，参数: {args}")
+    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}"
+        )
+
+        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

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",
+]

main/xiaozhi-server/core/providers/tools/mcp_endpoint/mcp_endpoint_client.py

@@ -0,0 +1,112 @@
+"""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, conn=None):
+        self.conn = conn
+        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

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)

main/xiaozhi-server/core/providers/tools/mcp_endpoint/mcp_endpoint_handler.py

@@ -0,0 +1,390 @@
+"""MCP接入点处理器"""
+
+import json
+import asyncio
+import re
+import websockets
+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, conn=None) -> MCPEndpointClient:
+    """连接到MCP接入点"""
+    if not mcp_endpoint_url or "你的" in mcp_endpoint_url or mcp_endpoint_url == "null":
+        return None
+
+    try:
+        websocket = await websockets.connect(mcp_endpoint_url)
+
+        mcp_client = MCPEndpointClient(conn)
+        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接入点初始化响应")
+                if result is not None and isinstance(result, dict):
+                    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}"
+                        )
+                else:
+                    logger.bind(tag=TAG).warning(
+                        "MCP接入点初始化响应结果为空或格式错误"
+                    )
+                return
+
+            elif msg_id == 2:  # mcpToolsListID
+                logger.bind(tag=TAG).debug("收到MCP接入点工具列表响应")
+                if (
+                    result is not None
+                    and 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 result is not None else ""
+                    )
+                    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接入点工具已获取，客户端准备就绪"
+                        )
+
+                        # 刷新工具缓存，确保MCP接入点工具被包含在函数列表中
+                        if (
+                            hasattr(mcp_client, "conn")
+                            and mcp_client.conn
+                            and hasattr(mcp_client.conn, "func_handler")
+                            and mcp_client.conn.func_handler
+                        ):
+                            mcp_client.conn.func_handler.tool_manager.refresh_tools()
+                            mcp_client.conn.func_handler.current_support_functions()
+
+                        logger.bind(tag=TAG).info(
+                            f"MCP接入点工具获取完成，共 {len(mcp_client.tools)} 个工具"
+                        )
+                else:
+                    logger.bind(tag=TAG).warning(
+                        "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

main/xiaozhi-server/core/providers/tools/server_mcp/__init__.py

@@ -0,0 +1,7 @@
+"""服务端MCP工具模块"""
+
+from .mcp_manager import ServerMCPManager
+from .mcp_executor import ServerMCPExecutor
+from .mcp_client import ServerMCPClient
+
+__all__ = ["ServerMCPManager", "ServerMCPExecutor", "ServerMCPClient"]

main/xiaozhi-server/core/providers/tools/server_mcp/mcp_client.py

@@ -0,0 +1,208 @@
+"""服务端MCP客户端"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+import asyncio
+import os
+import shutil
+import concurrent.futures
+from contextlib import AsyncExitStack
+from typing import Optional, List, Dict, Any
+
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+from mcp.client.sse import sse_client
+from config.logger import setup_logging
+from core.utils.util import sanitize_tool_name
+
+TAG = __name__
+
+
+class ServerMCPClient:
+    """服务端MCP客户端，用于连接和管理MCP服务"""
+
+    def __init__(self, config: Dict[str, Any]):
+        """初始化服务端MCP客户端
+
+        Args:
+            config: MCP服务配置字典
+        """
+        self.logger = setup_logging()
+        self.config = config
+
+        self._worker_task: Optional[asyncio.Task] = None
+        self._ready_evt = asyncio.Event()
+        self._shutdown_evt = asyncio.Event()
+
+        self.session: Optional[ClientSession] = None
+        self.tools: List = []  # 原始工具对象
+        self.tools_dict: Dict[str, Any] = {}
+        self.name_mapping: Dict[str, str] = {}
+
+    async def initialize(self):
+        """初始化MCP客户端连接"""
+        if self._worker_task:
+            return
+
+        self._worker_task = asyncio.create_task(
+            self._worker(), name="ServerMCPClientWorker"
+        )
+        await self._ready_evt.wait()
+
+        self.logger.bind(tag=TAG).info(
+            f"服务端MCP客户端已连接，可用工具: {[name for name in self.name_mapping.values()]}"
+        )
+
+    async def cleanup(self):
+        """清理MCP客户端资源"""
+        if not self._worker_task:
+            return
+
+        self._shutdown_evt.set()
+        try:
+            await asyncio.wait_for(self._worker_task, timeout=20)
+        except (asyncio.TimeoutError, Exception) as e:
+            self.logger.bind(tag=TAG).error(f"服务端MCP客户端关闭错误: {e}")
+        finally:
+            self._worker_task = None
+
+    def has_tool(self, name: str) -> bool:
+        """检查是否包含指定工具
+
+        Args:
+            name: 工具名称
+
+        Returns:
+            bool: 是否包含该工具
+        """
+        return name in self.tools_dict
+
+    def get_available_tools(self) -> List[Dict[str, Any]]:
+        """获取所有可用工具的定义
+
+        Returns:
+            List[Dict[str, Any]]: 工具定义列表
+        """
+        return [
+            {
+                "type": "function",
+                "function": {
+                    "name": name,
+                    "description": tool.description,
+                    "parameters": tool.inputSchema,
+                },
+            }
+            for name, tool in self.tools_dict.items()
+        ]
+
+    async def call_tool(self, name: str, args: dict) -> Any:
+        """调用指定工具
+
+        Args:
+            name: 工具名称
+            args: 工具参数
+
+        Returns:
+            Any: 工具执行结果
+
+        Raises:
+            RuntimeError: 客户端未初始化时抛出
+        """
+        if not self.session:
+            raise RuntimeError("服务端MCP客户端未初始化")
+
+        real_name = self.name_mapping.get(name, name)
+        loop = self._worker_task.get_loop()
+        coro = self.session.call_tool(real_name, args)
+
+        if loop is asyncio.get_running_loop():
+            return await coro
+
+        fut: concurrent.futures.Future = asyncio.run_coroutine_threadsafe(coro, loop)
+        return await asyncio.wrap_future(fut)
+
+    def is_connected(self) -> bool:
+        """检查MCP客户端是否连接正常
+
+        Returns:
+            bool: 如果客户端已连接并正常工作，返回True，否则返回False
+        """
+        # 检查工作任务是否存在
+        if self._worker_task is None:
+            return False
+
+        # 检查工作任务是否已经完成或取消
+        if self._worker_task.done():
+            return False
+
+        # 检查会话是否存在
+        if self.session is None:
+            return False
+
+        # 所有检查都通过，连接正常
+        return True
+
+    async def _worker(self):
+        """MCP客户端工作协程"""
+        async with AsyncExitStack() as stack:
+            try:
+                # 建立 StdioClient
+                if "command" in self.config:
+                    cmd = (
+                        shutil.which("npx")
+                        if self.config["command"] == "npx"
+                        else self.config["command"]
+                    )
+                    env = {**os.environ, **self.config.get("env", {})}
+                    params = StdioServerParameters(
+                        command=cmd,
+                        args=self.config.get("args", []),
+                        env=env,
+                    )
+                    stdio_r, stdio_w = await stack.enter_async_context(
+                        stdio_client(params)
+                    )
+                    read_stream, write_stream = stdio_r, stdio_w
+
+                # 建立SSEClient
+                elif "url" in self.config:
+                    if "API_ACCESS_TOKEN" in self.config:
+                        headers = {
+                            "Authorization": f"Bearer {self.config['API_ACCESS_TOKEN']}"
+                        }
+                    else:
+                        headers = {}
+                    sse_r, sse_w = await stack.enter_async_context(
+                        sse_client(self.config["url"], headers=headers)
+                    )
+                    read_stream, write_stream = sse_r, sse_w
+
+                else:
+                    raise ValueError("MCP客户端配置必须包含'command'或'url'")
+
+                self.session = await stack.enter_async_context(
+                    ClientSession(
+                        read_stream=read_stream,
+                        write_stream=write_stream,
+                        read_timeout_seconds=timedelta(seconds=15),
+                    )
+                )
+                await self.session.initialize()
+
+                # 获取工具
+                self.tools = (await self.session.list_tools()).tools
+                for t in self.tools:
+                    sanitized = sanitize_tool_name(t.name)
+                    self.tools_dict[sanitized] = t
+                    self.name_mapping[sanitized] = t.name
+
+                self._ready_evt.set()
+
+                # 挂起等待关闭
+                await self._shutdown_evt.wait()
+
+            except Exception as e:
+                self.logger.bind(tag=TAG).error(f"服务端MCP客户端工作协程错误: {e}")
+                self._ready_evt.set()
+                raise

main/xiaozhi-server/core/providers/tools/server_mcp/mcp_executor.py

@@ -0,0 +1,89 @@
+"""服务端MCP工具执行器"""
+
+from typing import Dict, Any, Optional
+from ..base import ToolType, ToolDefinition, ToolExecutor
+from plugins_func.register import Action, ActionResponse
+from .mcp_manager import ServerMCPManager
+
+
+class ServerMCPExecutor(ToolExecutor):
+    """服务端MCP工具执行器"""
+
+    def __init__(self, conn):
+        self.conn = conn
+        self.mcp_manager: Optional[ServerMCPManager] = None
+        self._initialized = False
+
+    async def initialize(self):
+        """初始化MCP管理器"""
+        if not self._initialized:
+            self.mcp_manager = ServerMCPManager(self.conn)
+            await self.mcp_manager.initialize_servers()
+            self._initialized = True
+
+    async def execute(
+        self, conn, tool_name: str, arguments: Dict[str, Any]
+    ) -> ActionResponse:
+        """执行服务端MCP工具"""
+        if not self._initialized or not self.mcp_manager:
+            return ActionResponse(
+                action=Action.ERROR,
+                response="MCP管理器未初始化",
+            )
+
+        try:
+            # 移除mcp_前缀（如果有）
+            actual_tool_name = tool_name
+            if tool_name.startswith("mcp_"):
+                actual_tool_name = tool_name[4:]
+
+            result = await self.mcp_manager.execute_tool(actual_tool_name, arguments)
+
+            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 self._initialized or not self.mcp_manager:
+            return {}
+
+        tools = {}
+        mcp_tools = self.mcp_manager.get_all_tools()
+
+        for tool in mcp_tools:
+            func_def = tool.get("function", {})
+            tool_name = func_def.get("name", "")
+            if tool_name == "":
+                continue
+            tools[tool_name] = ToolDefinition(
+                name=tool_name, description=tool, tool_type=ToolType.SERVER_MCP
+            )
+
+        return tools
+
+    def has_tool(self, tool_name: str) -> bool:
+        """检查是否有指定的服务端MCP工具"""
+        if not self._initialized or not self.mcp_manager:
+            return False
+
+        # 移除mcp_前缀（如果有）
+        actual_tool_name = tool_name
+        if tool_name.startswith("mcp_"):
+            actual_tool_name = tool_name[4:]
+
+        return self.mcp_manager.is_mcp_tool(actual_tool_name)
+
+    async def cleanup(self):
+        """清理MCP连接"""
+        if self.mcp_manager:
+            await self.mcp_manager.cleanup_all()

main/xiaozhi-server/core/providers/tools/server_mcp/mcp_manager.py

@@ -0,0 +1,158 @@
+"""服务端MCP管理器"""
+
+import asyncio
+import os
+import json
+from typing import Dict, Any, List
+from config.config_loader import get_project_dir
+from config.logger import setup_logging
+from .mcp_client import ServerMCPClient
+
+TAG = __name__
+logger = setup_logging()
+
+
+class ServerMCPManager:
+    """管理多个服务端MCP服务的集中管理器"""
+
+    def __init__(self, conn) -> None:
+        """初始化MCP管理器"""
+        self.conn = conn
+        self.config_path = get_project_dir() + "data/.mcp_server_settings.json"
+        if not os.path.exists(self.config_path):
+            self.config_path = ""
+            logger.bind(tag=TAG).warning(
+                f"请检查mcp服务配置文件：data/.mcp_server_settings.json"
+            )
+        self.clients: Dict[str, ServerMCPClient] = {}
+        self.tools = []
+
+    def load_config(self) -> Dict[str, Any]:
+        """加载MCP服务配置"""
+        if len(self.config_path) == 0:
+            return {}
+
+        try:
+            with open(self.config_path, "r", encoding="utf-8") as f:
+                config = json.load(f)
+            return config.get("mcpServers", {})
+        except Exception as e:
+            logger.bind(tag=TAG).error(
+                f"Error loading MCP config from {self.config_path}: {e}"
+            )
+            return {}
+
+    async def initialize_servers(self) -> None:
+        """初始化所有MCP服务"""
+        config = self.load_config()
+        for name, srv_config in config.items():
+            if not srv_config.get("command") and not srv_config.get("url"):
+                logger.bind(tag=TAG).warning(
+                    f"Skipping server {name}: neither command nor url specified"
+                )
+                continue
+
+            try:
+                # 初始化服务端MCP客户端
+                logger.bind(tag=TAG).info(f"初始化服务端MCP客户端: {name}")
+                client = ServerMCPClient(srv_config)
+                await client.initialize()
+                self.clients[name] = client
+                client_tools = client.get_available_tools()
+                self.tools.extend(client_tools)
+
+            except Exception as e:
+                logger.bind(tag=TAG).error(
+                    f"Failed to initialize MCP server {name}: {e}"
+                )
+
+        # 输出当前支持的服务端MCP工具列表
+        if hasattr(self.conn, "func_handler") and self.conn.func_handler:
+            self.conn.func_handler.current_support_functions()
+
+    def get_all_tools(self) -> List[Dict[str, Any]]:
+        """获取所有服务的工具function定义"""
+        return self.tools
+
+    def is_mcp_tool(self, tool_name: str) -> bool:
+        """检查是否是MCP工具"""
+        for tool in self.tools:
+            if (
+                tool.get("function") is not None
+                and tool["function"].get("name") == tool_name
+            ):
+                return True
+        return False
+
+    async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
+        """执行工具调用，失败时会尝试重新连接"""
+        logger.bind(tag=TAG).info(f"执行服务端MCP工具 {tool_name}，参数: {arguments}")
+
+        max_retries = 3  # 最大重试次数
+        retry_interval = 2  # 重试间隔(秒)
+
+        # 找到对应的客户端
+        client_name = None
+        target_client = None
+        for name, client in self.clients.items():
+            if client.has_tool(tool_name):
+                client_name = name
+                target_client = client
+                break
+
+        if not target_client:
+            raise ValueError(f"工具 {tool_name} 在任意MCP服务中未找到")
+
+        # 带重试机制的工具调用
+        for attempt in range(max_retries):
+            try:
+                return await target_client.call_tool(tool_name, arguments)
+            except Exception as e:
+                # 最后一次尝试失败时直接抛出异常
+                if attempt == max_retries - 1:
+                    raise
+
+                logger.bind(tag=TAG).warning(
+                    f"执行工具 {tool_name} 失败 (尝试 {attempt+1}/{max_retries}): {e}"
+                )
+
+                # 尝试重新连接
+                logger.bind(tag=TAG).info(
+                    f"重试前尝试重新连接 MCP 客户端 {client_name}"
+                )
+                try:
+                    # 关闭旧的连接
+                    await target_client.cleanup()
+
+                    # 重新初始化客户端
+                    config = self.load_config()
+                    if client_name in config:
+                        client = ServerMCPClient(config[client_name])
+                        await client.initialize()
+                        self.clients[client_name] = client
+                        target_client = client
+                        logger.bind(tag=TAG).info(
+                            f"成功重新连接 MCP 客户端: {client_name}"
+                        )
+                    else:
+                        logger.bind(tag=TAG).error(
+                            f"Cannot reconnect MCP client {client_name}: config not found"
+                        )
+                except Exception as reconnect_error:
+                    logger.bind(tag=TAG).error(
+                        f"Failed to reconnect MCP client {client_name}: {reconnect_error}"
+                    )
+
+                # 等待一段时间再重试
+                await asyncio.sleep(retry_interval)
+
+    async def cleanup_all(self) -> None:
+        """关闭所有 MCP客户端"""
+        for name, client in list(self.clients.items()):
+            try:
+                if hasattr(client, "cleanup"):
+                    await asyncio.wait_for(client.cleanup(), timeout=20)
+                logger.bind(tag=TAG).info(f"服务端MCP客户端已关闭: {name}")
+            except (asyncio.TimeoutError, Exception) as e:
+                logger.bind(tag=TAG).error(f"关闭服务端MCP客户端 {name} 时出错: {e}")
+        self.clients.clear()

main/xiaozhi-server/core/providers/tools/server_plugins/__init__.py

@@ -0,0 +1,5 @@
+"""服务端插件工具模块"""
+
+from .plugin_executor import ServerPluginExecutor
+
+__all__ = ["ServerPluginExecutor"]

main/xiaozhi-server/core/providers/tools/server_plugins/plugin_executor.py

@@ -0,0 +1,84 @@
+"""服务端插件工具执行器"""
+
+from typing import Dict, Any
+from ..base import ToolType, ToolDefinition, ToolExecutor
+from plugins_func.register import all_function_registry, Action, ActionResponse
+
+
+class ServerPluginExecutor(ToolExecutor):
+    """服务端插件工具执行器"""
+
+    def __init__(self, conn):
+        self.conn = conn
+        self.config = conn.config
+
+    async def execute(
+        self, conn, tool_name: str, arguments: Dict[str, Any]
+    ) -> ActionResponse:
+        """执行服务端插件工具"""
+        func_item = all_function_registry.get(tool_name)
+        if not func_item:
+            return ActionResponse(
+                action=Action.NOTFOUND, response=f"插件函数 {tool_name} 不存在"
+            )
+
+        try:
+            # 根据工具类型决定如何调用
+            if hasattr(func_item, "type"):
+                func_type = func_item.type
+                if func_type.code in [4, 5]:  # SYSTEM_CTL, IOT_CTL (需要conn参数)
+                    result = func_item.func(conn, **arguments)
+                elif func_type.code == 2:  # WAIT
+                    result = func_item.func(**arguments)
+                elif func_type.code == 3:  # CHANGE_SYS_PROMPT
+                    result = func_item.func(conn, **arguments)
+                else:
+                    result = func_item.func(**arguments)
+            else:
+                # 默认不传conn参数
+                result = func_item.func(**arguments)
+
+            return result
+
+        except Exception as e:
+            return ActionResponse(
+                action=Action.ERROR,
+                response=str(e),
+            )
+
+    def get_tools(self) -> Dict[str, ToolDefinition]:
+        """获取所有注册的服务端插件工具"""
+        tools = {}
+
+        # 获取必要的函数
+        necessary_functions = ["handle_exit_intent", "get_time", "get_lunar"]
+
+        # 获取配置中的函数
+        config_functions = self.config["Intent"][
+            self.config["selected_module"]["Intent"]
+        ].get("functions", [])
+        
+        # 转换为列表
+        if not isinstance(config_functions, list):
+            try:
+                config_functions = list(config_functions)
+            except TypeError:
+                config_functions = []
+
+        # 合并所有需要的函数
+        all_required_functions = list(set(necessary_functions + config_functions))
+
+        for func_name in all_required_functions:
+            func_item = all_function_registry.get(func_name)
+            if func_item:
+                tools[func_name] = ToolDefinition(
+                    name=func_name,
+                    description=func_item.description,
+                    tool_type=ToolType.SERVER_PLUGIN,
+                )
+
+        return tools
+
+    def has_tool(self, tool_name: str) -> bool:
+        """检查是否有指定的服务端插件工具"""
+        return tool_name in all_function_registry

main/xiaozhi-server/core/providers/tools/unified_tool_handler.py

@@ -0,0 +1,235 @@
+"""统一工具处理器"""
+
+import json
+from typing import Dict, List, Any, Optional
+from config.logger import setup_logging
+from plugins_func.loadplugins import auto_import_modules
+
+from .base import ToolType
+from plugins_func.register import Action, ActionResponse
+from .unified_tool_manager import ToolManager
+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:
+    """统一工具处理器"""
+
+    def __init__(self, conn):
+        self.conn = conn
+        self.config = conn.config
+        self.logger = setup_logging()
+
+        # 创建工具管理器
+        self.tool_manager = ToolManager(conn)
+
+        # 创建各类执行器
+        self.server_plugin_executor = ServerPluginExecutor(conn)
+        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(
+            ToolType.SERVER_PLUGIN, self.server_plugin_executor
+        )
+        self.tool_manager.register_executor(
+            ToolType.SERVER_MCP, self.server_mcp_executor
+        )
+        self.tool_manager.register_executor(
+            ToolType.DEVICE_IOT, self.device_iot_executor
+        )
+        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
+
+    async def _initialize(self):
+        """异步初始化"""
+        try:
+            # 自动导入插件模块
+            auto_import_modules("plugins_func.functions")
+
+            # 初始化服务端MCP
+            await self.server_mcp_executor.initialize()
+
+            # 初始化MCP接入点
+            await self._initialize_mcp_endpoint()
+
+            # 初始化Home Assistant（如果需要）
+            self._initialize_home_assistant()
+
+            self.finish_init = True
+            self.logger.info("统一工具处理器初始化完成")
+
+            # 输出当前支持的所有工具列表
+            self.current_support_functions()
+
+        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
+                and mcp_endpoint_url != "null"
+            ):
+                self.logger.info(f"正在初始化MCP接入点: {mcp_endpoint_url}")
+                mcp_endpoint_client = await connect_mcp_endpoint(
+                    mcp_endpoint_url, self.conn
+                )
+
+                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:
+            from plugins_func.functions.hass_init import append_devices_to_prompt
+
+            append_devices_to_prompt(self.conn)
+        except ImportError:
+            pass  # 忽略导入错误
+        except Exception as e:
+            self.logger.error(f"初始化Home Assistant失败: {e}")
+
+    def get_functions(self) -> List[Dict[str, Any]]:
+        """获取所有工具的函数描述"""
+        return self.tool_manager.get_function_descriptions()
+
+    def current_support_functions(self) -> List[str]:
+        """获取当前支持的函数名称列表"""
+        func_names = self.tool_manager.get_supported_tool_names()
+        self.logger.info(f"当前支持的函数列表: {func_names}")
+        return func_names
+
+    def upload_functions_desc(self):
+        """刷新函数描述列表"""
+        self.tool_manager.refresh_tools()
+        self.logger.info("函数描述列表已刷新")
+
+    def has_tool(self, tool_name: str) -> bool:
+        """检查是否有指定工具"""
+        return self.tool_manager.has_tool(tool_name)
+
+    async def handle_llm_function_call(
+        self, conn, function_call_data: Dict[str, Any]
+    ) -> Optional[ActionResponse]:
+        """处理LLM函数调用"""
+        try:
+            # 处理多函数调用
+            if "function_calls" in function_call_data:
+                responses = []
+                for call in function_call_data["function_calls"]:
+                    result = await self.tool_manager.execute_tool(
+                        call["name"], call.get("arguments", {})
+                    )
+                    responses.append(result)
+                return self._combine_responses(responses)
+
+            # 处理单函数调用
+            function_name = function_call_data["name"]
+            arguments = function_call_data.get("arguments", {})
+
+            # 如果arguments是字符串，尝试解析为JSON
+            if isinstance(arguments, str):
+                try:
+                    arguments = json.loads(arguments) if arguments else {}
+                except json.JSONDecodeError:
+                    self.logger.error(f"无法解析函数参数: {arguments}")
+                    return ActionResponse(
+                        action=Action.ERROR,
+                        response="无法解析函数参数",
+                    )
+
+            self.logger.debug(f"调用函数: {function_name}, 参数: {arguments}")
+
+            # 执行工具调用
+            result = await self.tool_manager.execute_tool(function_name, arguments)
+            return result
+
+        except Exception as e:
+            self.logger.error(f"处理function call错误: {e}")
+            return ActionResponse(action=Action.ERROR, response=str(e))
+
+    def _combine_responses(self, responses: List[ActionResponse]) -> ActionResponse:
+        """合并多个函数调用的响应"""
+        if not responses:
+            return ActionResponse(action=Action.NONE, response="无响应")
+
+        # 如果有任何错误，返回第一个错误
+        for response in responses:
+            if response.action == Action.ERROR:
+                return response
+
+        # 合并所有成功的响应
+        contents = []
+        responses_text = []
+
+        for response in responses:
+            if response.content:
+                contents.append(response.content)
+            if response.response:
+                responses_text.append(response.response)
+
+        # 确定最终的动作类型
+        final_action = Action.RESPONSE
+        for response in responses:
+            if response.action == Action.REQLLM:
+                final_action = Action.REQLLM
+                break
+
+        return ActionResponse(
+            action=final_action,
+            result="; ".join(contents) if contents else None,
+            response="; ".join(responses_text) if responses_text else None,
+        )
+
+    async def register_iot_tools(self, descriptors: List[Dict[str, Any]]):
+        """注册IoT设备工具"""
+        self.device_iot_executor.register_iot_tools(descriptors)
+        self.tool_manager.refresh_tools()
+        self.logger.info(f"注册了{len(descriptors)}个IoT设备的工具")
+
+    def get_tool_statistics(self) -> Dict[str, int]:
+        """获取工具统计信息"""
+        return self.tool_manager.get_tool_statistics()
+
+    async def cleanup(self):
+        """清理资源"""
+        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}")

main/xiaozhi-server/core/providers/tools/unified_tool_manager.py

@@ -0,0 +1,124 @@
+"""统一工具管理器"""
+
+from typing import Dict, List, Optional, Any
+from config.logger import setup_logging
+from plugins_func.register import Action, ActionResponse
+from .base import ToolType, ToolDefinition, ToolExecutor
+
+
+class ToolManager:
+    """统一工具管理器，管理所有类型的工具"""
+
+    def __init__(self, conn):
+        self.conn = conn
+        self.logger = setup_logging()
+        self.executors: Dict[ToolType, ToolExecutor] = {}
+        self._cached_tools: Optional[Dict[str, ToolDefinition]] = None
+        self._cached_function_descriptions: Optional[List[Dict[str, Any]]] = None
+
+    def register_executor(self, tool_type: ToolType, executor: ToolExecutor):
+        """注册工具执行器"""
+        self.executors[tool_type] = executor
+        self._invalidate_cache()
+        self.logger.info(f"注册工具执行器: {tool_type.value}")
+
+    def _invalidate_cache(self):
+        """使缓存失效"""
+        self._cached_tools = None
+        self._cached_function_descriptions = None
+
+    def get_all_tools(self) -> Dict[str, ToolDefinition]:
+        """获取所有工具定义"""
+        if self._cached_tools is not None:
+            return self._cached_tools
+
+        all_tools = {}
+        for tool_type, executor in self.executors.items():
+            try:
+                tools = executor.get_tools()
+                for name, definition in tools.items():
+                    if name in all_tools:
+                        self.logger.warning(f"工具名称冲突: {name}")
+                    all_tools[name] = definition
+            except Exception as e:
+                self.logger.error(f"获取{tool_type.value}工具时出错: {e}")
+
+        self._cached_tools = all_tools
+        return all_tools
+
+    def get_function_descriptions(self) -> List[Dict[str, Any]]:
+        """获取所有工具的函数描述（OpenAI格式）"""
+        if self._cached_function_descriptions is not None:
+            return self._cached_function_descriptions
+
+        descriptions = []
+        tools = self.get_all_tools()
+        for tool_definition in tools.values():
+            descriptions.append(tool_definition.description)
+
+        self._cached_function_descriptions = descriptions
+        return descriptions
+
+    def has_tool(self, tool_name: str) -> bool:
+        """检查是否存在指定工具"""
+        tools = self.get_all_tools()
+        return tool_name in tools
+
+    def get_tool_type(self, tool_name: str) -> Optional[ToolType]:
+        """获取工具类型"""
+        tools = self.get_all_tools()
+        tool_def = tools.get(tool_name)
+        return tool_def.tool_type if tool_def else None
+
+    async def execute_tool(
+        self, tool_name: str, arguments: Dict[str, Any]
+    ) -> ActionResponse:
+        """执行工具调用"""
+        try:
+            # 查找工具类型
+            tool_type = self.get_tool_type(tool_name)
+            if not tool_type:
+                return ActionResponse(
+                    action=Action.NOTFOUND,
+                    response=f"工具 {tool_name} 不存在",
+                )
+
+            # 获取对应的执行器
+            executor = self.executors.get(tool_type)
+            if not executor:
+                return ActionResponse(
+                    action=Action.ERROR,
+                    response=f"工具类型 {tool_type.value} 的执行器未注册",
+                )
+
+            # 执行工具
+            self.logger.info(f"执行工具: {tool_name}，参数: {arguments}")
+            result = await executor.execute(self.conn, tool_name, arguments)
+            self.logger.debug(f"工具执行结果: {result}")
+            return result
+
+        except Exception as e:
+            self.logger.error(f"执行工具 {tool_name} 时出错: {e}")
+            return ActionResponse(action=Action.ERROR, response=str(e))
+
+    def get_supported_tool_names(self) -> List[str]:
+        """获取所有支持的工具名称"""
+        tools = self.get_all_tools()
+        return list(tools.keys())
+
+    def refresh_tools(self):
+        """刷新工具缓存"""
+        self._invalidate_cache()
+        self.logger.info("工具缓存已刷新")
+
+    def get_tool_statistics(self) -> Dict[str, int]:
+        """获取工具统计信息"""
+        stats = {}
+        for tool_type, executor in self.executors.items():
+            try:
+                tools = executor.get_tools()
+                stats[tool_type.value] = len(tools)
+            except Exception as e:
+                self.logger.error(f"获取{tool_type.value}工具统计时出错: {e}")
+                stats[tool_type.value] = 0
+        return stats

main/xiaozhi-server/core/providers/tts/base.py

@@ -43,7 +43,6 @@ class TTSProviderBase(ABC):
         self.tts_text_buff = []
         self.punctuations = (
             "。",
-            ".",
             "？",
             "?",
             "！",
@@ -59,7 +58,6 @@ class TTSProviderBase(ABC):
             "、",
             ",",
             "。",
-            ".",
             "？",
             "?",
             "！",
@@ -171,7 +169,7 @@ class TTSProviderBase(ABC):
             )
         )
         # 对于单句的文本，进行分段处理
-        segments = re.split(r'([。！？!?；;\n])', content_detail)
+        segments = re.split(r"([。！？!?；;\n])", content_detail)
         for seg in segments:
             self.tts_text_queue.put(
                 TTSMessageDTO(

main/xiaozhi-server/core/providers/tts/doubao.py

@@ -37,7 +37,9 @@ class TTSProvider(TTSProviderBase):
         self.api_url = config.get("api_url")
         self.authorization = config.get("authorization")
         self.header = {"Authorization": f"{self.authorization}{self.access_token}"}
-        check_model_key("TTS", self.access_token)
+        model_key_msg = check_model_key("TTS", self.access_token)
+        if model_key_msg:
+            logger.bind(tag=TAG).error(model_key_msg)
 
     async def text_to_speak(self, text, output_file):
         request_json = {

main/xiaozhi-server/core/providers/tts/fishspeech.py

@@ -94,8 +94,9 @@ class TTSProvider(TTSProviderBase):
         self.format = config.get("response_format", "wav")
         self.audio_file_type = config.get("response_format", "wav")
         self.api_key = config.get("api_key", "YOUR_API_KEY")
-        have_key = check_model_key("FishSpeech TTS", self.api_key)
-        if not have_key:
+        model_key_msg = check_model_key("FishSpeech TTS", self.api_key)
+        if model_key_msg:
+            logger.bind(tag=TAG).error(model_key_msg)
             return
         self.normalize = str(config.get("normalize", True)).lower() in (
             "true",

main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py

@@ -12,6 +12,8 @@ from core.utils.util import check_model_key
 from core.providers.tts.base import TTSProviderBase
 from core.handle.abortHandle import handleAbortMessage
 from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType
+from asyncio import Task
+
 
 TAG = __name__
 logger = setup_logging()
@@ -141,6 +143,7 @@ class TTSProvider(TTSProviderBase):
         super().__init__(config, delete_audio_file)
         self.ws = None
         self.interface_type = InterfaceType.DUAL_STREAM
+        self._monitor_task = None  # 监听任务引用
         self.appId = config.get("appid")
         self.access_token = config.get("access_token")
         self.cluster = config.get("cluster")
@@ -157,7 +160,9 @@ class TTSProvider(TTSProviderBase):
         self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
             sample_rate=16000, channels=1, frame_size_ms=60
         )
-        check_model_key("TTS", self.access_token)
+        model_key_msg = check_model_key("TTS", self.access_token)
+        if model_key_msg:
+            logger.bind(tag=TAG).error(model_key_msg)
 
     async def open_audio_channels(self, conn):
         try:
@@ -202,6 +207,10 @@ class TTSProvider(TTSProviderBase):
                 if message.sentence_type == SentenceType.FIRST:
                     # 初始化参数
                     try:
+                        if not getattr(self.conn, "sentence_id", None): 
+                            self.conn.sentence_id = uuid.uuid4().hex
+                            logger.bind(tag=TAG).info(f"自动生成新的 会话ID: {self.conn.sentence_id}")
+
                         logger.bind(tag=TAG).info("开始启动TTS会话...")
                         future = asyncio.run_coroutine_threadsafe(
                             self.start_session(self.conn.sentence_id),
@@ -264,10 +273,9 @@ class TTSProvider(TTSProviderBase):
         try:
             # 建立新连接
             if self.ws is None:
-                await handleAbortMessage(self.conn)
-                logger.bind(tag=TAG).error(f"WebSocket连接不存在，终止发送文本")
+                logger.bind(tag=TAG).warning(f"WebSocket连接不存在，终止发送文本")
                 return
-            
+
             #  过滤Markdown
             filtered_text = MarkdownCleaner.clean_markdown(text)
 
@@ -287,6 +295,25 @@ class TTSProvider(TTSProviderBase):
     async def start_session(self, session_id):
         logger.bind(tag=TAG).info(f"开始会话～～{session_id}")
         try:
+            task = self._monitor_task
+            if (
+                task is not None
+                and isinstance(task, Task)
+                and not task.done()
+            ):
+                logger.bind(tag=TAG).info("等待上一个监听任务结束...")
+                if self.ws is not None:
+                    logger.bind(tag=TAG).info("强制关闭上一个WebSocket连接以唤醒监听任务...")
+                    try:
+                        await self.ws.close()
+                    except Exception as e:
+                        logger.bind(tag=TAG).warning(f"关闭上一个ws异常: {e}")
+                    self.ws = None
+                try:
+                    await asyncio.wait_for(task, timeout=8)
+                except Exception as e:
+                    logger.bind(tag=TAG).warning(f"等待监听任务异常: {e}")
+                self._monitor_task = None
             # 建立新连接
             await self._ensure_connection()
 
@@ -457,6 +484,8 @@ class TTSProvider(TTSProviderBase):
                 except:
                     pass
                 self.ws = None
+            # 监听任务退出时清理引用
+            self._monitor_task = None
 
     async def send_event(
         self,

main/xiaozhi-server/core/providers/tts/linkerai.py

@@ -166,7 +166,9 @@ class TTSProvider(TTSProviderBase):
                 ) as resp:
 
                     if resp.status != 200:
-                        logger.error(f"TTS请求失败: {resp.status}, {await resp.text()}")
+                        logger.bind(tag=TAG).error(
+                            f"TTS请求失败: {resp.status}, {await resp.text()}"
+                        )
                         self.tts_audio_queue.put((SentenceType.LAST, [], None))
                         return
 
@@ -229,7 +231,7 @@ class TTSProvider(TTSProviderBase):
                         self._process_before_stop_play_files()
 
         except Exception as e:
-            logger.error(f"TTS请求异常: {e}")
+            logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
             self.tts_audio_queue.put((SentenceType.LAST, [], None))
 
     def to_tts(self, text: str) -> list:
@@ -263,7 +265,7 @@ class TTSProvider(TTSProviderBase):
                 self.api_url, params=params, headers=headers, timeout=5
             ) as response:
                 if response.status_code != 200:
-                    logger.error(
+                    logger.bind(tag=TAG).error(
                         f"TTS请求失败: {response.status_code}, {response.text}"
                     )
                     return []
@@ -299,5 +301,5 @@ class TTSProvider(TTSProviderBase):
                 return opus_datas
 
         except Exception as e:
-            logger.error(f"TTS请求异常: {e}")
+            logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
             return []

main/xiaozhi-server/core/providers/tts/openai.py

@@ -25,7 +25,9 @@ class TTSProvider(TTSProviderBase):
         self.speed = float(speed) if speed else 1.0
 
         self.output_file = config.get("output_dir", "tmp/")
-        check_model_key("TTS", self.api_key)
+        model_key_msg = check_model_key("TTS", self.api_key)
+        if model_key_msg:
+            logger.bind(tag=TAG).error(model_key_msg)
 
     async def text_to_speak(self, text, output_file):
         headers = {

main/xiaozhi-server/core/providers/vad/silero.py

@@ -35,6 +35,10 @@ class VADProvider(VADProviderBase):
             pcm_frame = self.decoder.decode(opus_packet, 960)
             conn.client_audio_buffer.extend(pcm_frame)  # 将新数据加入缓冲区
 
+            # 确保帧计数器存在
+            if not hasattr(conn, "client_voice_frame_count"):
+                conn.client_voice_frame_count = 0
+
             # 处理缓冲区中的完整帧（每次处理512采样点）
             client_have_voice = False
             while len(conn.client_audio_buffer) >= 512 * 2:
@@ -50,18 +54,24 @@ class VADProvider(VADProviderBase):
                 # 检测语音活动
                 with torch.no_grad():
                     speech_prob = self.model(audio_tensor, 16000).item()
-                client_have_voice = speech_prob >= self.vad_threshold
+                is_voice = speech_prob >= self.vad_threshold
+
+                if is_voice:
+                    conn.client_voice_frame_count += 1
+                else:
+                    conn.client_voice_frame_count = 0
+
+                # 只有连续4帧检测到语音才认为有语音
+                client_have_voice = conn.client_voice_frame_count >= 4
 
                 # 如果之前有声音，但本次没有声音，且与上次有声音的时间差已经超过了静默阈值，则认为已经说完一句话
                 if conn.client_have_voice and not client_have_voice:
-                    stop_duration = (
-                        time.time() * 1000 - conn.client_have_voice_last_time
-                    )
+                    stop_duration = time.time() * 1000 - conn.last_activity_time
                     if stop_duration >= self.silence_threshold_ms:
                         conn.client_voice_stop = True
                 if client_have_voice:
                     conn.client_have_voice = True
-                    conn.client_have_voice_last_time = time.time() * 1000
+                    conn.last_activity_time = time.time() * 1000
 
             return client_have_voice
         except opuslib_next.OpusError as e:

main/xiaozhi-server/core/providers/vllm/openai.py

@@ -34,7 +34,9 @@ class VLLMProvider(VLLMProviderBase):
             except (ValueError, TypeError):
                 setattr(self, param, default)
 
-        check_model_key("VLLM", self.api_key)
+        model_key_msg = check_model_key("VLLM", self.api_key)
+        if model_key_msg:
+            logger.bind(tag=TAG).error(model_key_msg)
         self.client = openai.OpenAI(api_key=self.api_key, base_url=self.base_url)
 
     def response(self, question, base64_image):