Merge pull request #492 from xinnan-tech/mcp-client

增加mcp的tool功能，可自行配置mcp服务，实现命令控制

main/xiaozhi-server/core/connection.py

@@ -18,10 +18,11 @@ from concurrent.futures import ThreadPoolExecutor, TimeoutError
 from core.handle.sendAudioHandle import sendAudioMessage
 from core.handle.receiveAudioHandle import handleAudioMessage
 from core.handle.functionHandler import FunctionHandler
-from plugins_func.register import Action
+from plugins_func.register import Action, ActionResponse
 from config.private_config import PrivateConfig
 from core.auth import AuthMiddleware, AuthenticationError
 from core.utils.auth_code_gen import AuthCodeGenerator
+from core.mcp.manager import MCPManager
 
 TAG = __name__
 
@@ -100,6 +101,8 @@ class ConnectionHandler:
         self.use_function_call_mode = False
         if self.config["selected_module"]["Intent"] == 'function_call':
             self.use_function_call_mode = True
+        
+        self.mcp_manager = MCPManager(self)
 
     async def handle_connection(self, ws):
         try:
@@ -227,6 +230,9 @@ class ConnectionHandler:
             self.prompt = self.prompt + f"\nuser location:{self.client_ip_info}"
             self.dialogue.update_system_message(self.prompt)
 
+        """加载MCP工具"""
+        asyncio.run_coroutine_threadsafe(self.mcp_manager.initialize_servers(), self.loop)
+
     def change_system_prompt(self, prompt):
         self.prompt = prompt
         # 找到原来的role==system，替换原来的系统提示
@@ -348,7 +354,6 @@ class ConnectionHandler:
         functions = None
         if hasattr(self, 'func_handler'):
             functions = self.func_handler.get_functions()
-
         response_message = []
         processed_chars = 0  # 跟踪已处理的字符位置
 
@@ -463,8 +468,14 @@ class ConnectionHandler:
                     "id": function_id,
                     "arguments": function_arguments
                 }
-                result = self.func_handler.handle_llm_function_call(self, function_call_data)
-                self._handle_function_result(result, function_call_data, text_index + 1)
+
+                # 处理MCP工具调用
+                if self.mcp_manager.is_mcp_tool(function_name):
+                    result = self._handle_mcp_tool_call(function_call_data)
+                else:
+                    # 处理系统函数
+                    result = self.func_handler.handle_llm_function_call(self, function_call_data)
+                self._handle_function_result(result, function_call_data, text_index+1)
 
         # 处理最后剩余的文本
         full_text = "".join(response_message)
@@ -486,6 +497,42 @@ class ConnectionHandler:
 
         return True
 
+    def _handle_mcp_tool_call(self, function_call_data):
+        function_arguments = function_call_data["arguments"]
+        function_name = function_call_data["name"]
+        try:
+            args_dict = function_arguments
+            if isinstance(function_arguments, str):
+                try:
+                    args_dict = json.loads(function_arguments)
+                except json.JSONDecodeError:
+                    self.logger.bind(tag=TAG).error(f"无法解析 function_arguments: {function_arguments}")
+                    return ActionResponse(action=Action.REQLLM, result="参数解析失败", response="")
+                    
+            tool_result = asyncio.run_coroutine_threadsafe(self.mcp_manager.execute_tool(
+                function_name,
+                args_dict
+            ), self.loop).result()
+            # meta=None content=[TextContent(type='text', text='北京当前天气:\n温度: 21°C\n天气: 晴\n湿度: 6%\n风向: 西北 风\n风力等级: 5级', annotations=None)] isError=False
+            content_text = ""
+            if tool_result is not None and tool_result.content is not None:
+                for content in tool_result.content:
+                    content_type = content.type
+                    if content_type == "text":
+                        content_text = content.text
+                    elif content_type == "image":
+                        pass
+            
+            if len(content_text) > 0:
+                return ActionResponse(action=Action.REQLLM, result=content_text, response="")
+            
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"MCP工具调用错误: {e}")
+            return ActionResponse(action=Action.REQLLM, result="工具调用出错", response="")
+
+        return ActionResponse(action=Action.REQLLM, result="工具调用出错", response="")
+            
+
     def _handle_function_result(self, result, function_call_data, text_index):
         if result.action == Action.RESPONSE:  # 直接回复前端
             text = result.response
