Hook 系统

PicoClaw 提供了一套 Hook 系统，允许你观察事件、拦截 LLM 和工具调用、以及通过审批逻辑来控制工具执行——无需修改核心代码。

Hook 类型

类型	接口	阶段	是否可修改数据
Observer（观察者）	EventObserver	EventBus 广播	否
LLM interceptor（LLM 拦截器）	LLMInterceptor	before_llm / after_llm	是
Tool interceptor（工具拦截器）	ToolInterceptor	before_tool / after_tool	是
Tool approver（工具审批者）	ToolApprover	approve_tool	否，返回允许/拒绝

Hook 触发点

before_llm — 在每次 LLM 请求之前触发。拦截器可以改写请求内容。
after_llm — 在 LLM 响应之后触发。拦截器可以改写响应内容。
before_tool — 在工具执行之前触发。拦截器可以改写参数。
after_tool — 在工具执行之后触发。拦截器可以改写结果。
approve_tool — 在工具执行之前（before_tool 之后）触发。审批者返回允许或拒绝。

执行顺序

进程内 Hook（in-process）优先执行。
外部进程 Hook（process hooks）随后执行。
在同一组内，按 priority（优先级）排序，数值越小越先执行。
如果两个 Hook 优先级相同，则按名称（字典序）排序。

超时设置

全局默认值在 hooks.defaults 下配置：

字段	说明
`observer_timeout_ms`	观察者回调的最大执行时间，超时后将被取消。
`interceptor_timeout_ms`	拦截器的最大执行时间，超时后将被取消。
`approval_timeout_ms`	审批者的最大执行时间，超时后工具调用将被默认拒绝。

快速开始

在 PicoClaw 配置中添加以下内容即可启用一个 Python 外部进程 Hook：

{
  "hooks": {
    "enabled": true,
    "processes": {
      "py_review_gate": {
        "enabled": true,
        "priority": 100,
        "transport": "stdio",
        "command": ["python3", "/tmp/review_gate.py"],
        "observe": ["tool_exec_start", "tool_exec_end", "tool_exec_skipped"],
        "intercept": ["before_tool", "approve_tool"],
        "env": {
          "PICOCLAW_HOOK_LOG_FILE": "/tmp/picoclaw-hook-review-gate.log"
        }
      }
    }
  }
}

Go 进程内示例

直接在 Go 中注册 Hook：

package main

import (
    "context"
    "log"

    "github.com/anthropics/picoclaw/hook"
)

type auditHook struct{}

func (h *auditHook) Name() string { return "audit" }

func (h *auditHook) BeforeTool(ctx context.Context, req *hook.ToolRequest) (*hook.ToolRequest, error) {
    log.Printf("tool=%s args=%v", req.Name, req.Args)
    return req, nil // pass through unmodified
}

func init() {
    hook.Register(&auditHook{})
}

Python 外部进程 Hook 示例

以下 review_gate.py 实现了一个外部进程 Hook，用于观察工具事件、参与 before_tool 拦截和 approve_tool 审批。它仅记录日志，不会改写参数或拒绝执行。

#!/usr/bin/env python3
"""review_gate.py – PicoClaw process-hook (JSON-RPC over stdio).

Supports:
  hook.hello      – handshake
  hook.event      – observe events (log only)
  hook.before_tool – intercept before tool execution (pass-through)
  hook.approve_tool – approve tool execution (always allow)
"""

import json
import os
import sys

LOG_FILE = os.environ.get("PICOCLAW_HOOK_LOG_FILE", "/tmp/picoclaw-hook-review-gate.log")


def _log(msg: str) -> None:
    with open(LOG_FILE, "a") as f:
        f.write(msg + "\n")


def _respond(id: int | str | None, result: dict) -> None:
    payload = {"jsonrpc": "2.0", "id": id, "result": result}
    line = json.dumps(payload)
    sys.stdout.write(line + "\n")
    sys.stdout.flush()


def handle_hello(id, params):
    _log(f"hello: protocol_version={params.get('protocol_version')}")
    _respond(id, {"name": "py_review_gate", "protocol_version": 1})


