picoclaw

Hook System Guide

This document describes the hook system that is implemented in the current repository, not the older design draft.

The current implementation supports two mounting modes:

In-process hooks
Out-of-process process hooks (JSON-RPC over stdio)

The repository no longer ships standalone example source files. The Go and Python examples below are embedded directly in this document. If you want to use them, copy them into your own local files first.

Supported Hook Types

Type	Interface	Stage	Can modify data
Observer	`RuntimeEventObserver`	Runtime event bus broadcast	No
LLM interceptor	`LLMInterceptor`	`before_llm` / `after_llm`	Yes
Tool interceptor	`ToolInterceptor`	`before_tool` / `after_tool`	Yes
Tool approver	`ToolApprover`	`approve_tool`	No, returns allow/deny

The currently exposed synchronous hook points are:

before_llm
after_llm
before_tool
after_tool
approve_tool

Everything else is exposed as read-only events.

Hook Actions

Hooks can return different actions to control the flow:

Action	Applicable Stages	Effect
`continue`	All interceptors	Pass through without modification
`modify`	`before_llm`, `after_llm`, `before_tool`, `after_tool`	Modify request/response and continue
`respond`	`before_tool`	Return a tool result directly, skip actual tool execution
`deny_tool`	`before_tool`	Deny tool execution, return error message
`abort_turn`	All interceptors	Abort the current turn
`hard_abort`	All interceptors	Force stop the entire agent loop

The `respond` Action

The respond action is special: it allows a before_tool hook to provide the tool result directly, skipping the actual tool execution. This is useful for:

Plugin tool injection: External hooks can implement tools without registering them in the tool registry
Tool result caching: Return cached results for repeated tool calls
Tool mocking: Return mock results for testing purposes

When a hook returns respond with a HookResult, the agent loop:

Skips the actual tool execution
Uses the provided result as if the tool had executed
Continues the turn normally with the result

Example (Go in-process hook):

func (h *MyHook) BeforeTool(
    ctx context.Context,
    call *agent.ToolCallHookRequest,
) (*agent.ToolCallHookRequest, agent.HookDecision, error) {
    if call.Tool == "my_plugin_tool" {
        next := call.Clone()
        next.HookResult = &tools.ToolResult{
            ForLLM:  "Plugin tool executed successfully",
            Silent:  false,
            IsError: false,
        }
        return next, agent.HookDecision{Action: agent.HookActionRespond}, nil
    }
    return call, agent.HookDecision{Action: agent.HookActionContinue}, nil
}

Example (Python process hook):

def handle_before_tool(params: dict) -> dict:
    tool = params.get("tool", "")
    if tool == "my_plugin_tool":
        return {
            "action": "respond",
            "result": {
                "for_llm": "Plugin tool executed successfully",
                "silent": False,
                "is_error": False
            }
        }
    return {"action": "continue"}

Execution Order

HookManager sorts hooks like this:

In-process hooks first
Process hooks second
Lower priority first within the same source
Name order as the final tie-breaker

Timeouts

Global defaults live under hooks.defaults:

observer_timeout_ms
interceptor_timeout_ms
approval_timeout_ms

Note: the current implementation does not support per-process-hook timeout_ms. Timeouts are global defaults.

Quick Start

If your first goal is simply to prove that the hook flow works and observe real requests, the easiest path is the Python process-hook example below:

Enable hooks.enabled
Save the Python example from this document to a local file, for example /tmp/review_gate.py
Set PICOCLAW_HOOK_LOG_FILE
Restart the gateway
Watch the log file with tail -f

Example:

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

Watch it with:

tail -f /tmp/picoclaw-hook-review-gate.log

If you are developing PicoClaw itself rather than only validating the protocol, continue with the Go in-process example as well.

What The Two Examples Are For

Go in-process example Best for validating the host-side hook chain and understanding MountHook() plus the synchronous stages
Python process example Best for understanding the JSON-RPC over stdio protocol and verifying the message flow between PicoClaw and an external process

