picoclaw/docs/architecture/hooks/plugin-tool-injection.zh.md

插件工具注入示例

本文档展示如何利用 PicoClaw 的 hook 系统实现外部插件工具注入，让 LLM 能调用由外部 hook 进程实现的工具。

---

核心原理

通过 hook 系统的 respond action，外部 hook 可以：

1. 在 before_llm 中注入工具定义，让 LLM 知道有这个工具可用
2. 在 before_tool 中使用 respond action 直接返回工具执行结果，跳过 ToolRegistry

这样，外部 hook 可以完全实现插件工具，无需在 PicoClaw 内部注册任何工具。

---

完整示例：天气查询插件

下面是一个完整的 Python hook 示例，实现一个天气查询插件工具。

1. Hook 脚本实现

保存为 /tmp/weather_plugin.py：

#!/usr/bin/env python3
"""天气查询插件 hook 示例"""
from __future__ import annotations

import json
import sys
import signal
from typing import Any

# 模拟天气数据
WEATHER_DATA = {
    "北京": {"temp": 15, "weather": "晴", "humidity": 45},
    "上海": {"temp": 18, "weather": "多云", "humidity": 60},
    "广州": {"temp": 25, "weather": "晴", "humidity": 70},
    "深圳": {"temp": 26, "weather": "多云", "humidity": 75},
}


def get_weather(city: str) -> dict:
    """获取天气数据（模拟）"""
    data = WEATHER_DATA.get(city)
    if data:
        return {
            "for_llm": f"{city}天气：{data['weather']}，温度{data['temp']}°C，湿度{data['humidity']}%",
            "for_user": "",
            "silent": False,
            "is_error": False,
        }
    return {
        "for_llm": f"未找到城市 {city} 的天气数据",
        "for_user": "",
        "silent": False,
        "is_error": True,
    }


def handle_hello(params: dict) -> dict:
    return {"ok": True, "name": "weather-plugin"}


def handle_before_llm(params: dict) -> dict:
    """注入天气查询工具定义"""
    tools = params.get("tools", [])

    # 添加天气查询工具
    tools.append({
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如：北京、上海、广州"
                    }
                },
                "required": ["city"]
            }
        }
    })

    return {
        "action": "modify",
        "request": {
            "model": params.get("model"),
            "messages": params.get("messages", []),
            "tools": tools,
            "options": params.get("options", {}),
        }
    }


def handle_before_tool(params: dict) -> dict:
    """处理工具调用，直接返回结果"""
    tool = params.get("tool", "")
    args = params.get("arguments", {})

    if tool == "get_weather":
        city = args.get("city", "")
        result = get_weather(city)

        # 使用 respond action 直接返回结果，跳过 ToolRegistry
        return {
            "action": "respond",
            "result": result,
        }

    # 其他工具继续正常流程
    return {"action": "continue"}


def handle_request(method: str, params: dict) -> dict:
    if method == "hook.hello":
        return handle_hello(params)
    if method == "hook.before_llm":
        return handle_before_llm(params)
    if method == "hook.before_tool":
        return handle_before_tool(params)
    if method == "hook.after_llm":
        return {"action": "continue"}
    if method == "hook.after_tool":
        return {"action": "continue"}
    if method == "hook.approve_tool":
        return {"approved": True}
    raise KeyError(f"method not found: {method}")


def send_response(message_id: int, result: Any | None = None, error: str | None = None) -> None:
    payload: dict[str, Any] = {
        "jsonrpc": "2.0",
        "id": message_id,
    }
    if error is not None:
        payload["error"] = {"code": -32000, "message": error}
    else:
        payload["result"] = result if result is not None else {}

    sys.stdout.write(json.dumps(payload, ensure_ascii=True) + "\n")
    sys.stdout.flush()


def main() -> int:
    for raw_line in sys.stdin:
        line = raw_line.strip()
        if not line:
            continue

        try:
            message = json.loads(line)
        except json.JSONDecodeError:
            continue

        method = message.get("method")
        message_id = message.get("id", 0)
        params = message.get("params") or {}

        if not message_id:
            continue

        try:
            result = handle_request(str(method or ""), params)
            send_response(int(message_id), result=result)
        except KeyError as exc:
            send_response(int(message_id), error=str(exc))
        except Exception as exc:
            send_response(int(message_id), error=f"unexpected error: {exc}")

    return 0


if __name__ == "__main__":
    signal.signal(signal.SIGINT, lambda *_: raise SystemExit(0))
    signal.signal(signal.SIGTERM, lambda *_: raise SystemExit(0))
    raise SystemExit(main())

