摘要

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


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

1.1 真实场景

先描述一个场景。

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

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

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

怎么办?你有几个选择:

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

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

1.2 插件 vs Skill vs 内置工具

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

📦 Skill Layer

🧩 Plugin Layer

🔧 OpenClaw Core

内置工具
exec, read, write, browser, message

天气查询插件
tool: get_weather

数据库查询插件
tool: db_query

短信通知插件
tool: send_sms

客服技能
编排多个Tool+Prompt

运维技能
自动化巡检流程

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

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


2. 插件系统架构详解

2.1 整体架构

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

⚡ 运行时

🧩 插件层

🏗️ OpenClaw Gateway

注册/卸载

工具发现

加载/热重载

加载/热重载

加载/热重载

tool.call

tool.call

tool.call

插件管理器
Plugin Manager

工具注册中心
Tool Registry

能力路由
Capability Router

插件A
manifest.json + main.py

插件B
manifest.json + main.ts

插件N
manifest.json + index.js

Python Runtime

Node.js Runtime

Shell Runtime

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 能"发现"并使用插件提供的工具。

AI Agent 工具注册中心 插件管理器 插件 AI Agent 工具注册中心 插件管理器 插件 加载插件(manifest.json) 验证清单合法性 检查依赖与权限 注册工具(get_weather) 索引工具元数据 注册成功 插件已激活 查询可用工具 [exec, read, write, ..., get_weather] 调用 get_weather(city="上海") {temp:28, weather:"晴"}

工具注册的关键代码(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个阶段:

复制到plugins/目录

Gateway启动/热加载

验证通过

验证失败/依赖缺失

开始工作

暂停(维护)

新版本安装

恢复

移除

新版验证

移除

安装

验证

激活

禁用

运行

暂停

升级

卸载

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_validateon_loadon_pauseon_unload,覆盖插件全生命周期

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

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

思考题

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

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

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


参考资料

Logo

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地,聚焦技能开发、插件实践与部署教程,为开发者提供可直接落地的方案、工具与交流平台,助力高效构建与落地 AI 应用

更多推荐