> For the complete documentation index, see [llms.txt](https://yeasy.gitbook.io/harness_engineering_guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://yeasy.gitbook.io/harness_engineering_guide/di-san-bu-fen-xi-tong-ji-cheng-yu-gong-cheng-shi-jian/10_production/10.2_plugin_system.md).

# 10.2 插件与扩展体系设计

灵活的扩展机制是支持多样化应用场景的关键。本节介绍四层扩展架构(Plugin、Skill、Hook、Command)，对比 Claude Code 和 OpenClaw 的实现模式，并提供完整的插件系统实现和最佳实践。

## 10.2.1 概述

AI Agent Harness 需要灵活的扩展机制来支持不同的应用场景和第三方集成。本节介绍四层扩展架构：Plugin（插件）、Skill（技能）、Hook（钩子）、Command（命令），以及两大参考系统（Claude Code 和 OpenClaw）的实现模式。

**四层扩展架构**

```mermaid
graph TD
    A["<b>Application Layer</b>"]
    B["<b>Command Layer</b><br/>用户交互入口"]
    C["<b>Hook Layer</b><br/>执行流程插桩点"]
    D["<b>Skill Layer</b><br/>原子功能封装"]
    E["<b>Plugin Layer</b><br/>插件注册与加载"]

    A --> B
    B --> C
    C --> D
    D --> E

    style A fill:#e3f2fd
    style B fill:#fff3e0
    style C fill:#f3e5f5
    style D fill:#e8f5e9
    style E fill:#fce4ec
```

图 10-2：四层扩展架构

## 10.2.2 基础概念

### Plugin

插件是独立的模块包，包含一个或多个 Skill、Hook 和 Command。

**特点：**

* 自描述元数据(openclaw\.plugin.json)
* 独立的版本管理
* 权限声明与隔离
* 可动态加载卸载

### Skill

技能是原子级的功能单位，对应一个具体的 Agent 能力。

**分类：**

* SkillTool：工具调用技能（含参数 schema、返回类型）
* SkillQuery：查询技能（检索增强）
* SkillTransform：转换技能（数据处理）

在生产级 Harness 中，Skill 不只是插件包里的静态说明文件，也可以成为可训练、可版本化、可回滚的过程性知识资产。一个可借鉴的研究型方向是把 Skill 变更纳入评估流水线：通过 scored rollout 收集成功和失败轨迹，由优化器提出小范围编辑，再用 held-out 任务集决定是否接受。这样可以把“靠经验维护 Skill”升级为“像配置训练一样维护 Skill”，同时仍保持部署产物轻量：线上只加载最终 Skill 文件，不引入额外推理调用。

OpenAI 2026 年的 Codex 角色插件提供了一个产品化参照：插件不只是工具集合，也可以把应用连接、skills、instructions 和 workflows 打包成面向角色或团队的 Harness 能力包。设计内部插件时，应优先抽象“这个团队反复完成什么工作、依赖哪些系统、需要哪些验收标准”，而不是只列出可调用 API。

### Hook

Hook 是执行流程中的插桩点，允许插件在关键时刻介入。

**OpenClaw 7 种钩子类型：**

| 钩子类型                   | 触发时机     | 用例            |
| ---------------------- | -------- | ------------- |
| `command:new`          | 新命令注册    | 验证命令参数 schema |
| `gateway:startup`      | 系统启动     | 初始化外部连接       |
| `tool_result_persist`  | 工具结果保存   | 审计日志记录        |
| `before_model_resolve` | 模型解析前    | 注入系统提示词       |
| `before_prompt_build`  | 提示词构建前   | 上下文增强         |
| `before_tool_call`     | 工具调用前    | 参数验证/转换       |
| `agent_end`            | Agent 结束 | 清理资源/回调       |

### Command

命令是用户交互的直接入口，映射到一个 Skill 或流程。

理解了这四层扩展架构的基本概念后，我们来看看两个业界成熟的参考实现系统如何将这些概念落地。首先介绍 Claude Code 的扩展设计方案。

## 10.2.3 Claude Code 参考系统

> 注：Claude Code 自 2025-10 起对外开放 **Plugins** 机制和 **Plugin Marketplace**（通过 `.claude-plugin/marketplace.json` 描述包，`/plugin` 命令安装/启用；包内可携带 Skill/Command/Hook 三类扩展并通过 manifest 声明权限）。本节描述的扩展类型即对应 Plugin 内部的能力单元。具体 schema 见 code.claude.com 官方文档。

### 三种扩展机制

**1. Skill（能力扩展）**

Claude Code 插件是文件型扩展包，而不是导入 `claude_code` Python SDK。典型结构包括 `.claude-plugin/plugin.json`、`skills/<name>/SKILL.md`、`commands/`、`agents/`、`hooks/hooks.json` 和可选 `.mcp.json`。数据库查询这类能力应先写成 Skill/Command 文档和受控 MCP 工具，再由插件 manifest 声明权限与入口。

**2. Commands（命令扩展）**

项目级自定义命令以 Markdown 文件形式放在 `.claude/commands/`：

```markdown
---
description: Perform semantic search
argument-hint: [query]
allowed-tools: Bash(python scripts/search.py:*)
---

Run semantic search for:

$ARGUMENTS

Use the project search helper:

!`python scripts/search.py "$ARGUMENTS"`
```

**3. Hooks（流程钩子）**

Hook 通过设置文件声明，由 Claude Code 在指定事件触发时运行命令：

```json
{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash|Write|Edit",
        "hooks": [
          {
            "type": "command",
            "command": "$CLAUDE_PROJECT_DIR/.claude/hooks/audit-tool-call.py",
            "timeout": 10
          }
        ]
      }
    ]
  }
}
```

### 编译时与运行时 Feature Gates

Feature Gates 应由宿主应用或插件清单实现。Claude Code 的公开扩展面不要求插件导入 Python SDK；常见做法是在命令、Hook 脚本或 MCP 工具中读取环境变量、配置文件或远端开关：

```python
import os

def advanced_search_handler(query: str):
    """仅在启用 advanced_search 时执行高级检索"""
    if os.environ.get("ADVANCED_SEARCH_ENABLED") != "1":
        return perform_keyword_search(query)
    return perform_semantic_search(query)

class RecommendationEngine:
    def __init__(self, feature_flags: dict[str, bool]):
        self.enable_ml_ranking = feature_flags.get("ml_ranking", False)

    def rank_results(self, results):
        if self.enable_ml_ranking:
            return self.ml_rank(results)
        else:
            return self.simple_rank(results)
```

Claude Code 展示了轻量级但功能完整的扩展机制。与之相比，OpenClaw 采取了更强大、更标准化的插件规范，使大规模生产环境中的插件管理更加规范和可扩展。

## 10.2.4 OpenClaw 参考系统

### openclaw\.plugin.json 规范

openclaw\.plugin.json 规范的结构示例如下：

```json
{
  "name": "custom-search",
  "version": "1.0.0",
  "description": "Custom search plugin with ranking",
  "main": "dist/index.js",
  "permissions": [
    {
      "resource": "tool:search",
      "action": "call",
      "level": "Ask-first"
    },
    {
      "resource": "file:read",
      "action": "access",
      "level": "Approve-once"
    }
  ],
  "skills": [
    {
      "id": "custom-search",
      "name": "Custom Search",
      "type": "SkillTool",
      "params": {
        "type": "object",
        "properties": {
          "query": { "type": "string", "description": "Search query" },
          "limit": { "type": "integer", "minimum": 1, "maximum": 100 }
        },
        "required": ["query"]
      }
    }
  ],
  "hooks": [
    {
      "type": "before_tool_call",
      "handler": "validateSearchParams"
    },
    {
      "type": "tool_result_persist",
      "handler": "logSearchResult"
    }
  ],
  "commands": [
    {
      "name": "search",
      "alias": ["/search", "搜索"],
      "skill": "custom-search",
      "level": "Free"
    }
  ]
}
```

### 权限三级模型

权限三级模型的实现如下：

```python
class PermissionLevel(Enum):
    FREE = "Free"              # 无需审批
    ASK_FIRST = "Ask-first"    # 执行前询问
    APPROVE_ONCE = "Approve-once"  # 任务级一次授权
```

**应用场景：**

* `Free`：查询类操作、读取非敏感数据
* `Ask-first`：修改操作、外部 API 调用前询问
* `Approve-once`：计划明确的批量操作，在任务开始时一次性授权

### TypeScript 插件示例

TypeScript 插件的实现示例如下：

```typescript
// plugins/custom-search.ts
import { Plugin, Hook, Skill, Command } from '@openclaw/core';

export default class CustomSearchPlugin extends Plugin {
  async initialize() {
    // 注册 Skill
    this.registerSkill('custom-search', {
      handler: this.performSearch.bind(this),
      schema: {
        type: 'object',
        properties: {
          query: { type: 'string' },
          limit: { type: 'integer', default: 10 }
        }
      }
    });

    // 注册 Hook
    this.registerHook('before_tool_call', async (context) => {
      const { skillId, params } = context;
      if (skillId === 'custom-search') {
        // 验证查询长度
        if (params.query.length > 500) {
          throw new Error('Query too long');
        }
      }
    });

    // 注册 Command
    this.registerCommand({
      name: 'search',
      aliases: ['/search', '搜索'],
      skill: 'custom-search',
      level: 'Ask-first'
    });
  }

  async performSearch(params: { query: string; limit: number }) {
    // 实际搜索逻辑
    return {
      results: [],
      total: 0,
      timestamp: Date.now()
    };
  }
}
```

### ClawHub 技能生态

ClawHub 是 OpenClaw 的公开 skills/plugins 注册中心。由于注册数量变化很快，正文不硬编码技能总数，选型时应以 ClawHub 当前页面或 API 为准。

```python
class ClawHubRegistry:
    """ClawHub 技能注册与发现"""

    def search_skills(self, query: str, category: str = None):
        """搜索 ClawHub 技能"""
        # 查询:query="数据分析", category="data"
        # 返回匹配的技能列表

    def install_skill(self, skill_id: str, version: str = "latest"):
        """从 ClawHub 安装技能"""

    def publish_skill(self, plugin_config: dict, code: str):
        """发布技能到 ClawHub"""
```

### “错误作为观察”模式

OpenClaw 独特的错误处理哲学：系统不抛出异常打断流程，而是将错误信息注入观察流。

```python
from datetime import datetime

class ErrorAsObservation:
    """错误作为观察的实现"""

    async def handle_tool_error(self, skill_id: str, error: Exception, context):
        # 不中断,转化为观察
        observation = {
            "type": "tool_error",
            "skill": skill_id,
            "error_code": error.__class__.__name__,
            "message": str(error),
            "timestamp": datetime.now().isoformat(),
            "recoverable": self.is_recoverable(error),
            "suggested_action": self.suggest_recovery(error)
        }

        # 注入到 Agent 观察流
        await self.inject_observation(observation)

        # 如果可恢复,Agent 可选择重试或替代方案
        # 如果不可恢复,Agent 报告给用户
```

在掌握了 Claude Code 和 OpenClaw 两种参考实现的核心思想后，我们现在来展示如何将这些最佳实践整合到一个完整的、可扩展的插件架构系统中。

## 10.2.5 实战：可扩展插件架构设计

### 架构设计图

插件系统的完整架构设计如下：

```mermaid
graph TD
    A["<b>Plugin Manifest</b><br/>openclaw.plugin.json"]

    A --> B["<b>Skills</b><br/>原子功能"]
    A --> C["<b>Hooks</b><br/>插桩点"]
    A --> D["<b>Commands</b><br/>用户入口"]

    B --> E["<b>Permission</b><br/>Resolver"]
    C --> E
    D --> E

    E --> F["<b>Plugin Runtime</b><br/>(Lifecycle Mgmt)"]

    G["Discover"] --> H["Load"] --> I["Validate"] --> J["Register"] --> K["Execute"]

    style A fill:#e1f5ff
    style E fill:#fff3cd
    style F fill:#d4edda
    style B fill:#f8f9fa
    style C fill:#f8f9fa
    style D fill:#f8f9fa
```

图 10-3：插件系统完整架构与生命周期

### 完整代码实现

可扩展插件架构的完整实现代码如下：

```python
# core/plugin_manager.py
import json
import importlib.util
from typing import Dict, Any, List
from dataclasses import dataclass
from enum import Enum

class PermissionLevel(Enum):
    FREE = "Free"
    ASK_FIRST = "Ask-first"
    APPROVE_ONCE = "Approve-once"

@dataclass
class PluginManifest:
    name: str
    version: str
    main: str
    skills: List[Dict[str, Any]]
    hooks: List[Dict[str, str]]
    commands: List[Dict[str, Any]]
    permissions: List[Dict[str, Any]]

class PluginManager:
    def __init__(self):
        self.plugins: Dict[str, Any] = {}
        self.skills: Dict[str, Any] = {}
        self.hooks: Dict[str, List[callable]] = {}
        self.commands: Dict[str, Any] = {}

    def load_plugin(self, plugin_dir: str):
        """加载插件"""
        manifest_path = f"{plugin_dir}/openclaw.plugin.json"

        with open(manifest_path, 'r') as f:
            manifest_data = json.load(f)

        manifest = PluginManifest(**manifest_data)
        plugin_id = manifest.name

        # 动态导入插件模块
        spec = importlib.util.spec_from_file_location(
            plugin_id,
            f"{plugin_dir}/{manifest.main}"
        )
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        # 初始化插件类
        plugin_class = module.default
        plugin_instance = plugin_class()

        # 注册 Skills
        for skill_config in manifest.skills:
            self.register_skill(
                skill_config['id'],
                plugin_instance,
                skill_config
            )

        # 注册 Hooks
        for hook_config in manifest.hooks:
            self.register_hook(
                hook_config['type'],
                getattr(plugin_instance, hook_config['handler'])
            )

        # 注册 Commands
        for cmd_config in manifest.commands:
            self.register_command(cmd_config, plugin_id)

        self.plugins[plugin_id] = {
            'manifest': manifest,
            'instance': plugin_instance
        }

        return manifest

    def register_skill(self, skill_id: str, plugin, config: dict):
        """注册技能"""
        self.skills[skill_id] = {
            'plugin': plugin,
            'config': config,
            'handler': getattr(plugin, f'skill_{skill_id}', None)
        }

    def register_hook(self, hook_type: str, handler: callable):
        """注册钩子"""
        if hook_type not in self.hooks:
            self.hooks[hook_type] = []
        self.hooks[hook_type].append(handler)

    def register_command(self, cmd_config: dict, plugin_id: str):
        """注册命令"""
        cmd_id = cmd_config['name']
        self.commands[cmd_id] = {
            'plugin': plugin_id,
            'skill': cmd_config['skill'],
            'level': PermissionLevel[cmd_config['level'].upper().replace('-', '_')],
            'aliases': cmd_config.get('aliases', [])
        }

    async def execute_hooks(self, hook_type: str, context: dict):
        """执行钩子链"""
        if hook_type in self.hooks:
            for handler in self.hooks[hook_type]:
                await handler(context)

    async def call_skill(self, skill_id: str, params: dict) -> Any:
        """调用技能"""
        skill = self.skills.get(skill_id)
        if not skill:
            raise ValueError(f"Skill {skill_id} not found")

        # 执行 before_tool_call 钩子
        await self.execute_hooks('before_tool_call', {
            'skillId': skill_id,
            'params': params
        })

        # 调用技能处理器
        result = await skill['handler'](**params)

        # 执行 tool_result_persist 钩子
        await self.execute_hooks('tool_result_persist', {
            'skillId': skill_id,
            'result': result
        })

        return result

    def list_plugins(self) -> List[str]:
        """列出已加载的插件"""
        return list(self.plugins.keys())

    def get_plugin_info(self, plugin_id: str) -> Dict[str, Any]:
        """获取插件信息"""
        if plugin_id not in self.plugins:
            return None

        plugin = self.plugins[plugin_id]
        return {
            'name': plugin['manifest'].name,
            'version': plugin['manifest'].version,
            'skills': [s['id'] for s in plugin['manifest'].skills],
            'commands': [c['name'] for c in plugin['manifest'].commands]
        }

# 使用示例
if __name__ == "__main__":
    import asyncio

    manager = PluginManager()

    # 加载插件
    manager.load_plugin("./plugins/custom-search")

    # 调用技能
    result = asyncio.run(manager.call_skill(
        'custom-search',
        {'query': 'AI Agent', 'limit': 10}
    ))

    print(f"Search results: {result}")
```

完成了插件架构的实战设计后，我们在生产环境中还需要遵循一些关键的最佳实践，以确保插件系统的安全、性能和可维护性。

## 10.2.6 最佳实践

### 1. 插件隔离

插件隔离的实现方式如下：

```python
class IsolatedPluginContext:
    """插件隔离上下文"""

    def __init__(self, plugin_id: str):
        self.plugin_id = plugin_id
        self.env_vars = {}
        self.file_access = set()
        self.api_quota = {}

    def grant_permission(self, resource: str, action: str):
        """授予权限"""
        pass

    def check_permission(self, resource: str, action: str) -> bool:
        """检查权限"""
        pass
```

### 2. 版本兼容性

版本兼容性的处理方式如下：

```python
class VersionResolver:
    """解决插件版本冲突"""

    def check_compatibility(self, plugin_a: str, plugin_b: str) -> bool:
        """检查两个插件是否兼容"""
        # 比较依赖声明
        pass

    def resolve_dependency(self, plugins: List[str]) -> List[str]:
        """解决依赖冲突,返回正确的加载顺序"""
        # 拓扑排序
        pass
```

### 3. 热加载与卸载

热加载与卸载的实现方式如下：

```python
class HotReloadManager:
    """支持热加载/卸载插件"""

    async def reload_plugin(self, plugin_id: str):
        """重新加载插件(不中断服务)"""
        old_plugin = self.plugins[plugin_id]
        new_manifest = self.load_plugin(...)

        # 验证兼容性
        if not self.check_backward_compatibility(old_plugin, new_manifest):
            raise IncompatiblePluginError()

        # 平滑切换
        await self.pause_plugin(plugin_id)
        await self.unload_plugin(plugin_id)
        await self.load_plugin(plugin_id)
        await self.resume_plugin(plugin_id)

    async def unload_plugin(self, plugin_id: str):
        """卸载插件并清理资源"""
        plugin = self.plugins[plugin_id]
        await plugin['instance'].cleanup()
        del self.plugins[plugin_id]
```

## 10.2.7 总结

插件与扩展体系是 Harness 的核心竞争力，通过分层的 Plugin-Skill-Hook-Command 架构，实现了：

1. **高内聚**：每个插件独立，职责清晰
2. **低耦合**：通过 Hook 和 Permission 解耦
3. **可扩展**：支持热加载、版本管理、隔离执行
4. **安全**：三级权限模型、LLM 权限解释器
5. **生态**：ClawHub skills/plugins 注册中心

下一节将介绍如何在这个扩展体系上进行性能优化和成本控制。