摘要

OpenClaw 的插件系统是框架扩展性的核心。本文从插件架构设计出发，深入拆解插件注册机制、生命周期管理、依赖注入模式，以及从零开发一个自定义插件的完整流程。通过一个实战案例——"天气预报查询插件"的完整开发过程，你将掌握插件清单（manifest）编写、工具（Tool）注册、能力（Capability）声明、以及插件的调试与发布全流程。读完你会发现：原来让 AI Agent 拥有新能力，只需要一个插件脚本。

---

1. 引言：当内置工具不够用的时候

1.1 真实场景

先描述一个场景。

你的 OpenClaw Agent 目前已经很强大了——能读写文件、能搜索网页、能操作浏览器、能发送消息。内置的 exec、read、write、browser、message 等工具覆盖了大部分日常需求。

但有一天，老板说：“让 Agent 帮我查下明天上海的天气，自动推送到群里。”

你打开 OpenClaw 的工具列表一看——没有查天气的工具。

怎么办？你有几个选择：

方案	做法	问题
改源码	直接修改 OpenClaw 核心代码，新增 weather 工具	维护噩梦，升级就丢
写 Skill	创建 Skill，用 Python 脚本调用天气 API	Skill 太重，只是要一个工具
写插件	开发一个插件，注册 weather 工具	✅ 独立、轻量、可复用

这就是插件的价值——在不修改 OpenClaw 核心代码的前提下，为 Agent 添加新能力。

1.2 插件 vs Skill vs 内置工具

这三个概念容易混淆，先把它们的关系理清楚：

维度	内置工具	插件	Skill
职责	提供基础系统能力	扩展特定领域能力	编排多个工具完成复杂任务
修改核心	是	否	否
开发语言	TypeScript（核心）	Python / TypeScript	Markdown + Python
部署方式	随 Gateway 发布	独立文件/包	SKILL.md + 脚本
适用场景	文件、网络、消息	外部API、数据库、硬件	业务流程、自动化

💡 一句话总结：内置工具是"手"，插件是"手指"，Skill 是"怎么用这双手完成一个任务"。

---

2. 插件系统架构详解

2.1 整体架构

OpenClaw 的插件系统采用了经典的微内核架构——核心保持最小化，功能通过插件扩展。

2.2 插件清单（Manifest）

每个插件都必须包含一个 plugin.json（或 manifest.json）文件，这是插件的"身份证"：

{
  "name": "weather-plugin",
  "version": "1.0.0",
  "description": "查询全球主要城市天气预报，支持实时天气和未来7天预报",
  "author": "zhang-longsheng",
  "license": "MIT",
  
  "capabilities": [
    {
      "type": "tool",
      "id": "get_weather",
      "description": "查询指定城市的天气信息",
      "parameters": {
        "type": "object",
        "properties": {
          "city": {
            "type": "string",
            "description": "城市名称，如'上海'、'Beijing'"
          },
          "days": {
            "type": "number",
            "description": "预报天数（1-7），默认1",
            "default": 1
          }
        },
        "required": ["city"]
      }
    }
  ],
  
  "dependencies": {
    "requests": "^2.28.0"
  },
  
  "runtime": {
    "type": "python",
    "version": ">=3.9",
    "entry": "main.py"
  },
  
  "permissions": [
    "network:api.openweathermap.org",
    "filesystem:read:/tmp/weather_cache"
  ],
  
  "config": {
    "api_key": {
      "type": "string",
      "description": "OpenWeatherMap API Key",
      "required": true,
      "secret": true
    },
    "units": {
      "type": "string",
      "enum": ["metric", "imperial", "standard"],
      "default": "metric",
      "description": "温度单位"
    }
  }
}

Manifest 关键字段说明：

字段	必填	说明
name	✅	插件的唯一标识符
version	✅	语义化版本号
capabilities	✅	插件提供的能力列表（工具、中间件等）
dependencies	❌	运行时依赖的第三方包
runtime	✅	运行环境（Python/Node.js/Shell）
permissions	✅	插件需要的权限声明
config	❌	插件可配置项

2.3 工具注册机制

工具注册是插件系统最核心的机制。它让 AI Agent 能"发现"并使用插件提供的工具。

