""" 统一工具管理器 - 协调 Liugin、Skill 和 MCP 三套协议 自动判断并路由到正确的处理器 集成插件内存管理: TTL 自动淘汰 + 懒加载 """ import os import json import logging import threading from typing import Dict, List, Any, Optional, Union, Callable from colorama import Fore, Style logger = logging.getLogger(__name__) class UnifiedToolManager: """ 统一工具管理器 协调三种协议: 1. Liugin 协议 (plugins/) - 小狸自有协议 2. Skill 协议 (skills/) - Agent Skills 开源协议 3. MCP 协议 - Model Context Protocol (Anthropic/OpenAI 标准) 路由策略: - 优先使用 Skill 协议（如果存在同名技能） - 回退到 Liugin 协议 - 支持强制指定协议 - MCP 格式调用统一入口 内存管理: - 空闲插件自动淘汰（TTL 机制） - 调用时自动懒加载回来 - 核心插件可钉住不淘汰 """ # 内存管理默认配置 DEFAULT_PLUGIN_TTL = 300 # 5分钟 DEFAULT_MAX_IDLE_PLUGINS = 10 # 最多保留10个空闲插件 DEFAULT_CHECK_INTERVAL = 60 # 60秒检查一次 DEFAULT_PINNED_PLUGINS = ['tool_search'] # 默认钉住的插件 def __init__(self, cli_instance=None, plugin_ttl: int = None, max_idle_plugins: int = None, check_interval: int = None, pinned_plugins: list = None): self.cli = cli_instance self._lock = threading.RLock() # 两个独立的加载器 self._liugin_manager = None self._skill_loader = None # 统一的工具注册表 self._tools: Dict[str, Dict[str, Any]] = {} # 协议映射: tool_name -> protocol_type self._protocol_map: Dict[str, str] = {} # MCP 工具定义缓存 self._mcp_tools_cache: Optional[List[Dict]] = None # ── 原始文件路径缓存（用于懒加载） ── self._plugin_paths: Dict[str, str] = {} # name -> file_path self._skill_dirs: Dict[str, str] = {} # name -> skill_dir # ── 插件内存管理器 ── from xcli_core.plugin_memory_manager import PluginMemoryManager self._memory_mgr = PluginMemoryManager( plugin_ttl=plugin_ttl or self.DEFAULT_PLUGIN_TTL, max_idle_plugins=max_idle_plugins or self.DEFAULT_MAX_IDLE_PLUGINS, check_interval=check_interval or self.DEFAULT_CHECK_INTERVAL, pinned_plugins=pinned_plugins or self.DEFAULT_PINNED_PLUGINS, ) self._memory_mgr.set_callbacks( load_fn=self._lazy_load_plugin, unload_fn=self._unload_plugin, ) def initialize(self, plugins_dir: str = "plugins", skills_dir: str = "skills") -> None: """ 初始化并加载所有工具 Args: plugins_dir: Liugin 目录 skills_dir: Skill 目录 """ # 加载 Liugins self._plugins_dir = plugins_dir self._skills_dir = skills_dir self._load_plugins(plugins_dir) # 加载 Skills self._load_skills(skills_dir) # 构建缓存 self._build_mcp_cache() # 启动内存管理器 self._memory_mgr.start() print(f"{Fore.GREEN}[统一工具管理器] 加载完成: {len(self._tools)} 个工具{Style.RESET_ALL}") print(f" - Liugin 协议: {sum(1 for p in self._protocol_map.values() if p == 'liugin')}") print(f" - Skill 协议: {sum(1 for p in self._protocol_map.values() if p == 'skill')}") # 打印内存管理状态 status = self._memory_mgr.get_status() print(f" - 内存管理: TTL={status['ttl']}s, 最大空闲={status['max_idle']}, " f"钉住={status['pinned']}") def _load_plugins(self, plugins_dir: str) -> None: """加载 Liugin 协议工具（并注册到内存管理器）""" import importlib.util if not os.path.exists(plugins_dir): return # 获取启用的插件列表 enabled_plugins = self._get_enabled_plugins(plugins_dir) for filename in os.listdir(plugins_dir): if not filename.endswith('.py') or filename == '__init__.py': continue plugin_name = filename[:-3] # 检查是否启用 if enabled_plugins is not None and plugin_name.lower() not in enabled_plugins: continue plugin_path = os.path.join(plugins_dir, filename) try: spec = importlib.util.spec_from_file_location(plugin_name, plugin_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) # 兼容 Plugin 和 Liugin 两种类名 plugin_class = None if hasattr(module, 'Liugin'): plugin_class = module.Liugin elif hasattr(module, 'Plugin'): plugin_class = module.Plugin if plugin_class: plugin_instance = plugin_class() # 设置 CLI 引用 if hasattr(plugin_instance, 'set_cli') and self.cli: plugin_instance.set_cli(self.cli) tool_info = plugin_instance.get_tool_info() tool_name = tool_info.get('name', plugin_name) # 获取 MCP 工具定义（如果插件提供） mcp_definition = None if hasattr(plugin_instance, 'get_mcp_definition'): mcp_definition = plugin_instance.get_mcp_definition() # 注册到统一工具表 self._tools[tool_name] = { 'name': tool_name, 'description': tool_info.get('description', ''), 'keywords': tool_info.get('keywords', []), 'usage': tool_info.get('usage', ''), 'handler': plugin_instance.handle, 'protocol': 'liugin', 'instance': plugin_instance, 'mcp_definition': mcp_definition } self._protocol_map[tool_name] = 'liugin' # 记录路径（用于懒加载） self._plugin_paths[tool_name] = plugin_path # 注册到内存管理器 self._memory_mgr.register( name=tool_name, path=plugin_path, protocol='liugin', info_cache={ 'name': tool_name, 'description': tool_info.get('description', ''), 'keywords': tool_info.get('keywords', []), 'usage': tool_info.get('usage', ''), } ) except Exception as e: print(f"{Fore.RED}加载 Liugin '{plugin_name}' 失败: {e}{Style.RESET_ALL}") def _load_skills(self, skills_dir: str) -> None: """加载 Skill 协议工具 (支持 Markdown 和 Python 两种格式)""" if not os.path.exists(skills_dir): print(f"{Fore.YELLOW}Skills 目录不存在: {skills_dir}{Style.RESET_ALL}") return try: from skills.base import SkillLoader, MarkdownSkill, Skill self._skill_loader = SkillLoader(skills_dir) # 先发现所有技能 discovered = self._skill_loader.discover_skills(skills_dir) print(f"{Fore.CYAN}发现 {len(discovered)} 个技能: {[s['name'] for s in discovered]}{Style.RESET_ALL}") # 加载所有技能 self._skill_loader.load_all_skills(skills_dir) loaded_count = len(self._skill_loader.skills) if loaded_count == 0: print(f"{Fore.YELLOW}未加载任何 Skill 技能{Style.RESET_ALL}") return for skill_name, skill in self._skill_loader.skills.items(): # 设置 CLI 引用 (仅 Python 类技能) if hasattr(skill, 'set_cli') and self.cli: skill.set_cli(self.cli) # 获取工具定义 tool_def = skill.get_tool_definition() func_def = tool_def.get('function', {}) # 检查是否已有同名 Plugin，Skill 优先覆盖 if skill_name in self._tools: old_protocol = self._protocol_map[skill_name] print(f"{Fore.YELLOW}工具 '{skill_name}' 已存在({old_protocol})，使用 Skill 协议覆盖{Style.RESET_ALL}") # 判断技能格式 is_markdown = isinstance(skill, MarkdownSkill) skill_format = "markdown" if is_markdown else "python" # 注册到统一工具表 self._tools[skill_name] = { 'name': skill_name, 'description': func_def.get('description', ''), 'keywords': getattr(skill, 'tags', []), 'usage': getattr(skill, 'instructions', '')[:500] if is_markdown else getattr(skill, 'usage', ''), 'handler': self._create_skill_handler(skill), 'protocol': 'skill', 'format': skill_format, 'instance': skill, 'tool_definition': tool_def, 'anthropic_definition': skill.get_anthropic_tool_definition(), 'mcp_definition': tool_def # Skill 的 OpenAI 格式就是 MCP 格式 } self._protocol_map[skill_name] = 'skill' # 记录路径（用于懒加载） skill_path = os.path.join(skills_dir, skill_name) self._skill_dirs[skill_name] = skill_path # 注册到内存管理器 self._memory_mgr.register( name=skill_name, path=skill_path, protocol='skill', info_cache={ 'name': skill_name, 'description': func_def.get('description', ''), 'keywords': getattr(skill, 'tags', []), 'format': skill_format, } ) print(f"{Fore.GREEN}已加载 Skill ({skill_format}): {skill_name}{Style.RESET_ALL}") except ImportError as e: print(f"{Fore.YELLOW}Skill 协议模块未找到: {e}{Style.RESET_ALL}") except Exception as e: print(f"{Fore.RED}加载 Skills 失败: {e}{Style.RESET_ALL}") def _create_skill_handler(self, skill) -> callable: """ 为 Skill 创建兼容 Plugin 的 handler 支持 Markdown 和 Python 两种技能格式 """ from skills.base import MarkdownSkill, Skill def handler(args: str) -> str: import json kwargs = {} # 解析参数 args = args.strip() if args.startswith('{'): try: kwargs = json.loads(args) except json.JSONDecodeError: kwargs = {'operation': args} elif '=' in args: # key=value 格式: "directory=. file_pattern=*.py" for part in args.split(): if '=' in part: k, v = part.split('=', 1) kwargs[k.strip()] = v.strip() else: kwargs.setdefault('operation', part) else: kwargs = {'operation': args} # Markdown 技能: 返回指令供 AI 执行 if isinstance(skill, MarkdownSkill): return f"[Skill: {skill.name}]\n{skill.instructions}" # Python 技能: 直接执行 if hasattr(skill, 'run'): result = skill.run(**kwargs) if result.success: return result.message or str(result.data) else: return f"错误: {result.error}" return f"未知的技能类型: {type(skill)}" return handler # ==================== 懒加载/卸载 ==================== def _lazy_load_plugin(self, name: str) -> Optional[Dict[str, Any]]: """ 懒加载单个插件（由内存管理器回调） Args: name: 插件名 Returns: tool_dict 如果成功，None 如果失败 """ import importlib.util # ── 尝试 Liugin 插件 ── plugin_path = self._plugin_paths.get(name) if plugin_path and os.path.exists(plugin_path): return self._load_single_plugin(name, plugin_path) # ── 尝试 Skill ── skill_dir = self._skill_dirs.get(name) if skill_dir and os.path.exists(skill_dir): return self._load_single_skill(name, skill_dir) return None def _load_single_plugin(self, name: str, plugin_path: str) -> Optional[Dict[str, Any]]: """加载单个 Liugin 插件""" import importlib.util try: plugin_name = os.path.splitext(os.path.basename(plugin_path))[0] spec = importlib.util.spec_from_file_location(plugin_name, plugin_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) plugin_class = getattr(module, 'Liugin', None) or getattr(module, 'Plugin', None) if not plugin_class: return None plugin_instance = plugin_class() if hasattr(plugin_instance, 'set_cli') and self.cli: plugin_instance.set_cli(self.cli) tool_info = plugin_instance.get_tool_info() tool_name = tool_info.get('name', name) mcp_definition = None if hasattr(plugin_instance, 'get_mcp_definition'): mcp_definition = plugin_instance.get_mcp_definition() tool_dict = { 'name': tool_name, 'description': tool_info.get('description', ''), 'keywords': tool_info.get('keywords', []), 'usage': tool_info.get('usage', ''), 'handler': plugin_instance.handle, 'protocol': 'liugin', 'instance': plugin_instance, 'mcp_definition': mcp_definition } with self._lock: self._tools[tool_name] = tool_dict self._protocol_map[tool_name] = 'liugin' return tool_dict except Exception as e: logger.error(f"懒加载插件 '{name}' 失败: {e}") return None def _load_single_skill(self, name: str, skill_dir: str) -> Optional[Dict[str, Any]]: """加载单个 Skill""" try: from skills.base import SkillLoader, MarkdownSkill, Skill loader = SkillLoader(os.path.dirname(skill_dir)) skill = loader.load_skill(skill_dir) if not skill: return None if hasattr(skill, 'set_cli') and self.cli: skill.set_cli(self.cli) tool_def = skill.get_tool_definition() func_def = tool_def.get('function', {}) is_markdown = isinstance(skill, MarkdownSkill) skill_format = "markdown" if is_markdown else "python" tool_dict = { 'name': name, 'description': func_def.get('description', ''), 'keywords': getattr(skill, 'tags', []), 'usage': getattr(skill, 'instructions', '')[:500] if is_markdown else getattr(skill, 'usage', ''), 'handler': self._create_skill_handler(skill), 'protocol': 'skill', 'format': skill_format, 'instance': skill, 'tool_definition': tool_def, 'anthropic_definition': skill.get_anthropic_tool_definition(), 'mcp_definition': tool_def } with self._lock: self._tools[name] = tool_dict self._protocol_map[name] = 'skill' return tool_dict except Exception as e: logger.error(f"懒加载 Skill '{name}' 失败: {e}") return None def _unload_plugin(self, name: str) -> None: """ 卸载插件实例（由内存管理器回调） 从工具表中移除，释放实例引用，但保留路径缓存以便重新加载 """ with self._lock: if name in self._tools: tool = self._tools[name] # 清理实例引用 instance = tool.get('instance') if instance and hasattr(instance, 'cleanup'): try: instance.cleanup() except Exception: pass del self._tools[name] # 不删除 _protocol_map 和路径缓存，懒加载时需要 # 重建 MCP 缓存 self._build_mcp_cache() def _ensure_loaded(self, name: str) -> Optional[Dict[str, Any]]: """ 确保插件已加载（调用入口统一使用） 如果插件已加载，更新使用时间戳并返回 tool_dict 如果插件已卸载，自动懒加载后返回 tool_dict 如果插件不存在，返回 None """ # 先检查是否在工具表中 with self._lock: tool = self._tools.get(name) if tool: self._memory_mgr.mark_used(name) return tool # 不在工具表中，可能是已卸载，尝试懒加载 result = self._memory_mgr.ensure_loaded(name) if result: self._memory_mgr.mark_used(name) return result return None def _get_enabled_plugins(self, plugins_dir: str) -> Optional[List[str]]: """获取启用的插件列表""" # 检查环境变量 enabled_env = os.environ.get("XIAOLI_ENABLED_PLUGINS") if enabled_env is not None: if enabled_env == "": return [] return [p.lower() for p in enabled_env.split(",")] # 检查配置文件 config_file = os.path.join(os.path.dirname(plugins_dir), "plugins_config.py") if os.path.exists(config_file): try: with open(config_file, 'r', encoding='utf-8') as f: content = f.read() if "# 不启用任何插件" in content: return [] enabled = [] for line in content.split('\n'): if line.startswith('ENABLE_') and '= True' in line: name = line.split('=')[0].replace('ENABLE_', '').strip().lower() enabled.append(name) return enabled except Exception: pass return None # 加载全部 def _build_mcp_cache(self) -> None: """构建 MCP 工具定义缓存（线程安全）""" with self._lock: self._mcp_tools_cache = [] for tool in self._tools.values(): mcp_def = tool.get('mcp_definition') if mcp_def: # 确保符合 MCP 标准 normalized = self._normalize_mcp_definition(mcp_def) self._mcp_tools_cache.append(normalized) else: # 为没有 MCP 定义的 Plugin 生成默认定义 self._mcp_tools_cache.append(self._generate_mcp_definition(tool)) def _normalize_mcp_definition(self, mcp_def: Dict[str, Any]) -> Dict[str, Any]: """ 标准化 MCP 工具定义 支持多种输入格式并转换为标准 MCP 格式 """ # 如果已经是 MCP 格式（有 name 或 inputSchema） if 'name' in mcp_def and ('inputSchema' in mcp_def or 'description' in mcp_def): return mcp_def # 如果是 OpenAI Function Calling 格式，转换 if mcp_def.get('type') == 'function' and 'function' in mcp_def: func = mcp_def['function'] return { "name": func.get('name', ''), "description": func.get('description', ''), "inputSchema": func.get('parameters', { "type": "object", "properties": {} }) } # 其他情况，直接返回 return mcp_def def _generate_mcp_definition(self, tool: Dict[str, Any]) -> Dict[str, Any]: """ 为 Plugin 生成符合 MCP 标准的工具定义 MCP 官方格式 (2025-03-26): { "name": "tool_name", "title": "Human Readable Name", // optional "description": "Human-readable description", "inputSchema": { "type": "object", "properties": {...}, "required": [...] }, "outputSchema": {...}, // optional "annotations": {...} // optional } """ return { "name": tool['name'], "description": tool.get('description', ''), "inputSchema": { "type": "object", "properties": { "args": { "type": "string", "description": f"命令参数。用法:\\n{tool.get('usage', '')[:1000]}" } }, "required": ["args"] } } # ==================== 公共 API ==================== @property def tools(self) -> List[Dict[str, Any]]: """获取所有工具列表（兼容 LiuginManager 接口）""" return list(self._tools.values()) def get_tool(self, name: str) -> Optional[Dict[str, Any]]: """获取指定工具（支持懒加载）""" return self._ensure_loaded(name) def get_protocol(self, name: str) -> Optional[str]: """获取工具使用的协议""" return self._protocol_map.get(name) def execute(self, tool_name: str, args: str = "", **kwargs) -> Dict[str, Any]: """ 执行工具调用（支持懒加载） Args: tool_name: 工具名称 args: 参数字符串（Plugin 格式） **kwargs: 关键字参数（Skill 格式或 MCP arguments） Returns: {"result": "..."} 格式的结果 """ tool = self._ensure_loaded(tool_name) if not tool: return {"result": f"未找到工具: {tool_name}"} try: handler = tool['handler'] protocol = tool['protocol'] instance = tool.get('instance') if protocol == 'skill' and kwargs: # Skill 协议，使用 kwargs skill = tool['instance'] result = skill.run(**kwargs) return {"result": result.message or str(result.data) if result.success else f"错误: {result.error}"} elif protocol == 'liugin' and kwargs and not args: # Liugin 协议但有 MCP 结构化参数，需要转换 # 检查插件是否有 convert_mcp_args 方法 if instance and hasattr(instance, 'convert_mcp_args'): args = instance.convert_mcp_args(kwargs) else: args = self._convert_mcp_args_to_plugin_args(tool_name, kwargs, instance) # 执行 handler result = handler(args) return {"result": result if result is not None else "无结果"} except Exception as e: return {"result": f"工具执行错误: {e}"} # ==================== MCP 协议支持 ==================== def get_mcp_tools(self) -> List[Dict[str, Any]]: """ 获取 MCP 格式的工具定义列表 MCP (Model Context Protocol) 使用 OpenAI Function Calling 格式 作为标准工具定义格式 Returns: List of MCP tool definitions """ with self._lock: if self._mcp_tools_cache is None: self._build_mcp_cache() return self._mcp_tools_cache.copy() # 返回副本，避免外部修改 def mcp_call(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """ MCP 格式的工具调用入口（支持懒加载） 这是 MCP 协议的主要调用接口，接收结构化参数并返回结果 Args: tool_name: 工具名称 arguments: 结构化参数 (JSON对象) Returns: MCP 标准响应格式: { "content": [ { "type": "text", "text": "结果内容" } ], "isError": false } """ tool = self._ensure_loaded(tool_name) if not tool: return self._mcp_error(f"未找到工具: {tool_name}") try: protocol = tool['protocol'] instance = tool.get('instance') if protocol == 'liugin': # Liugin 协议: 将 arguments 转换为 args 字符串 args = self._convert_mcp_args_to_plugin_args(tool_name, arguments, instance) result = tool['handler'](args) return self._mcp_success(result if result else "执行完成") elif protocol == 'skill': # Skill 协议: 直接传递 arguments 作为 kwargs from skills.base import MarkdownSkill if isinstance(instance, MarkdownSkill): return self._mcp_success(f"[Skill: {instance.name}]\n{instance.instructions}") if hasattr(instance, 'run'): result = instance.run(**arguments) if result.success: return self._mcp_success(result.message or str(result.data)) else: return self._mcp_error(f"执行失败: {result.error}") return self._mcp_error(f"未知协议类型: {protocol}") except Exception as e: return self._mcp_error(f"工具执行错误: {str(e)}") def _convert_mcp_args_to_plugin_args(self, tool_name: str, arguments: Dict[str, Any], instance: Any) -> str: """ 将 MCP 结构化参数转换为 Plugin 的 args 字符串 Args: tool_name: 工具名称 arguments: MCP 参数对象 instance: 插件实例（可能有自定义转换方法） Returns: args 字符串 """ # 检查插件是否有自定义的 MCP 参数转换方法 if instance and hasattr(instance, 'convert_mcp_args'): return instance.convert_mcp_args(arguments) # 默认转换策略 if 'args' in arguments: # 如果有 args 字段，直接使用 return arguments['args'] # 将 arguments 转换为命令行格式 parts = [] # 第一个参数通常是操作/命令 if 'operation' in arguments: parts.append(arguments['operation']) elif 'action' in arguments: parts.append(arguments['action']) # 添加其他参数 for key, value in arguments.items(): if key in ('operation', 'action'): continue if isinstance(value, bool): if value: parts.append(f"-{key}") elif isinstance(value, (list, tuple)): for v in value: parts.append(f"--{key}") parts.append(self._quote_if_needed(str(v))) else: parts.append(f"--{key}") parts.append(self._quote_if_needed(str(value))) return ' '.join(parts) def _quote_if_needed(self, s: str) -> str: """如果字符串包含空格，添加引号""" if ' ' in s or '"' in s: return f'"{s}"' return s def _mcp_success(self, text: str) -> Dict[str, Any]: """生成 MCP 成功响应""" return { "content": [ { "type": "text", "text": text } ], "isError": False } def _mcp_error(self, error_msg: str) -> Dict[str, Any]: """生成 MCP 错误响应""" return { "content": [ { "type": "text", "text": error_msg } ], "isError": True } # ==================== 兼容性 API ==================== def get_openai_tools(self) -> List[Dict[str, Any]]: """ 获取 OpenAI Function Calling 格式的工具定义 OpenAI 格式: { "type": "function", "function": { "name": "...", "description": "...", "parameters": {...} } } """ tools = [] for mcp_def in self.get_mcp_tools(): tools.append({ "type": "function", "function": { "name": mcp_def.get("name", ""), "description": mcp_def.get("description", ""), "parameters": mcp_def.get("inputSchema", { "type": "object", "properties": {} }) } }) return tools def get_anthropic_tools(self) -> List[Dict[str, Any]]: """ 获取 Anthropic Tool Use 格式的工具定义 Anthropic 格式: { "name": "tool_name", "description": "...", "input_schema": {...} } """ tools = [] for mcp_def in self.get_mcp_tools(): tools.append({ "name": mcp_def.get("name", ""), "description": mcp_def.get("description", ""), "input_schema": mcp_def.get("inputSchema", { "type": "object", "properties": {} }) }) return tools def search_tools(self, query: str, limit: int = 5) -> List[Dict[str, Any]]: """ 搜索工具（包括已卸载的插件，使用缓存信息） Args: query: 搜索关键词 limit: 返回数量限制 Returns: 匹配的工具列表 """ query_lower = query.lower() # 标准化：移除所有分隔符（连字符、下划线、空格） query_normalized = query_lower.replace('-', '').replace('_', '').replace(' ', '') matches = [] # 搜索已加载的工具 for tool in self._tools.values(): score = self._score_tool_match(tool, query_lower, query_normalized) if score > 0: matches.append({ 'name': tool['name'], 'description': tool['description'], 'usage': tool.get('usage', ''), 'keywords': [kw.lower() for kw in tool.get('keywords', [])], 'protocol': tool['protocol'], 'instance': tool.get('instance'), 'score': score }) # 搜索已卸载但注册的插件（使用缓存信息） for name in self._memory_mgr.get_unloaded_names(): # 跳过已加载的（上面已搜索） if name in self._tools: continue info = self._memory_mgr.get_info_cache(name) if info: score = self._score_tool_match(info, query_lower, query_normalized) if score > 0: matches.append({ 'name': info.get('name', name), 'description': info.get('description', ''), 'usage': info.get('usage', ''), 'keywords': info.get('keywords', []), 'protocol': info.get('protocol', 'liugin'), 'instance': None, 'score': score, '_unloaded': True, # 标记为已卸载 }) # 按分数排序 matches.sort(key=lambda x: x['score'], reverse=True) return matches[:limit] def _score_tool_match(self, tool: Dict, query_lower: str, query_normalized: str) -> int: """计算工具匹配分数""" score = 0 name = tool.get('name', '').lower() name_normalized = name.replace('-', '').replace('_', '') desc = tool.get('description', '').lower() keywords = [kw.lower() for kw in tool.get('keywords', [])] # 名称完全匹配（忽略分隔符差异） if query_lower == name or query_normalized == name_normalized: score += 100 # 名称包含（标准化后匹配） elif query_normalized in name_normalized or query_lower in name: score += 50 # 描述包含 if query_lower in desc: score += 30 # 关键词匹配 for kw in keywords: kw_normalized = kw.replace('-', '').replace('_', '') if query_lower in kw or query_normalized in kw_normalized: score += 20 break return score def register_tool(self, name: str, handler: callable, description: str = "", protocol: str = "custom", mcp_definition: Dict = None, **kwargs) -> None: """ 动态注册工具 Args: name: 工具名称 handler: 处理函数 description: 描述 protocol: 协议类型 mcp_definition: MCP 工具定义（可选） """ with self._lock: self._tools[name] = { 'name': name, 'description': description, 'handler': handler, 'protocol': protocol, 'mcp_definition': mcp_definition, **kwargs } self._protocol_map[name] = protocol # 更新缓存 if self._mcp_tools_cache is not None: if mcp_definition: self._mcp_tools_cache.append(mcp_definition) else: self._mcp_tools_cache.append(self._generate_mcp_definition(self._tools[name])) def unregister_tool(self, name: str) -> bool: """注销工具""" with self._lock: if name in self._tools: del self._tools[name] del self._protocol_map[name] # 同时从内存管理器注销 if name in self._plugin_paths: del self._plugin_paths[name] if name in self._skill_dirs: del self._skill_dirs[name] # 重建缓存 self._build_mcp_cache() return True return False def get_all_tools_info(self) -> List[Dict[str, Any]]: """ 获取所有工具的简要信息（包括已卸载的） 用于系统提示词，让 AI 知道有哪些工具可用 已卸载的工具只有基本信息，调用时会自动加载 """ result = [] # 已加载的 for tool in self._tools.values(): result.append({ 'name': tool['name'], 'description': tool.get('description', ''), 'keywords': tool.get('keywords', []), 'protocol': tool.get('protocol', ''), 'loaded': True, }) # 已卸载的（从内存管理器缓存读取） for name in self._memory_mgr.get_unloaded_names(): if name in self._tools: continue info = self._memory_mgr.get_info_cache(name) if info: result.append({ 'name': info.get('name', name), 'description': info.get('description', ''), 'keywords': info.get('keywords', []), 'protocol': info.get('protocol', 'liugin'), 'loaded': False, }) return result def get_tool_by_name(self, name: str) -> Optional[Dict[str, Any]]: """根据名称获取工具（兼容 LiuginManager 接口，支持懒加载）""" return self._ensure_loaded(name) @property def memory_manager(self): """获取内存管理器实例""" return self._memory_mgr def get_skills_system_prompt(self) -> str: """ 获取所有技能的系统提示词 用于注入到 AI 的 system prompt 中， 让 AI 了解如何使用这些技能 """ skill_prompts = [] for name, tool in self._tools.items(): if tool['protocol'] == 'skill': instance = tool.get('instance') if instance: # 获取技能指令 instructions = getattr(instance, 'instructions', '') scripts_dir = getattr(instance, 'scripts_dir', '') prompt_parts = [f"## 技能: {name}"] prompt_parts.append(f"描述: {tool['description']}") if instructions: # 截取关键部分，避免过长 prompt_parts.append(f"\n{instructions}") if scripts_dir: prompt_parts.append(f"\n脚本目录: {scripts_dir}") skill_prompts.append('\n'.join(prompt_parts)) if skill_prompts: header = "# 已加载的 Agent Skills\n以下技能可用于执行特定任务，调用方式与工具相同：\n" return header + '\n\n---\n\n'.join(skill_prompts) return "" def get_tools_prompt(self) -> str: """ 获取工具使用提示词（兼容旧接口） 包含 Plugin 和 Skill 两种协议的工具说明 """ prompts = [] # Liugin 协议工具 plugin_tools = [t for t in self._tools.values() if t['protocol'] == 'liugin'] if plugin_tools: prompts.append("## Liugin 工具") for tool in plugin_tools: usage = tool.get('usage', '') if usage: prompts.append(f"### {tool['name']}\n{usage}\n") # Skill 协议工具（使用系统提示词） skills_prompt = self.get_skills_system_prompt() if skills_prompt: prompts.append(skills_prompt) return '\n\n'.join(prompts) # ==================== MCP Server 模式支持 ==================== def get_mcp_server_info(self) -> Dict[str, Any]: """ 获取 MCP Server 信息 用于 MCP 客户端连接时的初始化响应 """ return { "name": "xiaoli-tool-server", "version": "1.0.0", "protocolVersion": "2024-11-05", "capabilities": { "tools": { "listChanged": True } } } def handle_mcp_request(self, request: Dict[str, Any]) -> Dict[str, Any]: """ 处理 MCP 协议请求 统一处理 MCP 客户端的各种请求 Args: request: MCP 请求对象 { "jsonrpc": "2.0", "id": 1, "method": "tools/list" | "tools/call" | ..., "params": {...} } Returns: MCP 响应对象 { "jsonrpc": "2.0", "id": 1, "result": {...} } """ method = request.get('method', '') params = request.get('params', {}) req_id = request.get('id') result = None error = None try: if method == 'tools/list': result = {"tools": self.get_mcp_tools()} elif method == 'tools/call': tool_name = params.get('name') arguments = params.get('arguments', {}) if not tool_name: error = {"code": -32602, "message": "Missing tool name"} else: result = self.mcp_call(tool_name, arguments) elif method == 'initialize': result = self.get_mcp_server_info() elif method == 'ping': result = {} else: error = {"code": -32601, "message": f"Method not found: {method}"} except Exception as e: error = {"code": -32603, "message": str(e)} response = {"jsonrpc": "2.0", "id": req_id} if error: response["error"] = error else: response["result"] = result return response