2. 配置 PicoClaw

在配置文件中添加 hook 配置：

{
  "hooks": {
    "enabled": true,
    "processes": {
      "weather_plugin": {
        "enabled": true,
        "priority": 100,
        "transport": "stdio",
        "command": ["python3", "/tmp/weather_plugin.py"],
        "intercept": ["before_llm", "before_tool"]
      }
    }
  }
}

3. 测试效果

当用户问"北京今天天气怎么样？"时：

1. PicoClaw 发送 hook.before_llm，hook 注入 get_weather 工具定义
2. LLM 看到工具定义，决定调用 get_weather(city="北京")
3. PicoClaw 发送 hook.before_tool，hook 使用 respond action 返回天气数据
4. LLM 收到结果，回复用户"北京今天晴天，温度15°C"

---

流程图解

用户: "北京今天天气怎么样？"
        ↓
    PicoClaw
        ↓
    hook.before_llm
        ↓ (注入 get_weather 工具定义)
    LLM 请求
        ↓
    LLM 决定调用 get_weather(city="北京")
        ↓
    hook.before_tool
        ↓ (respond action 返回天气数据)
    直接返回结果给 LLM
        ↓ (跳过 ToolRegistry)
    LLM 回复: "北京今天晴天，温度15°C"

---

关键点说明

before_llm 注入工具定义

工具定义遵循 OpenAI function calling 格式：

{
  "type": "function",
  "function": {
    "name": "工具名称",
    "description": "工具描述",
    "parameters": {
      "type": "object",
      "properties": {
        "参数名": {
          "type": "string",
          "description": "参数描述"
        }
      },
      "required": ["必需参数列表"]
    }
  }
}

before_tool 使用 respond action

respond action 的响应格式：

{
  "action": "respond",
  "result": {
    "for_llm": "返回给 LLM 的内容",
    "for_user": "可选，发送给用户的内容",
    "silent": false,
    "is_error": false,
    "media": ["可选，媒体引用列表"],
    "response_handled": false
  }
}

字段	说明
for_llm	必须，LLM 会看到这个内容
for_user	可选，直接发送给用户
silent	为 true 时不发送给用户
is_error	为 true 时表示执行失败
media	可选，媒体文件引用列表（如图片、文件）
response_handled	为 true 时表示已处理用户请求，轮次将结束

---

媒体文件处理

respond action 支持返回媒体文件（图片、文件等）。有两种处理方式：

1. 自动发送（response_handled=true）

当 response_handled=true 时，媒体文件会自动发送给用户，轮次结束：

{
  "action": "respond",
  "result": {
    "for_llm": "图片已发送给用户",
    "for_user": "",
    "media": ["media://abc123"],
    "response_handled": true
  }
}

适用场景：

• 图像生成插件直接返回结果
• 文件下载插件发送文件给用户

2. LLM 可见（response_handled=false）

当 response_handled=false 时，媒体引用会传递给 LLM，LLM 可以在下一轮请求中看到内容：

{
  "action": "respond",
  "result": {
    "for_llm": "图片已加载，路径：/tmp/image.png [file:/tmp/image.png]",
    "media": ["media://abc123"]
  }
}

LLM 看到内容后，可以自主决定：

• 使用 send_file 工具发送给用户
• 分析图片内容并回复用户
• 其他处理方式

媒体引用格式

媒体引用使用 media:// 协议：

media://<store-id>

这些引用由 PicoClaw 的 MediaStore 管理，可以：

• 通过 channel 发送给用户
• 在 LLM vision 请求中转换为 base64

替代方案：使用现有工具

如果插件生成文件，可以返回文件路径让 LLM 调用 send_file 等工具：

{
  "action": "respond",
  "result": {
    "for_llm": "图片已生成，保存在 /tmp/generated_image.png。使用 send_file 工具发送给用户。",
    "for_user": "",
    "silent": false
  }
}

这种方式：

• 更解耦，LLM 自主决策发送时机
• 利用现有工具机制
• 支持批量发送、延迟发送等场景

---

多工具注入示例

可以同时注入多个工具：

def handle_before_llm(params: dict) -> dict:
    tools = params.get("tools", [])

    # 工具1：天气查询
    tools.append({
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询城市天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                },
                "required": ["city"]
            }
        }
    })

    # 工具2：计算器
    tools.append({
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学计算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "数学表达式"}
                },
                "required": ["expression"]
            }
        }
    })

    return {
        "action": "modify",
        "request": {
            "model": params.get("model"),
            "messages": params.get("messages", []),
            "tools": tools,
            "options": params.get("options", {}),
        }
    }


