Hooks#
Note
This is a new feature added in response to issue #156.
The hook system allows tools and plugins to register callbacks that execute at various points in gptme’s lifecycle. This enables powerful extensions like automatic linting, memory management, pre-commit checks, and more.
Hook Types#
The following hook types are available:
Message Lifecycle Hooks#
MESSAGE_PRE_PROCESS
: Before processing a user messageMESSAGE_POST_PROCESS
: After message processing completesMESSAGE_TRANSFORM
: Transform message content before processing
Tool Lifecycle Hooks#
TOOL_PRE_EXECUTE
: Before executing any toolTOOL_POST_EXECUTE
: After executing any toolTOOL_TRANSFORM
: Transform tool execution
File Operation Hooks#
FILE_PRE_SAVE
: Before saving a fileFILE_POST_SAVE
: After saving a fileFILE_PRE_PATCH
: Before patching a fileFILE_POST_PATCH
: After patching a file
Session Lifecycle Hooks#
SESSION_START
: At session startSESSION_END
: At session end
Generation Hooks#
GENERATION_PRE
: Before generating responseGENERATION_POST
: After generating responseGENERATION_INTERRUPT
: Interrupt generation
Usage#
Registering Hooks from Tools#
Tools can register hooks in their ToolSpec
definition:
from gptme.tools.base import ToolSpec
from gptme.hooks import HookType
from gptme.message import Message
def on_file_save(path, content, created):
"""Hook function called after a file is saved."""
if path.suffix == ".py":
# Run linting on Python files
return Message("system", f"Linted {path}")
return None
tool = ToolSpec(
name="linter",
desc="Automatic linting tool",
hooks={
"file_save": (
HookType.FILE_POST_SAVE.value, # Hook type
on_file_save, # Hook function
10 # Priority (higher = runs first)
)
}
)
Registering Hooks Programmatically#
You can also register hooks directly:
from gptme.hooks import register_hook, HookType
def my_hook_function(log, workspace):
"""Custom hook function."""
# Do something
return Message("system", "Hook executed!")
register_hook(
name="my_custom_hook",
hook_type=HookType.MESSAGE_PRE_PROCESS,
func=my_hook_function,
priority=0,
enabled=True
)
Hook Function Signatures#
Hook functions receive different arguments depending on the hook type:
# Message hooks
def message_hook(log, workspace):
pass
# Tool hooks
def tool_hook(tool_name, tool_use):
pass
# File hooks
def file_hook(path, content, created=False):
pass
# Session hooks
def session_hook(logdir, workspace, manager=None, initial_msgs=None):
pass
Hook functions can:
Return
None
(no action)Return a single
Message
objectReturn a generator that yields
Message
objectsRaise exceptions (which are caught and logged)
Managing Hooks#
Query Hooks#
from gptme.hooks import get_hooks, HookType
# Get all hooks
all_hooks = get_hooks()
# Get hooks of a specific type
tool_hooks = get_hooks(HookType.TOOL_POST_EXECUTE)
Enable/Disable Hooks#
from gptme.hooks import enable_hook, disable_hook
# Disable a hook
disable_hook("linter.file_save")
# Re-enable it
enable_hook("linter.file_save")
Unregister Hooks#
from gptme.hooks import unregister_hook, HookType
# Unregister from specific type
unregister_hook("my_hook", HookType.FILE_POST_SAVE)
# Unregister from all types
unregister_hook("my_hook")
Examples#
Pre-commit Hook#
Automatically run pre-commit checks after files are saved:
from pathlib import Path
from gptme.tools.base import ToolSpec
from gptme.hooks import HookType
from gptme.message import Message
import subprocess
def run_precommit(path: Path, content: str, created: bool):
"""Run pre-commit on saved file."""
try:
result = subprocess.run(
["pre-commit", "run", "--files", str(path)],
capture_output=True,
text=True,
timeout=30
)
if result.returncode != 0:
yield Message("system", f"Pre-commit checks failed:\n{result.stdout}")
else:
yield Message("system", "Pre-commit checks passed", hide=True)
except subprocess.TimeoutExpired:
yield Message("system", "Pre-commit checks timed out", hide=True)
tool = ToolSpec(
name="precommit",
desc="Automatic pre-commit checks",
hooks={
"precommit_check": (
HookType.FILE_POST_SAVE.value,
run_precommit,
5 # Run after other hooks
)
}
)
Memory/Context Hook#
Automatically add context at session start:
def add_context(logdir, workspace, initial_msgs):
"""Add relevant context at session start."""
context = load_relevant_context(workspace)
if context:
yield Message("system", f"Loaded context:\n{context}", pinned=True)
tool = ToolSpec(
name="memory",
desc="Automatic context loading",
hooks={
"load_context": (
HookType.SESSION_START.value,
add_context,
10
)
}
)
Linting Hook#
Automatically lint files after saving:
def lint_file(path: Path, content: str, created: bool):
"""Lint Python files."""
if path.suffix != ".py":
return
import subprocess
result = subprocess.run(
["ruff", "check", str(path)],
capture_output=True,
text=True
)
if result.returncode != 0:
yield Message("system", f"Linting issues:\n{result.stdout}")
tool = ToolSpec(
name="linter",
desc="Automatic Python linting",
hooks={
"lint": (HookType.FILE_POST_SAVE.value, lint_file, 5)
}
)
Best Practices#
Keep hooks fast: Hooks run synchronously and can slow down operations
Handle errors gracefully: Use try-except to prevent hook failures from breaking the system
Use priorities wisely: Higher priority hooks run first (use for dependencies)
Return Messages appropriately: Use
hide=True
for verbose/debug messagesTest hooks thoroughly: Hooks run in the main execution path
Document hook behavior: Explain what your hooks do and when they run
Consider disabling hooks: Make hooks easy to disable via configuration
Thread Safety#
The hook registry is thread-safe. Each thread maintains its own tool state, and hooks are registered per-thread.
When running in server mode with multiple workers, hooks must be registered in each worker process.
Configuration#
Hooks can be configured via environment variables:
# Example: disable specific hooks
export GPTME_HOOKS_DISABLED="linter.lint,precommit.precommit_check"
# Example: set hook priorities
export GPTME_HOOK_PRIORITY_LINTER=20
Migration Guide#
Converting Existing Features to Hooks#
If you have features that should be hooks:
Identify the appropriate hook type: Choose from the available hook types
Extract the logic: Move the feature logic into a hook function
Register the hook: Add it to a ToolSpec or register programmatically
Test thoroughly: Ensure the hook works in all scenarios
Update documentation: Document the new hook
Example: Converting pre-commit checks to a hook#
Before (hard-coded in chat.py):
# In chat.py
if check_for_modifications(log):
run_precommit_checks()
After (as a hook):
# In a tool
def precommit_hook(log, workspace):
if check_for_modifications(log):
run_precommit_checks()
tool = ToolSpec(
name="precommit",
hooks={
"check": (HookType.MESSAGE_POST_PROCESS.value, precommit_hook, 5)
}
)
API Reference#
Hook system for extending gptme functionality at various lifecycle points.
- class gptme.hooks.Hook#
Bases:
object
A registered hook.
- class gptme.hooks.HookRegistry#
Bases:
object
Registry for managing hooks.
- __init__(hooks: dict[~gptme.hooks.HookType, list[~gptme.hooks.Hook]] = <factory>, _lock: ~typing.Any = <factory>) None #
- get_hooks(hook_type: HookType | None = None) list[Hook] #
Get all registered hooks, optionally filtered by type.
- register(name: str, hook_type: HookType, func: Callable[[...], Any | Generator[Message, None, None]], priority: int = 0, enabled: bool = True) None #
Register a hook.
- trigger(hook_type: HookType, *args, **kwargs) Generator[Message, None, None] #
Trigger all hooks of a given type.
- Parameters:
hook_type – The type of hook to trigger
*args – Variable positional arguments to pass to hook functions
**kwargs – Variable keyword arguments to pass to hook functions
- Yields:
Messages from hooks
- class gptme.hooks.HookType#
-
Types of hooks that can be registered.
- FILE_POST_PATCH = 'file_post_patch'#
- FILE_POST_SAVE = 'file_post_save'#
- FILE_PRE_PATCH = 'file_pre_patch'#
- FILE_PRE_SAVE = 'file_pre_save'#
- GENERATION_INTERRUPT = 'generation_interrupt'#
- GENERATION_POST = 'generation_post'#
- GENERATION_PRE = 'generation_pre'#
- LOOP_CONTINUE = 'loop_continue'#
- MESSAGE_POST_PROCESS = 'message_post_process'#
- MESSAGE_PRE_PROCESS = 'message_pre_process'#
- MESSAGE_TRANSFORM = 'message_transform'#
- SESSION_END = 'session_end'#
- SESSION_START = 'session_start'#
- TOOL_POST_EXECUTE = 'tool_post_execute'#
- TOOL_PRE_EXECUTE = 'tool_pre_execute'#
- TOOL_TRANSFORM = 'tool_transform'#
- __new__(value)#
- class gptme.hooks.StopPropagation#
Bases:
object
Sentinel class that hooks can yield to stop execution of lower-priority hooks.
Usage:
def my_hook(): if some_condition_failed: yield Message("system", "Error occurred") yield StopPropagation() # Stop remaining hooks
- gptme.hooks.clear_hooks(hook_type: HookType | None = None) None #
Clear all hooks, optionally filtered by type.
- gptme.hooks.register_hook(name: str, hook_type: HookType, func: Callable[[...], Any | Generator[Message, None, None]], priority: int = 0, enabled: bool = True) None #
Register a hook with the global registry.
See Also#
Tools - Tool system documentation
Configuration - Configuration options
Issue #156 - Original feature request