@@ -609,6 +656,8 @@ class ConnectionHandler:
 
     async def close(self, ws=None):
         """资源清理方法"""
+        # 清理MCP资源
+        await self.mcp_manager.cleanup_all()
 
         # 触发停止事件并清理资源
         if self.stop_event:

main/xiaozhi-server/core/mcp/MCPClient.py

@@ -0,0 +1,86 @@
+from datetime import timedelta
+from typing import Optional
+from contextlib import AsyncExitStack
+import os, shutil
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+from config.logger import setup_logging
+
+TAG = __name__
+
+class MCPClient:
+    def __init__(self, config):
+        # Initialize session and client objects
+        self.session: Optional[ClientSession] = None
+        self.exit_stack = AsyncExitStack()
+        self.logger = setup_logging()
+        self.config = config
+        self.tolls = []
+
+    async def initialize(self):
+        args = self.config.get("args", [])
+
+        command = (
+            shutil.which("npx")
+            if self.config["command"] == "npx"
+            else self.config["command"]
+        )
+        
+        env={**os.environ}
+        if self.config.get("env"):
+            env.update(self.config["env"])
+        
+        server_params = StdioServerParameters(
+            command=command,
+            args=args,
+            env=env
+        )
+        
+        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
+        self.stdio, self.write = stdio_transport
+        time_out_delta =  timedelta(seconds=15)
+        self.session = await self.exit_stack.enter_async_context(ClientSession(read_stream=self.stdio, write_stream=self.write, read_timeout_seconds=time_out_delta))
+        
+        await self.session.initialize()
+        
+        # List available tools
+        response = await self.session.list_tools()
+        tools = response.tools
+        self.tools = tools
+        self.logger.bind(tag=TAG).info(f"Connected to server with tools:{[tool.name for tool in tools]}")
+    
+    def has_tool(self, tool_name):
+        return any(tool.name == tool_name for tool in self.tools)
+    
+    def get_available_tools(self):
+        available_tools = [{"type": "function", "function":{ 
+            "name": tool.name,
+            "description": tool.description,
+            "parameters": tool.inputSchema
+        } } for tool in self.tools]
+
+        return available_tools
+    
+    async def call_tool(self, tool_name: str, tool_args: dict):
+        self.logger.bind(tag=TAG).info(f"MCPClient Calling tool {tool_name} with args: {tool_args}")
+        try:
+            response = await self.session.call_tool(tool_name, tool_args)
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"Error calling tool {tool_name}: {e}")
+            from types import SimpleNamespace
+            error_content = SimpleNamespace(
+                type='text',
+                text=f"Error calling tool {tool_name}: {e}"
+            )
+            error_response = SimpleNamespace(
+                content=[error_content],
+                isError=True
+            )
+            return error_response
+        self.logger.bind(tag=TAG).info(f"MCPClient Response from tool {tool_name}: {response}")
+        return response
+
+    async def cleanup(self):
+        """Clean up resources"""
+        await self.exit_stack.aclose()

main/xiaozhi-server/core/mcp/manager.py

