update: 单模块部署也能使用mcp接入点 (#1763)

* 优化:OTA接口需要适应最小示例参数 #1757

* update:修复智能体选择的模型不存在导致的500错误 #1752

* update: 单模块部署也能使用mcp接入点

main/xiaozhi-server/app.py

@@ -5,7 +5,7 @@ import asyncio
 from aioconsole import ainput
 from config.settings import load_config
 from config.logger import setup_logging
-from core.utils.util import get_local_ip
+from core.utils.util import get_local_ip, validate_mcp_endpoint
 from core.http_server import SimpleHttpServer
 from core.websocket_server import WebSocketServer
 from core.utils.util import check_ffmpeg_installed
@@ -77,6 +77,17 @@ async def main():
         get_local_ip(),
         port,
     )
+    mcp_endpoint = config.get("mcp_endpoint", None)
+    if mcp_endpoint is not None and "你" not in mcp_endpoint:
+        # 校验MCP接入点格式
+        if validate_mcp_endpoint(mcp_endpoint):
+            logger.bind(tag=TAG).info("mcp接入点是\t{}", mcp_endpoint)
+            # 将mcp计入点地址转成调用点
+            mcp_endpoint = mcp_endpoint.replace("/mcp/", "/call/")
+            config["mcp_endpoint"] = mcp_endpoint
+        else:
+            logger.bind(tag=TAG).error("mcp接入点不符合规范")
+            config["mcp_endpoint"] = "你的接入点 websocket地址"
 
     # 获取WebSocket配置，使用安全的默认值
     websocket_port = 8000

main/xiaozhi-server/config.yaml

@@ -104,7 +104,8 @@ wakeup_words:
   - "喵喵同学"
   - "小滨小滨"
   - "小冰小冰"
-# MCP接入点地址
+# MCP接入点地址，地址格式为：ws://你的mcp接入点ip或者域名:端口号/mcp/?token=你的token
+# 详细教程 https://github.com/xinnan-tech/xiaozhi-esp32-server/blob/main/docs/mcp-endpoint-integration.md
 mcp_endpoint: 你的接入点 websocket地址
 # 插件的基础配置
 plugins:

main/xiaozhi-server/core/connection.py

@@ -214,7 +214,17 @@ class ConnectionHandler:
             self.logger.bind(tag=TAG).error(f"Connection error: {str(e)}-{stack_trace}")
             return
         finally:
-            await self._save_and_close(ws)
+            try:
+                await self._save_and_close(ws)
+            except Exception as final_error:
+                self.logger.bind(tag=TAG).error(f"最终清理时出错: {final_error}")
+                # 确保即使保存记忆失败，也要关闭连接
+                try:
+                    await self.close(ws)
+                except Exception as close_error:
+                    self.logger.bind(tag=TAG).error(
+                        f"强制关闭连接时出错: {close_error}"
+                    )
 
     async def _save_and_close(self, ws):
         """保存记忆并关闭连接"""
@@ -232,7 +242,10 @@ class ConnectionHandler:
                     except Exception as e:
                         self.logger.bind(tag=TAG).error(f"保存记忆失败: {e}")
                     finally:
-                        loop.close()
+                        try:
+                            loop.close()
+                        except Exception:
+                            pass
 
                 # 启动线程保存记忆，不等待完成
                 threading.Thread(target=save_memory_task, daemon=True).start()
@@ -240,7 +253,12 @@ class ConnectionHandler:
             self.logger.bind(tag=TAG).error(f"保存记忆失败: {e}")
         finally:
             # 立即关闭连接，不等待记忆保存完成
-            await self.close(ws)
+            try:
+                await self.close(ws)
+            except Exception as close_error:
+                self.logger.bind(tag=TAG).error(
+                    f"保存记忆后关闭连接失败: {close_error}"
+                )
 
     async def _route_message(self, message):
         """消息路由"""
@@ -837,13 +855,22 @@ class ConnectionHandler:
         """资源清理方法"""
         try:
             # 取消超时任务
-            if self.timeout_task:
+            if self.timeout_task and not self.timeout_task.done():
                 self.timeout_task.cancel()
+                try:
+                    await self.timeout_task
+                except asyncio.CancelledError:
+                    pass
                 self.timeout_task = None
 
             # 清理工具处理器资源
             if hasattr(self, "func_handler") and self.func_handler:
-                await self.func_handler.cleanup()
+                try:
+                    await self.func_handler.cleanup()
+                except Exception as cleanup_error:
+                    self.logger.bind(tag=TAG).error(
+                        f"清理工具处理器时出错: {cleanup_error}"
+                    )
 
             # 触发停止事件
             if self.stop_event:
@@ -853,19 +880,58 @@ class ConnectionHandler:
             self.clear_queues()
 
             # 关闭WebSocket连接
