# 5.4 动态发现与加载

本节介绍工具的动态发现和加载机制，包括延迟加载的决策、多级优先级加载策略、Schema缓存、MCP动态注册和上下文缓存等技术，阐明如何构建灵活高效的工具加载系统。

## 5.4.1 动态加载的必要性

静态工具列表的问题：

* 无法适应运行时变化
* 新工具需要重新部署
* 无法根据上下文选择性加载

## 5.4.2 Claude Code 的 shouldDefer 机制

shouldDefer机制决定了工具是否需要延迟加载。下面的流程图展示了动态工具加载的完整流程：

```mermaid
graph TD
    A["工具请求"] -->|检查上下文| B["shouldDefer?"]
    B -->|否| C["立即加载"]
    B -->|是| D["延迟加载"]

    C -->|导入模块| E["创建实例"]
    E -->|缓存| F["已加载工具"]
    F -->|返回| G["就绪可用"]

    D -->|稍后检查| H["上下文就绪?"]
    H -->|是| C
    H -->|否| I["返回不可用"]

    style B fill:#fff3e0
    style C fill:#c8e6c9
    style D fill:#ffccbc
    style G fill:#e8f5e9
```

图 5-3：shouldDefer延迟加载机制

核心实现如下：

```python
class ToolLoader:
    """延迟加载机制"""

    def __init__(self):
        self.loaded_tools = {}
        self.tool_metadata = {}

    def should_defer_loading(self, tool_name: str,
                            context: Dict) -> bool:
        """判断是否延迟加载工具"""
        # 示例:某些工具只在特定上下文才加载
        if tool_name == "database_query":
            return context.get("user_role") != "admin"
        return False

    async def load_tool_on_demand(self, tool_name: str,
                                 context: Dict) -> Optional[Tool]:
        """按需加载工具"""
        if tool_name in self.loaded_tools:
            return self.loaded_tools[tool_name]

        if self.should_defer_loading(tool_name, context):
            return None

        # 动态导入和加载
        try:
            spec = self.tool_metadata.get(tool_name)
            if not spec:
                return None

            module_path = spec["module"]
            if not module_path.startswith("my_harness_tools."):
                raise ValueError("Tool spec must use a trusted module namespace")

            module = await self._async_import(module_path)
            tool_class = getattr(module, spec["class"])
            tool = tool_class(**spec.get("args", {}))

            self.loaded_tools[tool_name] = tool
            return tool

        except Exception as e:
            print(f"Failed to load tool '{tool_name}': {e}")
            return None

    async def _async_import(self, module_path: str):
        """异步导入模块"""
        import importlib
        return importlib.import_module(module_path)
```

## 5.4.3 OpenClaw 的 6 级技能加载优先级

实现的代码结构：

```python
class SkillPriority(Enum):
    """技能加载优先级"""
    SYSTEM = 1          # 固定系统技能
    SESSION = 2         # 会话上下文技能
    USER = 3           # 用户自定义
    DOMAIN = 4         # 领域专业技能
    BUILTIN = 5        # 通用工具
    DISCOVERY = 6      # 动态发现

class SkillLoader:
    """6 级优先级技能加载器"""

    def __init__(self, storage):
        self.storage = storage
        self.loaded_skills = {}

    async def load_skills_for_session(self, session_id: str):
        """为会话加载所有技能"""
        skills = []

        # 1. 系统技能
        skills.extend(await self._load_system_skills())

        # 2. 会话上下文技能
        skills.extend(await self._load_session_skills(session_id))

        # 3. 用户自定义技能
        skills.extend(await self._load_user_skills(session_id))

        # 4. 领域技能
        domain = await self._detect_domain(session_id)
        skills.extend(await self._load_domain_skills(domain))

        # 5. 内置工具
        skills.extend(await self._load_builtin_tools())

        # 6. 动态发现技能
        skills.extend(await self._discover_skills_dynamically(session_id))

        return skills

    async def _load_system_skills(self):
        """加载系统技能(始终可用)"""
        return await self.storage.get_system_skills()

    async def _load_session_skills(self, session_id: str):
        """加载会话特定的技能"""
        return await self.storage.get_session_skills(session_id)

    async def _load_user_skills(self, session_id: str):
        """加载用户定义的技能"""
        user_id = await self.storage.get_user_for_session(session_id)
        return await self.storage.get_user_skills(user_id)

    async def _load_domain_skills(self, domain: str):
        """加载特定领域的技能"""
        if not domain:
            return []
        return await self.storage.get_domain_skills(domain)

    async def _load_builtin_tools(self):
        """加载内置工具"""
        return [
            BashTool(),
            FileReadTool(),
            FileWriteTool(),
            # ... 其他内置工具
        ]

    async def _discover_skills_dynamically(self, session_id: str):
        """动态发现技能"""
        # 扫描技能目录或 MCP
        discovered = await self.storage.discover_available_skills()
        return discovered

    async def _detect_domain(self, session_id: str) -> str:
        """根据会话内容检测领域"""
        recent_messages = await self.storage.get_recent_messages(session_id, limit=5)
        # 使用 NLP 检测领域
        text = " ".join(m.get_text() for m in recent_messages)
        return self._classify_domain(text)

    def _classify_domain(self, text: str) -> str:
        """简单的领域分类"""
        keywords = {
            "data": ["sql", "database", "query", "table"],
            "web": ["html", "css", "javascript", "react"],
            "system": ["bash", "shell", "linux", "windows"],
            "code": ["python", "java", "javascript"]
        }

        text_lower = text.lower()
        for domain, kws in keywords.items():
            if any(kw in text_lower for kw in kws):
                return domain
        return "general"
```

