Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051"""The model's side of hooks: Python callables, and the inner-tool gate."""
import albedo_hooksfrom albedo_api import PythonApi
class Hooks: """Python hooks: around your own tool calls (the `python` tool, and inside cells `run` as `Bash` and `files.read/edit/write` as `Read`, `Edit`, `Write`), on each user prompt, and when a turn is about to end."""
def register(self, event: str, callback, matcher: str | None = None): """Call `callback(input)` around matching tool calls. `event` is "PreToolUse", "PostToolUse", "UserPromptSubmit" or "Stop"; `matcher` is a regex on the tool name (None matches all; tool events only). `input` is a dict like a command hook's stdin: hook_event_name, then tool_name, tool_input (and tool_response after) for tool events, prompt for UserPromptSubmit, last_assistant_message and stop_hook_active for Stop. "Stop" with a block keeps the turn going. Return None, or a dict in the Claude Code hook output format, e.g. {"decision": "block", "reason": "..."}, {"hookSpecificOutput": {"permissionDecision": "deny", "permissionDecisionReason": "..."}}, {"hookSpecificOutput": {"permissionDecision": "allow", "updatedInput": {...}}}, or {"hookSpecificOutput": {"additionalContext": "..."}}. A blocked call raises PermissionError. Callbacks run before command hooks, sync or async, and fail open if they raise. They live and die with this kernel. Answers a function that removes the callback.""" return albedo_hooks.register(event, callback, matcher)
def registered(self) -> list[tuple[str, str | None]]: """The (event, matcher) of every callback registered here.""" return [(event, matcher) for event, matcher, _ in albedo_hooks.registered()]
def clear(self) -> None: """Remove every callback registered here.""" albedo_hooks.clear()
def __repr__(self) -> str: return f"hooks({len(albedo_hooks.registered())} callbacks)"
def setup(api: PythonApi) -> dict[str, object]: albedo_hooks.install(api.host_sync)
def shutdown() -> None: albedo_hooks.install(None) albedo_hooks.clear()
api.on_shutdown(shutdown) return {"hooks": Hooks()}