def handle_before_tool(params: dict) -> dict:
    tool = params.get("tool", "")
    args = params.get("arguments", {})

    if tool == "get_weather":
        return {
            "action": "respond",
            "result": get_weather(args.get("city", "")),
        }

    if tool == "calculate":
        # 简单计算示例
        try:
            expr = args.get("expression", "")
            result = eval(expr)  # 注意：实际使用时需要安全处理
            return {
                "action": "respond",
                "result": {
                    "for_llm": f"计算结果: {result}",
                    "silent": False,
                    "is_error": False,
                },
            }
        except Exception as e:
            return {
                "action": "respond",
                "result": {
                    "for_llm": f"计算错误: {e}",
                    "silent": False,
                    "is_error": True,
                },
            }

    return {"action": "continue"}

---

与内置工具共存

注入的插件工具与 PicoClaw 内置工具共存：

• 内置工具（如 bash、read_file）正常通过 ToolRegistry 执行
• 插件工具通过 hook 的 respond action 返回结果
• handle_before_tool 中只处理插件工具，其他工具返回 continue

---

Go 进程内 Hook 示例

如果需要在 Go 代码中实现插件工具注入：

package myhooks

import (
    "context"
    "github.com/sipeed/picoclaw/pkg/agent"
    "github.com/sipeed/picoclaw/pkg/tools"
)

type WeatherPluginHook struct{}

func (h *WeatherPluginHook) BeforeLLM(
    ctx context.Context,
    req *agent.LLMHookRequest,
) (*agent.LLMHookRequest, agent.HookDecision, error) {
    // 注入工具定义
    req.Tools = append(req.Tools, agent.ToolDefinition{
        Type: "function",
        Function: agent.FunctionDefinition{
            Name:        "get_weather",
            Description: "查询城市天气",
            Parameters: map[string]any{
                "type": "object",
                "properties": map[string]any{
                    "city": map[string]any{
                        "type":        "string",
                        "description": "城市名称",
                    },
                },
                "required": []string{"city"},
            },
        },
    })

    return req, agent.HookDecision{Action: agent.HookActionContinue}, nil
}

func (h *WeatherPluginHook) BeforeTool(
    ctx context.Context,
    call *agent.ToolCallHookRequest,
) (*agent.ToolCallHookRequest, agent.HookDecision, error) {
    if call.Tool == "get_weather" {
        city := call.Arguments["city"].(string)

        // 设置 HookResult，使用 respond action
        next := call.Clone()
        next.HookResult = &tools.ToolResult{
            ForLLM:  getWeatherData(city),
            Silent:  false,
            IsError: false,
        }

        return next, agent.HookDecision{Action: agent.HookActionRespond}, nil
    }

    return call, agent.HookDecision{Action: agent.HookActionContinue}, nil
}

func getWeatherData(city string) string {
    // 实现天气查询逻辑
    return fmt.Sprintf("%s天气：晴，温度20°C", city)
}

---

总结

通过 hook 系统的 respond action，外部进程可以：

1. 注入工具定义：让 LLM 知道有新工具可用
2. 提供工具实现：直接返回执行结果，无需注册到 ToolRegistry
3. 与内置工具共存：不影响 PicoClaw 原有工具的正常运行

这为插件开发提供了灵活、优雅的解决方案。

---

安全边界说明

绕过审批检查

重要：respond action 会绕过 ApproveTool 审批检查。

这意味着：

• before_tool hook 可以为任何工具名称返回 respond，包括敏感工具（如 bash）
• 工具不会经过审批流程，直接返回 hook 提供的结果
• 这是为了支持插件工具而设计，但也带来了安全风险

安全建议

1. 审查 hook 配置：确保只有可信的 hook 进程被启用
2. 限制 hook 权限：在 hook 实现中添加自己的安全检查
3. 优先使用 deny_tool：对于拒绝执行，使用 deny_tool action 而非 respond 返回错误

示例：hook 内置安全检查

def handle_before_tool(params: dict) -> dict:
    tool = params.get("tool", "")
    args = params.get("arguments", {})

    # 安全检查：只处理插件工具
    if tool in ["get_weather", "calculate"]:
        return {
            "action": "respond",
            "result": execute_plugin_tool(tool, args),
        }

    # 其他工具继续正常流程（会经过审批）
    return {"action": "continue"}

这样可以确保 hook 只影响插件工具，不影响系统工具的审批流程。