跳转至

插件开发指南

OpenClaw-Py 支持通过 Python entry points 扩展功能。本指南介绍如何开发自定义插件。

1. 插件结构

一个典型的 OpenClaw 插件包含：

my-pyclaw-plugin/
├── pyproject.toml          # 包定义，包含 entry point
├── src/
│   └── my_plugin/
│       ├── __init__.py     # 插件主入口
│       └── tools.py        # 自定义工具（可选）
└── README.md

2. 定义 Entry Point

在 pyproject.toml 中注册插件：

[project]
name = "my-pyclaw-plugin"
version = "0.1.0"

[project.entry-points."pyclaw.plugins"]
my_plugin = "my_plugin:MyPlugin"

3. 实现插件类

# src/my_plugin/__init__.py
from dataclasses import dataclass
from typing import Any, Callable

@dataclass
class PluginInfo:
    """插件元信息"""
    name: str
    version: str
    description: str = ""
    gateway_methods: dict[str, Callable] = None
    tools: list[Any] = None

class MyPlugin:
    """自定义 OpenClaw 插件"""

    @property
    def name(self) -> str:
        return "my-plugin"

    @property
    def version(self) -> str:
        return "0.1.0"

    @property
    def gateway_methods(self) -> dict[str, Callable]:
        """返回网关 RPC 方法"""
        return {
            "my_plugin.hello": self._handle_hello,
        }

    @property
    def tools(self) -> list[Any]:
        """返回 Agent 工具列表"""
        return []

    async def _handle_hello(self, params: dict, conn: Any) -> None:
        """RPC 方法实现"""
        await conn.send_ok("my_plugin.hello", {"message": "Hello from plugin!"})


# 入口函数
def load_plugin() -> MyPlugin:
    return MyPlugin()

4. Gateway Methods

Gateway methods 是通过 WebSocket 暴露的 RPC 接口：

async def my_method(params: dict[str, Any] | None, conn: GatewayConnection) -> None:
    """
    params: 客户端传递的参数，可能为 None
    conn: GatewayConnection 实例，用于发送响应
    """
    if not params:
        await conn.send_error("my_method", "invalid_params", "Missing params")
        return

    # 处理逻辑
    result = do_something(params)

    # 发送成功响应
    await conn.send_ok("my_method", {"result": result})

可用的 conn 方法

方法 用途
conn.send_ok(method, payload) 发送成功响应
conn.send_error(method, code, message) 发送错误响应
conn.send_event(event, payload) 发送推送事件

5. Agent Tools

Agent Tools 是 AI Agent 可以调用的工具：

from dataclasses import dataclass
from typing import Any

@dataclass
class MyTool:
    """自定义 Agent 工具"""

    name: str = "my_tool"
    description: str = "A custom tool for specific tasks"
    owner_only: bool = False

    def schema(self) -> dict:
        """返回 JSON Schema 定义"""
        return {
            "name": self.name,
            "description": self.description,
            "parameters": {
                "type": "object",
                "properties": {
                    "input": {
                        "type": "string",
                        "description": "Input to process"
                    }
                },
                "required": ["input"]
            }
        }

    async def execute(self, tool_call_id: str, params: dict[str, Any]) -> Any:
        """执行工具"""
        result = process(params.get("input", ""))
        return ToolResult(
            tool_call_id=tool_call_id,
            content=[{"type": "text", "text": result}],
            is_error=False
        )

6. 安装和测试

安装插件

pip install my-pyclaw-plugin

验证加载

pyclaw gateway status

插件会在 Gateway 启动时自动加载，日志中会显示：

INFO: Plugin loaded: my-plugin v0.1.0 (1 methods, 0 tools)

测试 RPC 方法

pyclaw gateway call my_plugin.hello --params '{}'

7. 最佳实践

错误处理

async def my_method(params: dict | None, conn: GatewayConnection) -> None:
    try:
        # 业务逻辑
        pass
    except ValidationError as e:
        await conn.send_error("my_method", "invalid_params", str(e))
    except Exception as e:
        logger.exception("Unexpected error in my_method")
        await conn.send_error("my_method", "internal_error", str(e))

参数验证

def validate_params(params: dict | None) -> tuple[dict | None, str | None]:
    """验证参数，返回 (validated_params, error_message)"""
    if not params:
        return None, "Missing params"

    required = params.get("required_field")
    if not required:
        return None, "Missing required_field"

    return params, None

资源清理

class MyPlugin:
    def __init__(self):
        self._resources = []

    async def cleanup(self) -> None:
        """清理资源"""
        for r in self._resources:
            await r.close()

8. 示例：天气插件

# src/weather_plugin/__init__.py
import httpx
from typing import Any

class WeatherPlugin:
    @property
    def name(self) -> str:
        return "weather-plugin"

    @property
    def version(self) -> str:
        return "1.0.0"

    @property
    def gateway_methods(self) -> dict:
        return {
            "weather.get": self._get_weather,
        }

    async def _get_weather(self, params: dict | None, conn: Any) -> None:
        if not params or "city" not in params:
            await conn.send_error("weather.get", "invalid_params", "Missing 'city'")
            return

        city = params["city"]
        try:
            async with httpx.AsyncClient() as client:
                resp = await client.get(f"https://wttr.in/{city}?format=j1")
                data = resp.json()

            await conn.send_ok("weather.get", {
                "city": city,
                "temp": data["current_condition"][0]["temp_C"],
                "condition": data["current_condition"][0]["weatherDesc"][0]["value"]
            })
        except Exception as e:
            await conn.send_error("weather.get", "api_error", str(e))


def load_plugin() -> WeatherPlugin:
    return WeatherPlugin()

9. 调试

启用调试日志

export PYCLAW_LOG_LEVEL=DEBUG
pyclaw gateway

查看已加载插件

pyclaw gateway status --deep

测试单个方法

pyclaw gateway call my_plugin.hello --params '{"name": "test"}' --json

10. 发布

构建包

python -m build

发布到 PyPI

twine upload dist/*

命名规范

  • 包名：pyclaw-<name>-plugin 或 openclaw-<name>
  • Entry point：使用简短、唯一的标识符

参考