Both examples are intentionally safe: they only log, never rewrite, and never deny.

Go In-Process Example

The following is a minimal logging hook for in-process use. It implements:

RuntimeEventObserver
LLMInterceptor
ToolInterceptor
ToolApprover

It only records activity. It does not rewrite requests or reject tools.

You can save it as your own Go file, for example pkg/myhooks/example_logger.go:

package myhooks

import (
	"context"
	"encoding/json"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/sipeed/picoclaw/pkg/agent"
	runtimeevents "github.com/sipeed/picoclaw/pkg/events"
	"github.com/sipeed/picoclaw/pkg/logger"
)

type ExampleLoggerHookOptions struct {
	LogFile   string `json:"log_file,omitempty"`
	LogEvents bool   `json:"log_events,omitempty"`
}

type ExampleLoggerHook struct {
	logFile   string
	logEvents bool
	mu        sync.Mutex
}

func NewExampleLoggerHook(opts ExampleLoggerHookOptions) *ExampleLoggerHook {
	return &ExampleLoggerHook{
		logFile:   strings.TrimSpace(opts.LogFile),
		logEvents: opts.LogEvents,
	}
}

func (h *ExampleLoggerHook) OnRuntimeEvent(ctx context.Context, evt runtimeevents.Event) error {
	_ = ctx
	if h == nil || !h.logEvents {
		return nil
	}
	h.record("event", evt.Scope, map[string]any{
		"event":   evt.Kind.String(),
		"payload": evt.Payload,
	}, nil)
	return nil
}

func (h *ExampleLoggerHook) BeforeLLM(
	ctx context.Context,
	req *agent.LLMHookRequest,
) (*agent.LLMHookRequest, agent.HookDecision, error) {
	_ = ctx
	h.record("before_llm", req.Meta, req, agent.HookDecision{Action: agent.HookActionContinue})
	return req, agent.HookDecision{Action: agent.HookActionContinue}, nil
}

func (h *ExampleLoggerHook) AfterLLM(
	ctx context.Context,
	resp *agent.LLMHookResponse,
) (*agent.LLMHookResponse, agent.HookDecision, error) {
	_ = ctx
	h.record("after_llm", resp.Meta, resp, agent.HookDecision{Action: agent.HookActionContinue})
	return resp, agent.HookDecision{Action: agent.HookActionContinue}, nil
}

func (h *ExampleLoggerHook) BeforeTool(
	ctx context.Context,
	call *agent.ToolCallHookRequest,
) (*agent.ToolCallHookRequest, agent.HookDecision, error) {
	_ = ctx
	h.record("before_tool", call.Meta, call, agent.HookDecision{Action: agent.HookActionContinue})
	return call, agent.HookDecision{Action: agent.HookActionContinue}, nil
}

func (h *ExampleLoggerHook) AfterTool(
	ctx context.Context,
	result *agent.ToolResultHookResponse,
) (*agent.ToolResultHookResponse, agent.HookDecision, error) {
	_ = ctx
	h.record("after_tool", result.Meta, result, agent.HookDecision{Action: agent.HookActionContinue})
	return result, agent.HookDecision{Action: agent.HookActionContinue}, nil
}

func (h *ExampleLoggerHook) ApproveTool(
	ctx context.Context,
	req *agent.ToolApprovalRequest,
) (agent.ApprovalDecision, error) {
	_ = ctx
	decision := agent.ApprovalDecision{Approved: true}
	h.record("approve_tool", req.Meta, req, decision)
	return decision, nil
}