## 5.4.4 JSON Schema 缓存

具体的缓存实现：

```python
class SchemaCache:
    """工具 Schema 缓存"""

    def __init__(self, max_age: int = 3600):
        self.cache = {}
        self.max_age = max_age
        self.timestamps = {}

    def get_schema(self, tool_name: str) -> Optional[Dict]:
        """获取缓存的 Schema"""
        if tool_name in self.cache:
            age = time.time() - self.timestamps[tool_name]
            if age < self.max_age:
                return self.cache[tool_name]
            else:
                del self.cache[tool_name]
                del self.timestamps[tool_name]
        return None

    def set_schema(self, tool_name: str, schema: Dict):
        """缓存 Schema"""
        self.cache[tool_name] = schema
        self.timestamps[tool_name] = time.time()

    def build_all_schemas(self, tools: List[Tool]) -> Dict[str, Dict]:
        """批量构建 Schema 缓存"""
        schemas = {}
        for tool in tools:
            schema = {
                "name": tool.name(),
                "description": tool.description(),
                "input_schema": tool.input_schema()
            }
            self.set_schema(tool.name(), schema)
            schemas[tool.name()] = schema
        return schemas
```

## 5.4.5 MCP 动态注册

MCP(Model Context Protocol)支持动态工具注册：

> ⚠️ **注意**：以下代码为概念示意，非真实 MCP 调用方式。真实 MCP 客户端使用 JSON-RPC 2.0 over stdio 或 Streamable HTTP，不使用 REST 风格的 `/tools/list` 和 `/tools/call` 端点。详见第 9.3 节与 9.4 节 MCP 协议详解。

```python
class MCPToolRegistry:
    """MCP 协议的工具注册"""

    def __init__(self):
        self.mcp_servers = {}
        self.available_tools = {}

    async def discover_mcp_tools(self, server_url: str):
        """发现 MCP 服务器的工具"""
        try:
            # 连接到 MCP 服务器
            async with aiohttp.ClientSession() as session:
                async with session.get(f"{server_url}/tools") as resp:
                    tools = await resp.json()
                    return tools
        except Exception as e:
            print(f"Failed to discover MCP tools: {e}")
            return []

    def register_mcp_tool(self, server_url: str,
                        tool_name: str,
                        tool_spec: Dict):
        """注册来自 MCP 的工具"""
        if server_url not in self.mcp_servers:
            self.mcp_servers[server_url] = {}

        self.mcp_servers[server_url][tool_name] = tool_spec
        self.available_tools[tool_name] = {
            "source": "mcp",
            "server": server_url,
            "spec": tool_spec
        }

    async def execute_mcp_tool(self, tool_name: str,
                              input_data: Dict) -> Any:
        """执行 MCP 工具"""
        tool_info = self.available_tools.get(tool_name)
        if not tool_info:
            raise ValueError(f"Tool not found: {tool_name}")

        server_url = tool_info["server"]

        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{server_url}/tools/call",
                json={"name": tool_name, "arguments": input_data}
            ) as resp:
                return await resp.json()
```

## 5.4.6 FileStateCache 上下文缓存

实现如下：

```python
class FileStateCache:
    """文件状态缓存:减少重复工具调用"""

    def __init__(self, ttl: int = 60):  # 缓存时间(秒)
        self.cache = {}
        self.timestamps = {}
        self.ttl = ttl

    def get(self, key: str) -> Optional[Dict]:
        """获取缓存"""
        if key in self.cache:
            age = time.time() - self.timestamps[key]
            if age < self.ttl:
                return self.cache[key]
            else:
                del self.cache[key]
                del self.timestamps[key]
        return None

    def set(self, key: str, value: Dict):
        """设置缓存"""
        self.cache[key] = value
        self.timestamps[key] = time.time()

    def invalidate(self, pattern: str):
        """失效匹配模式的缓存"""
        import re
        regex = re.compile(pattern)
        to_delete = [k for k in self.cache.keys() if regex.match(k)]
        for key in to_delete:
            del self.cache[key]
            del self.timestamps[key]

# 使用:缓存文件列表结果
cache = FileStateCache()

async def file_list_with_cache(directory: str) -> List[str]:
    """带缓存的文件列表"""
    cache_key = f"file_list:{directory}"

    # 检查缓存
    cached = cache.get(cache_key)
    if cached:
        return cached["files"]

    # 执行结构化文件列表工具，避免把目录名拼进 shell 命令
    files = await file_list_tool.call({"path": directory})

    # 缓存结果
    cache.set(cache_key, {"files": files})

    return files
```

## 5.4.7 本节小结

动态工具加载是现代智能体系统的必需能力：

1. **shouldDefer 延迟加载**：基于上下文决定是否加载工具
2. **6 级优先级加载**：系统 → 会话 → 用户 → 领域 → 内置 → 发现
3. **Schema 缓存**：减少重复构建 Schema 的开销
4. **MCP 动态注册**：支持第三方工具的即插即用
5. **FileStateCache**：减少重复工具调用，提高性能

这些机制共同构成了一个灵活、高效、可扩展的工具加载系统。


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://yeasy.gitbook.io/harness_engineering_guide/di-er-bu-fen-harness-he-xin-zi-xi-tong/05_tool_layer/5.4_dynamic_loading.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.