工具注册的关键代码（Gateway 内部）：

# 这是 OpenClaw Gateway 内部工具注册的简化逻辑
class ToolRegistry:
    """工具注册中心——管理所有可用工具"""
    
    def __init__(self):
        self._tools = {}  # tool_id -> ToolDefinition
        self._builtin_tools = {}  # 内置工具
        self._plugin_tools = {}   # 插件工具
    
    def register(self, tool_def: ToolDefinition, source: str = "plugin"):
        """注册一个新工具"""
        if tool_def.id in self._tools:
            raise ConflictError(f"工具 {tool_def.id} 已存在")
        
        # 1. 验证参数schema
        validate_parameter_schema(tool_def.parameters)
        
        # 2. 注册到工具索引
        self._tools[tool_def.id] = tool_def
        
        # 3. 标记来源
        if source == "plugin":
            self._plugin_tools[tool_def.id] = tool_def
        
        logger.info(f"工具已注册: {tool_def.id} (来源: {source})")
    
    def unregister(self, tool_id: str):
        """注销一个工具"""
        if tool_id not in self._tools:
            raise NotFoundError(f"工具 {tool_id} 未找到")
        del self._tools[tool_id]
        if tool_id in self._plugin_tools:
            del self._plugin_tools[tool_id]
        logger.info(f"工具已注销: {tool_id}")
    
    def list_for_agent(self, agent_id: str) -> list:
        """返回Agent可用的工具列表"""
        # 过滤掉Agent无权限的工具
        available = []
        for tool_id, tool in self._tools.items():
            if self._check_permission(tool, agent_id):
                available.append(tool)
        return available

💡 上述代码展示了工具注册的核心流程：注册 → 验证 → 索引 → 权限过滤。实际生产代码还会涉及并发安全和缓存优化。

---

3. 自定义插件开发实战

3.1 项目结构

让我们从零开发一个天气预报查询插件，完整的项目结构：

weather-plugin/
├── plugin.json          # 插件清单（身份证）
├── main.py              # 插件入口（核心逻辑）
├── requirements.txt     # Python 依赖
└── README.md            # 使用文档

3.2 核心代码实现

main.py —— 插件的核心逻辑：

"""
weather-plugin/main.py
天气预报查询插件的核心实现

该插件演示了 OpenClaw 插件开发的完整模式：
1. 工具声明 → 被 Agent 发现
2. 工具执行 → 被 Agent 调用
3. 错误处理 → 优雅降级
4. 配置管理 → 外部化参数
"""

import json
import os
import time
from typing import Optional, Dict, Any
import requests  # 外部依赖在 manifest.json 中声明

# ============================================
# 1. 插件元数据（从 manifest 中补充运行时信息）
# ============================================
PLUGIN_META = {
    "capabilities": ["tool:get_weather", "tool:get_forecast"],
    "events_listened": [],
    "hooks_implemented": ["on_load", "on_unload"]
}


# ============================================
# 2. 配置管理（从 plugin.json 的 config 段读取）
# ============================================
class PluginConfig:
    """插件的配置管理器
    
    负责从环境变量、配置文件、或 Gateway 配置中读取插件参数。
    优先级：环境变量 > Gateway 配置 > 默认值
    """
    
    def __init__(self):
        # API Key优先从环境变量读取（安全最佳实践）
        self.api_key = os.environ.get(
            "OPENWEATHER_API_KEY",
            "your_default_key_here"
        )
        self.units = os.environ.get("WEATHER_UNITS", "metric")
        self.cache_enabled = True
        self.cache_ttl = 600  # 缓存10分钟
        self._cache = {}
    
    def get_cache(self, city: str) -> Optional[dict]:
        if not self.cache_enabled:
            return None
        cached = self._cache.get(city)
        if cached and (time.time() - cached["ts"]) < self.cache_ttl:
            return cached["data"]
        return None
    
    def set_cache(self, city: str, data: dict):
        self._cache[city] = {"data": data, "ts": time.time()}


# ============================================
# 3. 工具实现
# ============================================