func (h *ExampleLoggerHook) record(stage string, refs any, payload any, decision any) {
	logger.InfoCF("hooks", "Example hook observed", map[string]any{
		"stage": stage,
	})
	if h == nil || h.logFile == "" {
		return
	}

	entry := map[string]any{
		"ts":       time.Now().UTC(),
		"stage":    stage,
		"refs":     refs,
		"payload":  payload,
		"decision": decision,
	}

	body, err := json.Marshal(entry)
	if err != nil {
		logger.WarnCF("hooks", "Example hook log encode failed", map[string]any{
			"stage": stage,
			"error": err.Error(),
		})
		return
	}

	h.mu.Lock()
	defer h.mu.Unlock()

	if dir := filepath.Dir(h.logFile); dir != "" && dir != "." {
		if err := os.MkdirAll(dir, 0o755); err != nil {
			logger.WarnCF("hooks", "Example hook log mkdir failed", map[string]any{
				"stage": stage,
				"path":  h.logFile,
				"error": err.Error(),
			})
			return
		}
	}

	file, err := os.OpenFile(h.logFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
	if err != nil {
		logger.WarnCF("hooks", "Example hook log open failed", map[string]any{
			"stage": stage,
			"path":  h.logFile,
			"error": err.Error(),
		})
		return
	}
	defer func() { _ = file.Close() }()

	if _, err := file.Write(append(body, '\n')); err != nil {
		logger.WarnCF("hooks", "Example hook log write failed", map[string]any{
			"stage": stage,
			"path":  h.logFile,
			"error": err.Error(),
		})
	}
}

Mounting It In Code

If code mounting is enough, call this after AgentLoop is initialized:

hook := myhooks.NewExampleLoggerHook(myhooks.ExampleLoggerHookOptions{
    LogFile:   "/tmp/picoclaw-hook-example-logger.log",
    LogEvents: true,
})

if err := al.MountHook(agent.NamedHook("example-logger", hook)); err != nil {
    panic(err)
}

If You Also Want Config Mounting

The hook system supports builtin hooks, but that requires you to compile the factory into your binary. In practice, that means you need registration code like this alongside the hook definition above:

package myhooks

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/sipeed/picoclaw/pkg/agent"
	"github.com/sipeed/picoclaw/pkg/config"
)

func init() {
	if err := agent.RegisterBuiltinHook("example_logger", func(
		ctx context.Context,
		spec config.BuiltinHookConfig,
	) (any, error) {
		_ = ctx

		var opts ExampleLoggerHookOptions
		if len(spec.Config) > 0 {
			if err := json.Unmarshal(spec.Config, &opts); err != nil {
				return nil, fmt.Errorf("decode example_logger config: %w", err)
			}
		}
		return NewExampleLoggerHook(opts), nil
	}); err != nil {
		panic(err)
	}
}

Only after you register that builtin will the following config work:

{
  "hooks": {
    "enabled": true,
    "builtins": {
      "example_logger": {
        "enabled": true,
        "priority": 10,
        "config": {
          "log_file": "/tmp/picoclaw-hook-example-logger.log",
          "log_events": true
        }
      }
    }
  }
}

How To Observe It

If log_file is set, each hook call is appended as JSON Lines
If log_file is not set, the hook still writes summaries to the gateway log
Requests that only hit the LLM path usually show before_llm and after_llm
Requests that trigger tools usually also show before_tool, approve_tool, and after_tool
If log_events=true, you will also see event

Typical log lines:

{"ts":"2026-03-21T14:10:00Z","stage":"before_tool","meta":{"session_key":"session-1"},"payload":{"tool":"echo_text","arguments":{"text":"hello"}},"decision":{"action":"continue"}}
{"ts":"2026-03-21T14:10:00Z","stage":"approve_tool","meta":{"session_key":"session-1"},"payload":{"tool":"echo_text","arguments":{"text":"hello"}},"decision":{"approved":true}}

If you only see before_llm and after_llm, that usually means the request did not trigger any tool call, not that the hook failed to mount.

Python Process-Hook Example

The following script is a minimal process-hook example. It uses only the Python standard library and supports:

hook.hello
hook.runtime_event
hook.before_tool
hook.approve_tool

It only records activity. It does not rewrite or deny anything.

Save it to any local path, for example /tmp/review_gate.py:

#!/usr/bin/env python3
from __future__ import annotations

