diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 98cf5c5c..16e1432d 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/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: @@ -203,6 +206,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，替换原来的系统提示 @@ -324,7 +330,6 @@ class ConnectionHandler: functions = None if hasattr(self, 'func_handler'): functions = self.func_handler.get_functions() - response_message = [] processed_chars = 0 # 跟踪已处理的字符位置 @@ -436,8 +441,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) @@ -459,6 +470,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 @@ -582,6 +629,8 @@ class ConnectionHandler: async def close(self, ws=None): """资源清理方法""" + # 清理MCP资源 + await self.mcp_manager.cleanup_all() # 触发停止事件并清理资源 if self.stop_event: diff --git a/main/xiaozhi-server/core/mcp/MCPClient.py b/main/xiaozhi-server/core/mcp/MCPClient.py new file mode 100644 index 00000000..3c32e077 --- /dev/null +++ b/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=10) + 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() \ No newline at end of file diff --git a/main/xiaozhi-server/core/mcp/manager.py b/main/xiaozhi-server/core/mcp/manager.py new file mode 100644 index 00000000..28efd87d --- /dev/null +++ b/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() diff --git a/main/xiaozhi-server/mcp_server_settings.json b/main/xiaozhi-server/mcp_server_settings.json new file mode 100644 index 00000000..a38bbdfb --- /dev/null +++ b/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" + } + } +} diff --git a/main/xiaozhi-server/plugins_func/register.py b/main/xiaozhi-server/plugins_func/register.py index ccb03b44..fd990760 100644 --- a/main/xiaozhi-server/plugins_func/register.py +++ b/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 diff --git a/main/xiaozhi-server/requirements.txt b/main/xiaozhi-server/requirements.txt index 51433cb9..164d594d 100755 --- a/main/xiaozhi-server/requirements.txt +++ b/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 \ No newline at end of file +mcp==1.4.1 + +cnlunar==0.2.0