class WeatherTool:
    """天气预报工具实现
    
    这个类封装了天气查询的核心逻辑，包括：
    - API 调用
    - 缓存处理
    - 错误降级
    - 结果格式化
    """
    
    BASE_URL = "https://api.openweathermap.org/data/2.5/weather"
    
    def __init__(self, config: PluginConfig):
        self.config = config
    
    def get_weather(self, city: str, days: int = 1) -> Dict[str, Any]:
        """
        查询指定城市的天气信息
        
        Args:
            city: 城市名称，支持中英文
            days: 预报天数（1-7），默认当天
        
        Returns:
            标准化的天气信息字典
        """
        # 1. 检查缓存
        cached = self.config.get_cache(city)
        if cached:
            return {"source": "cache", **cached}
        
        # 2. 构建 API 请求
        params = {
            "q": city,
            "appid": self.config.api_key,
            "units": self.config.units,
            "lang": "zh_cn"
        }
        
        try:
            # 3. 调用天气 API
            resp = requests.get(
                self.BASE_URL,
                params=params,
                timeout=10
            )
            resp.raise_for_status()
            
            data = resp.json()
            
            # 4. 格式化结果为结构化数据
            weather_info = {
                "city": data.get("name", city),
                "temperature": data["main"]["temp"],
                "feels_like": data["main"]["feels_like"],
                "humidity": data["main"]["humidity"],
                "pressure": data["main"]["pressure"],
                "weather": data["weather"][0]["description"],
                "wind_speed": data["wind"]["speed"],
                "visibility": data.get("visibility", 0),
                "timestamp": int(time.time())
            }
            
            # 5. 写入缓存
            self.config.set_cache(city, weather_info)
            
            return {"source": "api", **weather_info}
            
        except requests.exceptions.Timeout:
            # 超时降级——返回本地缓存或默认值
            return {
                "source": "fallback",
                "city": city,
                "error": "API请求超时",
                "suggestion": "请稍后重试或检查网络连接"
            }
        
        except requests.exceptions.HTTPError as e:
            # 处理 API 错误（如城市不存在）
            if e.response.status_code == 404:
                return {
                    "source": "fallback",
                    "city": city,
                    "error": f"未找到城市'{city}'的天气信息",
                    "suggestion": "请检查城市名称是否正确"
                }
            raise


# ============================================
# 4. 插件生命周期钩子
# ============================================

config = PluginConfig()
tool = None

def on_load():
    """插件加载时的初始化逻辑
    
    在 OpenClaw Gateway 启动或热加载插件时自动调用。
    这里完成：
    - 配置初始化
    - 依赖检查
    - 连接建立
    """
    global config, tool
    
    # 验证必要配置
    if not config.api_key or config.api_key == "your_default_key_here":
        print("⚠️ 天气预报插件：API Key 未配置，将使用模拟数据")
    
    tool = WeatherTool(config)
    print(f"✅ 天气预报插件已加载 (units: {config.units})")
    return {"status": "loaded", "capabilities": PLUGIN_META["capabilities"]}


def on_unload():
    """插件卸载时的清理逻辑
    
    在 OpenClaw Gateway 关闭或插件被禁用时自动调用。
    这里完成：
    - 连接关闭
    - 缓存清理
    - 资源释放
    """
    global config, tool
    config._cache.clear()
    tool = None
    print("🛑 天气预报插件已卸载")
    return {"status": "unloaded"}


# ============================================
# 5. 工具导出（核心接口）
# ============================================

def handle_get_weather(params: dict) -> dict:
    """处理 get_weather 工具调用
    
    这是 Agent 调用工具时的入口函数。
    函数名格式必须为 handle_{tool_id}
    
    Args:
        params: Agent 传入的参数字典
    
    Returns:
        工具执行结果
    """
    city = params.get("city", "北京")
    days = params.get("days", 1)
    
    if not tool:
        return {"error": "插件未初始化"}
    
    result = tool.get_weather(city, days)
    return result


# 导出函数映射表——Plugin Manager 通过此表发现工具调用入口
EXPORTS = {
    "get_weather": handle_get_weather
}

💡 这段代码展示了插件开发的四大模块：配置管理 → 工具实现 → 生命周期钩子 → 工具导出。将每部分独立是为了保证插件的可测试性和可维护性。

3.3 插件配置与安装