import json
import os
import signal
import sys
from datetime import datetime, timezone
from typing import Any

LOG_EVENTS = os.getenv("PICOCLAW_HOOK_LOG_EVENTS", "1").lower() not in {"0", "false", "no"}
LOG_FILE = os.getenv("PICOCLAW_HOOK_LOG_FILE", "").strip()


def append_log(entry: dict[str, Any]) -> None:
    if not LOG_FILE:
        return

    payload = {
        "ts": datetime.now(timezone.utc).isoformat(),
        **entry,
    }
    try:
        log_dir = os.path.dirname(LOG_FILE)
        if log_dir:
            os.makedirs(log_dir, exist_ok=True)
        with open(LOG_FILE, "a", encoding="utf-8") as handle:
            handle.write(json.dumps(payload, ensure_ascii=True) + "\n")
    except OSError as exc:
        log_stderr(f"failed to write hook log file {LOG_FILE}: {exc}")


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 {}

    append_log({
        "direction": "out",
        "id": message_id,
        "response": payload.get("result"),
        "error": payload.get("error"),
    })

    try:
        sys.stdout.write(json.dumps(payload, ensure_ascii=True) + "\n")
        sys.stdout.flush()
    except BrokenPipeError:
        raise SystemExit(0) from None


def log_stderr(message: str) -> None:
    try:
        sys.stderr.write(message + "\n")
        sys.stderr.flush()
    except BrokenPipeError:
        raise SystemExit(0) from None


def handle_shutdown_signal(signum: int, _frame: Any) -> None:
    raise KeyboardInterrupt(f"received signal {signum}")


def handle_before_tool(params: dict[str, Any]) -> dict[str, Any]:
    _ = params
    return {"action": "continue"}


def handle_approve_tool(params: dict[str, Any]) -> dict[str, Any]:
    _ = params
    return {"approved": True}


def handle_request(method: str, params: dict[str, Any]) -> dict[str, Any]:
    if method == "hook.hello":
        return {"ok": True, "name": "python-review-gate"}
    if method == "hook.before_tool":
        return handle_before_tool(params)
    if method == "hook.approve_tool":
        return handle_approve_tool(params)
    if method == "hook.before_llm":
        return {"action": "continue"}
    if method == "hook.after_llm":
        return {"action": "continue"}
    if method == "hook.after_tool":
        return {"action": "continue"}
    raise KeyError(f"method not found: {method}")


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

            try:
                message = json.loads(line)
            except json.JSONDecodeError as exc:
                log_stderr(f"failed to decode request: {exc}")
                append_log({
                    "direction": "in",
                    "decode_error": str(exc),
                    "raw": line,
                })
                continue

            method = message.get("method")
            message_id = message.get("id", 0)
            params = message.get("params") or {}
            if not isinstance(params, dict):
                params = {}

            append_log({
                "direction": "in",
                "id": message_id,
                "method": method,
                "params": params,
                "notification": not bool(message_id),
            })

            if not message_id:
                if method == "hook.runtime_event" and LOG_EVENTS:
                    log_stderr(f"observed event: {params.get('kind')}")
                continue

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

            send_response(int(message_id), result=result)
    except KeyboardInterrupt:
        return 0

    return 0


if __name__ == "__main__":
    signal.signal(signal.SIGINT, handle_shutdown_signal)
    signal.signal(signal.SIGTERM, handle_shutdown_signal)
    raise SystemExit(main())

Configuration

{
  "hooks": {
    "enabled": true,
    "processes": {
      "py_review_gate": {
        "enabled": true,
        "priority": 100,
        "transport": "stdio",
        "command": [
          "python3",
          "/abs/path/to/review_gate.py"
        ],
        "observe": [
          "agent.tool.exec_start",
          "agent.tool.exec_end",
          "agent.tool.exec_skipped"
        ],
        "intercept": [
          "before_tool",
          "approve_tool"
        ],
        "env": {
          "PICOCLAW_HOOK_LOG_FILE": "/tmp/picoclaw-hook-review-gate.log"
        }
      }
    }
  }
}