def handle_event(id, params):
    _log(f"event: {params.get('type')} — {json.dumps(params.get('data', {}))}")
    _respond(id, {})


def handle_before_tool(id, params):
    tool = params.get("name", "<unknown>")
    _log(f"before_tool: {tool}")
    # Pass through unmodified
    _respond(id, {"args": params.get("args", {})})


def handle_approve_tool(id, params):
    tool = params.get("name", "<unknown>")
    _log(f"approve_tool: {tool} → allow")
    _respond(id, {"allow": True})


DISPATCH = {
    "hook.hello": handle_hello,
    "hook.event": handle_event,
    "hook.before_tool": handle_before_tool,
    "hook.approve_tool": handle_approve_tool,
}


def main() -> None:
    _log("review_gate started")
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except json.JSONDecodeError:
            _log(f"bad json: {line}")
            continue

        method = msg.get("method", "")
        handler = DISPATCH.get(method)
        if handler:
            handler(msg.get("id"), msg.get("params", {}))
        else:
            _log(f"unknown method: {method}")
    _log("review_gate exiting")


if __name__ == "__main__":
    main()

外部进程 Hook 协议

外部进程 Hook 通过 stdio 上的 JSON-RPC 2.0 协议与 PicoClaw 通信（每行一个 JSON 对象）。

PicoClaw 启动进程后发送 hook.hello，参数为 {"protocol_version": 1}。
进程必须回复 {"name": "<hook_name>", "protocol_version": 1}。
随后 PicoClaw 会根据配置发送 hook.event、hook.before_tool、hook.after_tool 或 hook.approve_tool 消息。
进程需要为每个请求回复一个 JSON-RPC 响应。

从 PicoClaw 的角度来看，所有通信都是同步的：它发送请求后等待恰好一个响应（受配置的超时时间限制）。

配置参考

内置 Hook — `hooks.builtins.<name>`

字段	类型	说明
`enabled`	bool	是否启用此内置 Hook。
`priority`	int	执行顺序（数值越小越先执行）。
`config`	object	传递给内置 Hook 的特定配置。

外部进程 Hook — `hooks.processes.<name>`

字段	类型	说明
`enabled`	bool	是否启用此外部进程 Hook。
`priority`	int	执行顺序（数值越小越先执行）。
`transport`	string	传输协议。目前仅支持 `"stdio"`。
`command`	string[]	启动进程的命令及参数。
`dir`	string	进程的工作目录。
`env`	object	传递给进程的额外环境变量。
`observe`	string[]	此 Hook 要观察的事件类型列表。
`intercept`	string[]	此 Hook 要拦截的 Hook 触发点列表。

适用范围与限制

Hook 系统最适合以下场景：

LLM 请求改写 — 规范化提示词、注入系统上下文、执行策略。
工具参数规范化 — 在执行前清理或转换参数。
执行前工具审批 — 使用自定义逻辑控制危险操作。
审计 — 记录所有 LLM 和工具活动，用于合规或调试。

尚不支持的功能：

外部 Hook 向对话中发送消息。
暂停执行以等待人工审批。
完整的消息级拦截（目前仅支持 LLM 请求/响应和工具调用/结果的拦截）。

故障排查

Hook 未触发 — 确认 enabled: true，并检查事件类型或 Hook 触发点是否已列在 observe 或 intercept 中。
超时错误 — 增大 hooks.defaults 中对应的超时值。检查外部进程 Hook 是否在每次响应后刷新了 stdout。
进程 Hook 启动时崩溃 — 手动运行该命令，检查是否缺少依赖或存在语法错误。
JSON 解析错误 — 确保每行恰好一个 JSON 对象，stdout 上不要有多余输出（调试信息请使用 stderr 或日志文件）。

Hook 类型​

Hook 触发点​

执行顺序​

超时设置​

快速开始​

Go 进程内示例​

Python 外部进程 Hook 示例​

外部进程 Hook 协议​

配置参考​

内置 Hook — hooks.builtins.<name>​

外部进程 Hook — hooks.processes.<name>​

适用范围与限制​

故障排查​