-            if ws:
-                await ws.close()
-            elif self.websocket:
-                await self.websocket.close()
+            try:
+                if ws:
+                    # 安全地检查WebSocket状态并关闭
+                    try:
+                        if hasattr(ws, "closed") and not ws.closed:
+                            await ws.close()
+                        elif hasattr(ws, "state") and ws.state.name != "CLOSED":
+                            await ws.close()
+                        else:
+                            # 如果没有closed属性，直接尝试关闭
+                            await ws.close()
+                    except Exception:
+                        # 如果关闭失败，忽略错误
+                        pass
+                elif self.websocket:
+                    try:
+                        if (
+                            hasattr(self.websocket, "closed")
+                            and not self.websocket.closed
+                        ):
+                            await self.websocket.close()
+                        elif (
+                            hasattr(self.websocket, "state")
+                            and self.websocket.state.name != "CLOSED"
+                        ):
+                            await self.websocket.close()
+                        else:
+                            # 如果没有closed属性，直接尝试关闭
+                            await self.websocket.close()
+                    except Exception:
+                        # 如果关闭失败，忽略错误
+                        pass
+            except Exception as ws_error:
+                self.logger.bind(tag=TAG).error(f"关闭WebSocket连接时出错: {ws_error}")
 
             # 最后关闭线程池（避免阻塞）
             if self.executor:
-                self.executor.shutdown(wait=False)
+                try:
+                    self.executor.shutdown(wait=False)
+                except Exception as executor_error:
+                    self.logger.bind(tag=TAG).error(
+                        f"关闭线程池时出错: {executor_error}"
+                    )
                 self.executor = None
 
             self.logger.bind(tag=TAG).info("连接资源已释放")
         except Exception as e:
             self.logger.bind(tag=TAG).error(f"关闭连接时出错: {e}")
+        finally:
+            # 确保停止事件被设置
+            if self.stop_event:
+                self.stop_event.set()
 
     def clear_queues(self):
         """清空所有任务队列"""
@@ -922,9 +988,19 @@ class ConnectionHandler:
                     ):
                         if not self.stop_event.is_set():
                             self.logger.bind(tag=TAG).info("连接超时，准备关闭")
-                            await self.close(self.websocket)
+                            # 设置停止事件，防止重复处理
+                            self.stop_event.set()
+                            # 使用 try-except 包装关闭操作，确保不会因为异常而阻塞
+                            try:
+                                await self.close(self.websocket)
+                            except Exception as close_error:
+                                self.logger.bind(tag=TAG).error(
+                                    f"超时关闭连接时出错: {close_error}"
+                                )
                         break
                 # 每10秒检查一次，避免过于频繁
                 await asyncio.sleep(10)
         except Exception as e:
             self.logger.bind(tag=TAG).error(f"超时检查任务出错: {e}")
+        finally:
+            self.logger.bind(tag=TAG).info("超时检查任务已退出")

main/xiaozhi-server/core/utils/util.py

@@ -982,3 +982,28 @@ def sanitize_tool_name(name: str) -> str:
     """Sanitize tool names for OpenAI compatibility."""
     # 支持中文、英文字母、数字、下划线和连字符
     return re.sub(r"[^a-zA-Z0-9_\-\u4e00-\u9fff]", "_", name)
+
+
+def validate_mcp_endpoint(mcp_endpoint: str) -> bool:
+    """
+    校验MCP接入点格式
+
+    Args:
+        mcp_endpoint: MCP接入点字符串
+
+    Returns:
+        bool: 是否有效
+    """
+    # 1. 检查是否以ws开头
+    if not mcp_endpoint.startswith("ws"):
+        return False
+
+    # 2. 检查是否包含key、call字样
+    if "key" in mcp_endpoint.lower() or "call" in mcp_endpoint.lower():
+        return False
+
+    # 3. 检查是否包含/mcp/字样
+    if "/mcp/" not in mcp_endpoint:
+        return False
+
+    return True

main/xiaozhi-server/core/websocket_server.py

@@ -57,8 +57,25 @@ class WebSocketServer:
         self.active_connections.add(handler)
         try:
             await handler.handle_connection(websocket)
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"处理连接时出错: {e}")
         finally:
+            # 确保从活动连接集合中移除
             self.active_connections.discard(handler)
+            # 强制关闭连接（如果还没有关闭的话）
+            try:
+                # 安全地检查WebSocket状态并关闭
+                if hasattr(websocket, "closed") and not websocket.closed:
+                    await websocket.close()
+                elif hasattr(websocket, "state") and websocket.state.name != "CLOSED":
+                    await websocket.close()
+                else:
+                    # 如果没有closed属性，直接尝试关闭
+                    await websocket.close()
+            except Exception as close_error:
+                self.logger.bind(tag=TAG).error(
+                    f"服务器端强制关闭连接时出错: {close_error}"
+                )
 
     async def _http_response(self, websocket, request_headers):
         # 检查是否为 WebSocket 升级请求