安装到 OpenClaw：

# 1. 将插件目录复制到 Gateway 的 plugins 目录
cp -r weather-plugin/ /etc/openclaw/plugins/

# 2. 在 openclaw.yaml 中启用插件

# openclaw.yaml
plugins:
  enabled: true
  directory: /etc/openclaw/plugins
  
  # 启用特定插件
  installed:
    weather-plugin:
      enabled: true
      config:
        api_key: "${OPENWEATHER_API_KEY}"  # 从环境变量读取
        units: metric
  
  # 插件管理策略
  management:
    auto_update: false
    fail_strategy: "continue"  # 插件加载失败不阻塞 Gateway 启动
    hot_reload: true           # 开发模式下启用热重载

# 3. 重启 Gateway 或触发热重载
openclaw gateway restart
# 或
openclaw plugins reload weather-plugin

3.4 测试插件

"""
test_weather_plugin.py
插件测试脚本——验证工具是否正常工作
"""

import json
from main import on_load, handle_get_weather

# 1. 初始化插件
result = on_load()
print(f"插件状态: {result['status']}")
print(f"提供的能力: {result['capabilities']}")

# 2. 测试天气查询
test_cases = [
    {"city": "上海", "days": 1},
    {"city": "Beijing", "days": 3},
    {"city": "不存在城市XYZ", "days": 1},
]

for params in test_cases:
    print(f"\n📝 查询: {params}")
    result = handle_get_weather(params)
    if "error" in result:
        print(f"   ❌ 错误: {result['error']}")
        print(f"   💡 建议: {result.get('suggestion', '无')}")
    else:
        print(f"   🌤️ {result['city']}: {result['weather']}, "
              f"{result['temperature']}°C, "
              f"湿度 {result['humidity']}%")
    print(f"   数据来源: {result.get('source', 'unknown')}")

预期输出：

✅ 天气预报插件已加载 (units: metric)
插件状态: loaded
提供的能力: ['tool:get_weather', 'tool:get_forecast']

📝 查询: {'city': '上海', 'days': 1}
   🌤️ 上海: 晴, 28.0°C, 湿度 65%
   数据来源: api

📝 查询: {'city': 'Beijing', 'days': 3}
   🌤️ Beijing: 多云, 22.0°C, 湿度 45%
   数据来源: api

📝 查询: {'city': '不存在城市XYZ', 'days': 1}
   ❌ 错误: 未找到城市'不存在城市XYZ'的天气信息
   💡 建议: 请检查城市名称是否正确
   数据来源: fallback

---

4. 插件生命周期管理

4.1 完整生命周期

一个 OpenClaw 插件的完整生命周期包括6个阶段：

4.2 生命周期各阶段详解

阶段	触发事件	插件需做的事	回调函数
安装	复制到 plugins/ 目录	无	—
验证	Gateway 启动/热加载	检查依赖、验证配置	on_validate()
激活	验证通过	初始化连接、预热缓存	on_load()
运行	持续运行中	响应工具调用	handle_*()
暂停	管理员暂停	关闭连接、停止接收新请求	on_pause()
卸载	移除插件	释放资源、清理数据	on_unload()

4.3 热重载机制

开发模式下，修改插件代码后无需重启整个 Gateway：

plugins:
  management:
    hot_reload: true
    watch_interval: 5  # 每5秒检查一次文件变化

热重载的流程：

---

5. 高级插件模式

5.1 中间件插件

除了提供"工具"，插件还可以作为"中间件"——在 Agent 处理流程的特定节点插入逻辑：

"""
logging-middleware/main.py
请求日志中间件插件——记录所有工具调用的详细信息
"""

import time
import json

EXPORTS = {}

def on_load():
    print("✅ 日志中间件已加载")
    return {"status": "loaded", "capabilities": ["middleware:tool_logger"]}

# 中间件钩子——在工具调用前后执行
def before_tool_call(tool_id: str, params: dict, agent_id: str):
    """工具调用前触发"""
    return {
        "timestamp": time.time(),
        "tool_id": tool_id,
        "agent_id": agent_id,
        "params": params
    }

