""" Tool 目录聚合服务 从插件元数据、MCP registry/docs 和插件状态聚合 ToolCatalogItem。 """ import logging from typing import List, Optional, Dict, Any from pathlib import Path from app.plugins.models import ( ToolCatalogItem, PluginStatus, ) logger = logging.getLogger(__name__) class ToolCatalogService: """Tool 目录服务 聚合插件元数据、MCP registry 和 Pydantic schema， 为前端提供中文可读工具目录。 """ def __init__(self): self._tools: Dict[str, ToolCatalogItem] = {} def scan_tools(self, plugins_dir: Path, plugin_manager): """扫描所有插件的工具 Args: plugins_dir: 插件目录 plugin_manager: 插件管理器实例 """ self._tools.clear() for plugin_name, plugin_info in plugin_manager._plugins.items(): try: plugin_tools = self._extract_plugin_tools( plugin_name, plugin_info, plugins_dir / plugin_name ) self._tools.update(plugin_tools) except Exception as e: logger.error(f"提取插件 {plugin_name} 工具失败: {e}") def sync_mcp_tools(self, registry, plugin_manager): """从 MCP 注册表同步工具，补齐未在 plugin.json tools 字段声明的工具。""" for tool_name, tool_info in registry.get_tool_docs().items(): try: # MCP 注册表是运行时事实来源，目录必须以它为准，避免真实可调用工具在目录页为空。 self._tools[tool_name] = self._build_mcp_tool_item( tool_name=tool_name, tool_info=tool_info, plugin_manager=plugin_manager, ) except Exception as e: logger.error(f"同步 MCP 工具 {tool_name} 到 Tool 目录失败: {e}") def _extract_plugin_tools( self, plugin_name: str, plugin_info: dict, plugin_dir: Path ) -> Dict[str, ToolCatalogItem]: """提取单个插件的工具 Args: plugin_name: 插件名 plugin_info: 插件信息 plugin_dir: 插件目录 Returns: {tool_name: ToolCatalogItem} 字典 """ tools = {} metadata_dict = plugin_info.get("metadata_dict", {}) status = plugin_info.get("status", PluginStatus.DISABLED) # 从 plugin.json 的 tools 字段提取 tools_config = metadata_dict.get("tools", []) for tool_config in tools_config: tool_name = tool_config.get("name") if not tool_name: continue # 构建 ToolCatalogItem tool_item = self._build_tool_item( tool_name=tool_name, tool_config=tool_config, plugin_name=plugin_name, plugin_status=status, plugin_dir=plugin_dir, ) tools[tool_name] = tool_item return tools def _build_tool_item( self, tool_name: str, tool_config: dict, plugin_name: str, plugin_status: PluginStatus, plugin_dir: Path, ) -> ToolCatalogItem: """构建 ToolCatalogItem Args: tool_name: 工具名 tool_config: 工具配置（来自 plugin.json） plugin_name: 插件名 plugin_status: 插件状态 plugin_dir: 插件目录 Returns: ToolCatalogItem """ # 检查元数据完整性 incomplete_reasons = [] display_name = tool_config.get("display_name") if not display_name: incomplete_reasons.append("缺少中文名称") purpose = tool_config.get("purpose") if not purpose: incomplete_reasons.append("缺少用途说明") parameters = tool_config.get("parameters") if not parameters: incomplete_reasons.append("缺少参数说明") input_examples = tool_config.get("input_examples") if not input_examples: incomplete_reasons.append("缺少输入示例") output_examples = tool_config.get("output_examples") if not output_examples: incomplete_reasons.append("缺少输出示例") common_errors = tool_config.get("common_errors") if not common_errors: incomplete_reasons.append("缺少常见错误说明") # 确定可用状态 availability = "available" unavailable_reason = None if plugin_status == PluginStatus.DISABLED: availability = "unavailable" unavailable_reason = "插件已禁用" elif plugin_status == PluginStatus.DEPENDENCY_ERROR: availability = "unavailable" unavailable_reason = "插件依赖缺失" elif plugin_status == PluginStatus.ERROR: availability = "unavailable" unavailable_reason = "插件加载错误" elif plugin_status == PluginStatus.INCOMPLETE: availability = "unavailable" unavailable_reason = "插件元数据不完整" elif incomplete_reasons: availability = "incomplete" return ToolCatalogItem( name=tool_name, display_name=display_name or tool_name, source="plugin", plugin_name=plugin_name, purpose=purpose or "", applies_to=tool_config.get("applies_to", []), not_for=tool_config.get("not_for", []), parameters=parameters or {}, input_examples=input_examples or [], output_examples=output_examples or [], common_errors=common_errors or [], risk_level=tool_config.get("risk_level", "low"), availability=availability, unavailable_reason=unavailable_reason, incomplete_reasons=incomplete_reasons if incomplete_reasons else None, ) def _build_mcp_tool_item( self, tool_name: str, tool_info: Dict[str, Any], plugin_manager, ) -> ToolCatalogItem: """将 MCP registry 的工具信息转换为 ToolCatalogItem。 Phase 2.9 合并式：先看 self._tools 是否有 plugin.json 已填的明细（来自 scan_tools）， 用其 parameters / input_examples / common_errors / risk_level 等作为 base； 再用 tool_info 覆盖 availability / unavailable_reason 等运行时字段。 """ plugin_name = tool_info.get("plugin", "") plugin_status = plugin_manager.get_plugin_status(plugin_name) availability = tool_info.get("availability", "available") unavailable_reason = tool_info.get("unavailable_reason") incomplete_reasons = tool_info.get("incomplete_reasons") if plugin_status in { PluginStatus.DISABLED, PluginStatus.DEPENDENCY_ERROR, PluginStatus.ERROR, PluginStatus.INCOMPLETE, }: # 插件生命周期状态优先级高于 MCP 工具自身状态，避免禁用插件的工具仍显示可用。 availability = "unavailable" unavailable_reason = plugin_manager.get_unavailable_reason(plugin_name) # Phase 2.9：从已 scan 的 plugin.json 工具明细合并 base 字段 existing = self._tools.get(tool_name) if existing is not None: # 已有 plugin.json 明细：保留 base，只覆盖运行时字段 return ToolCatalogItem( name=tool_name, display_name=existing.display_name or tool_info.get("title") or tool_info.get("tool") or tool_name, source="mcp", plugin_name=plugin_name, purpose=existing.purpose or tool_info.get("description") or "", applies_to=existing.applies_to or [], not_for=existing.not_for or [], parameters=existing.parameters or {}, input_examples=existing.input_examples or [], output_examples=existing.output_examples or [], common_errors=existing.common_errors or [], risk_level=existing.risk_level or "low", availability=availability, unavailable_reason=unavailable_reason, incomplete_reasons=incomplete_reasons, ) # 无 plugin.json 明细：使用装饰器扩展字段（如果 MCP registry 已携带） # 注意：mcp_tool 装饰器扩展的字段需要从 tool_info 中读取（plugin_registry 已透传） return ToolCatalogItem( name=tool_name, display_name=tool_info.get("title") or tool_info.get("tool") or tool_name, source="mcp", plugin_name=plugin_name, purpose=tool_info.get("description") or "", applies_to=tool_info.get("applies_to") or [], not_for=tool_info.get("not_for") or [], parameters=tool_info.get("parameters") or {}, input_examples=tool_info.get("input_examples") or [], output_examples=tool_info.get("output_examples") or [], common_errors=tool_info.get("common_errors") or [], risk_level=tool_info.get("risk_level") or "low", availability=availability, unavailable_reason=unavailable_reason, incomplete_reasons=incomplete_reasons, ) def get_all_tools( self, plugin_name: Optional[str] = None, source: Optional[str] = None, availability: Optional[str] = None, q: Optional[str] = None, ) -> List[ToolCatalogItem]: """获取工具列表 Args: plugin_name: 按插件名过滤 source: 按来源过滤 availability: 按可用状态过滤 q: 搜索关键词 Returns: ToolCatalogItem 列表 """ result = list(self._tools.values()) # 按插件名过滤 if plugin_name: result = [t for t in result if t.plugin_name == plugin_name] # 按来源过滤 if source: result = [t for t in result if t.source == source] # 按可用状态过滤 if availability: result = [t for t in result if t.availability == availability] # 搜索关键词 if q: q_lower = q.lower() result = [ t for t in result if q_lower in t.name.lower() or q_lower in t.display_name.lower() or q_lower in t.purpose.lower() ] return result def get_tool(self, tool_name: str) -> Optional[ToolCatalogItem]: """获取单个工具详情 Args: tool_name: 工具名 Returns: ToolCatalogItem 或 None """ return self._tools.get(tool_name) # 全局实例 tool_catalog_service = ToolCatalogService()