Environment Variables

PICOCLAW_HOOK_LOG_EVENTS Whether to write hook.runtime_event summaries to stderr, enabled by default
PICOCLAW_HOOK_LOG_FILE Path to an external log file. When set, the script appends inbound hook requests, notifications, and outbound responses as JSON Lines

Note: PICOCLAW_HOOK_LOG_FILE has no default. If you do not set it, the script does not write any file logs.

How To Confirm It Received Hooks

Watch two places:

Gateway logs Useful for confirming that the host successfully started the process and for seeing event summaries written to stderr
PICOCLAW_HOOK_LOG_FILE Useful for seeing the exact requests the script received and the exact responses it returned

Typical interpretation:

Only hook.hello The process started and completed the handshake, but no business hook request has arrived yet
hook.runtime_event The observe configuration is working
hook.before_tool The intercept: ["before_tool", ...] configuration is working
hook.approve_tool The approval hook path is working

Because this example never rewrites or denies, the expected responses look like:

{"direction":"out","id":7,"response":{"action":"continue"},"error":null}
{"direction":"out","id":8,"response":{"approved":true},"error":null}

A complete sample:

{"ts":"2026-03-21T14:12:00+00:00","direction":"in","id":1,"method":"hook.hello","params":{"name":"py_review_gate","version":1,"modes":["observe","tool","approve"]},"notification":false}
{"ts":"2026-03-21T14:12:00+00:00","direction":"out","id":1,"response":{"ok":true,"name":"python-review-gate"},"error":null}
{"ts":"2026-03-21T14:12:05+00:00","direction":"in","id":0,"method":"hook.runtime_event","params":{"kind":"agent.tool.exec_start"},"notification":true}
{"ts":"2026-03-21T14:12:05+00:00","direction":"in","id":7,"method":"hook.before_tool","params":{"tool":"echo_text","arguments":{"text":"hello"}},"notification":false}
{"ts":"2026-03-21T14:12:05+00:00","direction":"out","id":7,"response":{"action":"continue"},"error":null}

Additional notes:

Timestamps are UTC
notification=true means it was a notification such as hook.runtime_event, which does not expect a response
id increases within a single hook process; if the process restarts, the counter starts over

Process-Hook Protocol

Current process hooks use JSON-RPC over stdio:

PicoClaw starts the external process
Requests and responses are exchanged as one JSON message per line
hook.runtime_event is a notification and does not need a response
hook.before_llm, hook.after_llm, hook.before_tool, hook.after_tool, and hook.approve_tool are request/response calls

The host does not currently accept new RPCs initiated by the process hook. In practice, that means an external hook can only respond to PicoClaw calls; it cannot call back into the host to send channel messages.

Configuration Fields

`hooks.builtins.<name>`

enabled
priority
config

`hooks.processes.<name>`

enabled
priority
transport Currently only stdio is supported
command
dir
env
observe
intercept

Troubleshooting

If a hook looks like it is not firing, check these in order:

hooks.enabled
Whether the target builtin or process hook is enabled
Whether the process-hook command path is correct
Whether you are watching the correct log file
Whether the current request actually reached the stage you care about
Whether observe or intercept contains the hook point you want

A practical minimal troubleshooting pair is:

Use the Python process-hook example from this document to validate the external protocol
Use the Go in-process example from this document to validate the host-side chain

If the Python side shows hook.hello but no business hook requests, the protocol is usually fine; the current request simply did not trigger the stage you expected.

Scope And Limits

The current hook system is best suited for:

LLM request rewriting
Tool argument normalization
Pre-execution tool approval
Auditing and observability

It is not yet well suited for:

External hooks actively sending channel messages
Suspending a turn and waiting for human approval replies
Full inbound/outbound message interception across the whole platform

If you want a real human approval workflow, use hooks as the approval entry point and keep the state machine plus channel interaction in a separate ApprovalManager.

This site is open source. Improve this page.