@@ -0,0 +1,110 @@
+"""MCP服务管理器"""
+import os, json
+from typing import Dict, Any, List
+from .MCPClient import MCPClient
+from config.logger import setup_logging
+from core.utils.util import get_project_dir
+from plugins_func.register import register_function, ActionResponse, Action, ToolType
+
+TAG = __name__
+
+class MCPManager:
+    """管理多个MCP服务的集中管理器"""
+
+    def __init__(self,conn) -> None:
+        """
+        初始化MCP管理器
+        """
+        self.conn = conn
+        self.logger = setup_logging()
+        self.config_path = get_project_dir() + 'data/.mcp_server_settings.json'
+        if os.path.exists(self.config_path) == False:
+            self.config_path = ""
+            self.logger.bind(tag=TAG).warning(f"请检查mcp服务配置文件：data/.mcp_server_settings.json")
+        self.client: Dict[str, MCPClient] = {}
+        self.tools = []
+
+    def load_config(self) -> Dict[str, Any]:
+        """加载MCP服务配置
+        Returns:
+            Dict[str, Any]: 服务配置字典
+        """
+        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:
+            self.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"):
+                self.logger.bind(tag=TAG).warning(f"Skipping server {name}: command not specified")
+                continue
+
+            try:
+                client = MCPClient(srv_config)
+                await client.initialize()
+                self.client[name] = client
+                self.logger.bind(tag=TAG).info(f"Initialized MCP client: {name}")
+                client_tools = client.get_available_tools()
+                self.tools.extend(client_tools)
+                for tool in client_tools:
+                    func_name = "mcp_"+tool["function"]["name"]
+                    register_function(func_name, tool, ToolType.MCP_CLIENT)(self.execute_tool)
+                    self.conn.func_handler.function_registry.register_function(func_name)
+
+            except Exception as e:
+                self.logger.bind(tag=TAG).error(f"Failed to initialize MCP server {name}: {e}")
+        self.conn.func_handler.upload_functions_desc()
+
+    def get_all_tools(self) -> List[Dict[str, Any]]:
+        """获取所有服务的工具function定义
+        Returns:
+            List[Dict[str, Any]]: 所有工具的function定义列表
+        """
+        return self.tools
+
+    def is_mcp_tool(self, tool_name: str) -> bool:
+        """检查是否是MCP工具
+        Args:
+            tool_name: 工具名称
+        Returns:
+            bool: 是否是MCP工具
+        """
+        for tool in self.tools:
+            if tool.get("function") != 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:
+        """执行工具调用
+        Args:
+            tool_name: 工具名称
+            arguments: 工具参数 
+        Returns:
+            Any: 工具执行结果
+        Raises:
+            ValueError: 工具未找到时抛出
+        """
+        self.logger.bind(tag=TAG).info(f"Executing tool {tool_name} with arguments: {arguments}")
+        for client in self.client.values():
+            if client.has_tool(tool_name):
+                return await client.call_tool(tool_name, arguments)
+            
+        raise ValueError(f"Tool {tool_name} not found in any MCP server")
+
+    async def cleanup_all(self) -> None:
+        for name, client in self.client.items():
+            try:
+                await client.cleanup()
+                self.logger.bind(tag=TAG).info(f"Cleaned up MCP client: {name}")
+            except Exception as e:
+                self.logger.bind(tag=TAG).error(f"Error cleaning up MCP client {name}: {e}")
+        self.client.clear()

main/xiaozhi-server/mcp_server_settings.json

@@ -0,0 +1,31 @@
+{
+  "des": [
+    "在data目录下创建.mcp_server_settings.json文件，可以选择下面的MCP服务，也可以自行添加新的MCP服务。",
+    "后面不断测试补充好用的mcp服务，欢迎大家一起补充。",
+    "记得删除注释行,des属性仅为说明,不会被解析。",
+    "des和link属性，仅为说明安装方式，方便大家查看原始链接，不是必须项。"
+  ],
+  "mcpServers": {
+    "filesystem": {
+      "command": "npx",
+      "args": [
+        "-y",
+        "@modelcontextprotocol/server-filesystem",
+        "/Users/username/Desktop",
+        "/path/to/other/allowed/dir"
+      ],
+      "link":"https://github.com/modelcontextprotocol/servers/tree/main/src/filesystem"
+    },
+    "playwright": {
+      "command": "npx",
+      "args": ["-y", "@executeautomation/playwright-mcp-server"],
+      "des" : "run 'npx playwright install' first",
+      "link": "https://github.com/executeautomation/mcp-playwright"
+    },
+    "windows-cli": {
+      "command": "npx",
+      "args": ["-y", "@simonb97/server-win-cli"],
+      "link": "https://github.com/SimonB97/win-cli-mcp-server"
+    }
+  }
+}

main/xiaozhi-server/plugins_func/register.py

@@ -12,6 +12,7 @@ class ToolType(Enum):
     CHANGE_SYS_PROMPT = (3, "修改系统提示词，切换角色性格或职责")
     SYSTEM_CTL = (4, "系统控制，影响正常的对话流程，如退出、播放音乐等，需要传递conn参数")
     IOT_CTL = (5, "IOT设备控制，需要传递conn参数")
+    MCP_CLIENT = (6, "MCP客户端")
 
     def __init__(self, code, message):
         self.code = code

main/xiaozhi-server/requirements.txt

@@ -22,4 +22,6 @@ mem0ai==0.1.62
 bs4==0.0.2
 modelscope==1.23.2
 sherpa_onnx==1.11.0
-cnlunar==0.2.0
+mcp==1.4.1
+
+cnlunar==0.2.0