def after_tool_call(tool_id: str, result: dict, context: dict):
    """工具调用后触发——记录结果和耗时"""
    elapsed = time.time() - context.get("timestamp", 0)
    log_entry = {
        "tool": tool_id,
        "duration_ms": round(elapsed * 1000),
        "success": "error" not in result,
        "result_size": len(json.dumps(result))
    }
    print(f"📊 [Middleware] {log_entry}")
    return log_entry

# 注册中间件钩子
EXPORTS["middleware"] = {
    "before_tool_call": before_tool_call,
    "after_tool_call": after_tool_call
}

💡 中间件模式适合：请求日志记录、性能监控、权限二次校验、数据脱敏等横切关注点。它不改变工具的核心逻辑，只在外层做拦截处理。

5.2 事件驱动插件

插件还可以监听 OpenClaw 的内部事件：

# event-listener/main.py

EVENTS = {
    "gateway.started",       # Gateway 启动完成
    "session.created",       # 新会话创建
    "message.received",      # 收到新消息
    "tool.called",           # 工具被调用
    "model.response",        # 模型返回响应
    "channel.connected",     # 渠道连接
}

def on_event(event_type: str, payload: dict):
    """统一事件处理器"""
    if event_type == "message.received":
        # 自动拦截特定内容
        if "紧急" in payload.get("content", ""):
            print(f"🚨 检测到紧急消息: {payload.get('session_id')}")
    
    elif event_type == "channel.connected":
        print(f"📡 渠道已连接: {payload.get('channel')}")

5.3 多能力插件

一个插件可以提供多种能力类型：

{
  "name": "enterprise-wechat-plugin",
  "version": "2.0.0",
  "capabilities": [
    {
      "type": "tool",
      "id": "send_wechat_msg",
      "description": "发送企业微信消息"
    },
    {
      "type": "tool",
      "id": "get_wechat_contacts",
      "description": "获取企业微信通讯录"
    },
    {
      "type": "middleware",
      "id": "wechat_auth",
      "description": "企业微信身份验证中间件"
    },
    {
      "type": "event_handler",
      "id": "wechat_webhook",
      "description": "企业微信回调事件处理",
      "events": ["wechat.message.received", "wechat.member.added"]
    }
  ]
}

---

6. 插件调试与排错

6.1 常见问题排查

症灶	可能原因	排查命令
插件未加载	manifest.json 格式错误	openclaw plugins validate weather-plugin
工具未被发现	capabilities 声明遗漏	openclaw plugins inspect weather-plugin
运行时404	工具ID 与 handle 函数不匹配	检查 handle_{tool_id} 命名
权限拒绝	permissions 声明不全	openclaw plugins check-perm weather-plugin
依赖缺失	requirements.txt 未安装	pip install -r requirements.txt

6.2 开启调试日志

logging:
  plugins: debug  # 输出插件详细日志
  tools: debug    # 输出工具调用日志

---

7. 总结

本文从零开始，完整走通了 OpenClaw 插件系统的开发全流程：

核心要点：

1. 插件 vs Skill vs 内置工具：插件是中间层——比内置工具更灵活（不改核心），比 Skill 更轻量（专注单一工具）

2. 三个核心文件就能搞定一个插件：plugin.json（声明能力）+ main.py（实现逻辑）+ requirements.txt（声明依赖）

3. 四个生命周期钩子：on_validate → on_load → on_pause → on_unload，覆盖插件全生命周期

4. 三种插件模式：工具插件（提供新 tool）、中间件插件（拦截请求）、事件驱动插件（监听事件）

5. 热重载是开发利器：开发时启用 hot_reload: true，修改代码无需重启 Gateway

思考题：

1. 如果你要开发一个"数据库查询"插件，你会如何设计工具的参数 schema 和返回格式，让 AI Agent 能"理解"查询结果？

2. 中间件插件的执行顺序如何设计？如果两个中间件都拦截了 before_tool_call，谁先执行？如何避免循环拦截？

3. 插件的权限声明机制（permissions）本质上是一种沙箱策略。你认为这种策略足够安全吗？还有哪些潜在的安全风险？

---

参考资料

• OpenClaw 官方文档 - 插件开发指南
• JSON Schema 规范
• OpenWeatherMap API 文档
• Python requests 库文档
• 微内核架构模式