From 8504c181c0b89f7876e9653a5012a4c29ef33584 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=84=E5=87=A4=E7=A7=91=E6=8A=80?= Date: Thu, 20 Mar 2025 08:59:45 +0800 Subject: [PATCH] =?UTF-8?q?mcp=E5=88=9D=E6=AD=A5=E8=B0=83=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 31 +++++- main/xiaozhi-server/core/mcp/MCPClient.py | 96 +++++++++++++++++ main/xiaozhi-server/core/mcp/manager.py | 102 ++++++++++++++++++ .../plugins_func/functions/get_time.py | 2 +- main/xiaozhi-server/plugins_func/register.py | 1 + 5 files changed, 228 insertions(+), 4 deletions(-) create mode 100644 main/xiaozhi-server/core/mcp/MCPClient.py create mode 100644 main/xiaozhi-server/core/mcp/manager.py diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 718398c3..b5e7dcf2 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -22,6 +22,7 @@ from plugins_func.register import Action 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__ @@ -98,6 +99,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: @@ -151,6 +154,8 @@ class ConnectionHandler: # 异步初始化 await self.loop.run_in_executor(None, self._initialize_components) + # 初始化MCP服务 + await self.mcp_manager.initialize_servers() # tts 消化线程 tts_priority = threading.Thread(target=self._tts_priority_thread, daemon=True) @@ -317,7 +322,9 @@ class ConnectionHandler: # Define intent functions functions = self.func_handler.get_functions() - + mcpfuncs = self.mcp_manager.get_all_tools() # MCP工具 + functions.extend(mcpfuncs) + print("functions with mcp", functions) response_message = [] processed_chars = 0 # 跟踪已处理的字符位置 @@ -429,8 +436,24 @@ 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) + #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): + try: + tool_result = asyncio.create_task(self.mcp_manager.execute_tool( + function_name, + function_arguments + )) + self._handle_mcp_tool_result(tool_result, function_call_data, text_index+1) + except Exception as e: + self.logger.bind(tag=TAG).error(f"MCP工具调用错误: {e}") + response_message.append(f"MCP工具调用失败: {str(e)}") + 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) @@ -565,6 +588,8 @@ class ConnectionHandler: async def close(self): """资源清理方法""" + # 清理MCP资源 + await self.mcp_manager.cleanup_all() # 清理其他资源 self.stop_event.set() diff --git a/main/xiaozhi-server/core/mcp/MCPClient.py b/main/xiaozhi-server/core/mcp/MCPClient.py new file mode 100644 index 00000000..903ce60b --- /dev/null +++ b/main/xiaozhi-server/core/mcp/MCPClient.py @@ -0,0 +1,96 @@ +import asyncio +from typing import Optional +from contextlib import AsyncExitStack +import os +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): + main_command = self.config["command"] + args = self.config.get("args", []) + + server_script_path = main_command + if args: + # 如果第一个参数是路径，使用其目录作为工作目录 + possible_path = args[0] + if os.path.exists(possible_path): + server_script_path = possible_path + await self.connect_to_server(server_script_path) + + + async def connect_to_server(self, server_script_path: str): + """Connect to an MCP server + + Args: + server_script_path: Path to the server script (.py or .js) + """ + is_python = server_script_path.endswith('.py') + is_js = server_script_path.endswith('.js') + if not (is_python or is_js): + raise ValueError("Server script must be a .py or .js file") + + env = self.config.get("env", {}) + command = "python" if is_python else "node" + server_params = StdioServerParameters( + command=command, + args=[server_script_path], + env=env + ) + + stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params)) + self.stdio, self.write = stdio_transport + self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write)) + + 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 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): + response = await self.session.call_tool(tool_name, tool_args) + return response + + async def cleanup(self): + """Clean up resources""" + await self.exit_stack.aclose() + +async def main(): + if len(sys.argv) < 2: + print("Usage: python client.py ") + sys.exit(1) + + client = MCPClient() + try: + await client.connect_to_server(sys.argv[1]) + await client.chat_loop() + finally: + await client.cleanup() + +if __name__ == "__main__": + import sys + asyncio.run(main()) \ 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..f9b4a742 --- /dev/null +++ b/main/xiaozhi-server/core/mcp/manager.py @@ -0,0 +1,102 @@ +"""MCP服务管理器""" +import 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.config_path = get_project_dir() + 'data/.mcp_server_settings.json' + self.client: Dict[str, MCPClient] = {} + self.logger = setup_logging() + self.tools = [] + + def load_config(self) -> Dict[str, Any]: + """加载MCP服务配置 + Returns: + Dict[str, Any]: 服务配置字典 + """ + 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}") + + 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: 工具未找到时抛出 + """ + for client in self.client.values(): + if any(tool_name == tool["name"] for tool in client.tools): + 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/plugins_func/functions/get_time.py b/main/xiaozhi-server/plugins_func/functions/get_time.py index 1d141ed2..9a9aebfe 100644 --- a/main/xiaozhi-server/plugins_func/functions/get_time.py +++ b/main/xiaozhi-server/plugins_func/functions/get_time.py @@ -6,7 +6,7 @@ get_time_function_desc = { "function": { "name": "get_time", "description": "获取当前时间、日期、星期几", - "parameters": {} + 'parameters': {'type': 'object', 'properties': {}, 'required': []} } } 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