Hooks

Contents

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 message

  • MESSAGE_POST_PROCESS: After message processing completes

  • MESSAGE_TRANSFORM: Transform message content before processing

Tool Lifecycle Hooks#

  • TOOL_PRE_EXECUTE: Before executing any tool

  • TOOL_POST_EXECUTE: After executing any tool

  • TOOL_TRANSFORM: Transform tool execution

File Operation Hooks#

  • FILE_PRE_SAVE: Before saving a file

  • FILE_POST_SAVE: After saving a file

  • FILE_PRE_PATCH: Before patching a file

  • FILE_POST_PATCH: After patching a file

Session Lifecycle Hooks#

  • SESSION_START: At session start

  • SESSION_END: At session end

Generation Hooks#

  • GENERATION_PRE: Before generating response

  • GENERATION_POST: After generating response

  • GENERATION_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 object

  • Return a generator that yields Message objects

  • Raise 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#

  1. Keep hooks fast: Hooks run synchronously and can slow down operations

  2. Handle errors gracefully: Use try-except to prevent hook failures from breaking the system

  3. Use priorities wisely: Higher priority hooks run first (use for dependencies)

  4. Return Messages appropriately: Use hide=True for verbose/debug messages

  5. Test hooks thoroughly: Hooks run in the main execution path

  6. Document hook behavior: Explain what your hooks do and when they run

  7. 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:

  1. Identify the appropriate hook type: Choose from the available hook types

  2. Extract the logic: Move the feature logic into a hook function

  3. Register the hook: Add it to a ToolSpec or register programmatically

  4. Test thoroughly: Ensure the hook works in all scenarios

  5. 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.FilePostSaveHook#

Bases: Protocol

Hook called after saving a file.

__init__(*args, **kwargs)#
class gptme.hooks.GenerationPreHook#

Bases: Protocol

Hook called before generating response.

Parameters:

  • messages – List of conversation messages

  • workspace – Workspace directory path (optional)

  • manager – Conversation manager (optional, currently always None)

__init__(*args, **kwargs)#
class gptme.hooks.Hook#

Bases: object

A registered hook.

__init__(name: str, hook_type: HookType, func: SessionStartHook | SessionEndHook | ToolExecuteHook | MessageProcessHook | LoopContinueHook | GenerationPreHook | FilePostSaveHook, priority: int = 0, enabled: bool = True) → None#
enabled: bool = True#
func: SessionStartHook | SessionEndHook | ToolExecuteHook | MessageProcessHook | LoopContinueHook | GenerationPreHook | FilePostSaveHook#
hook_type: HookType#
name: str#
priority: int = 0#
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#
disable_hook(name: str, hook_type: HookType | None = None) → None#

Disable a hook by name.

enable_hook(name: str, hook_type: HookType | None = None) → None#

Enable a hook by name.

get_hooks(hook_type: HookType | None = None) → list[Hook]#

Get all registered hooks, optionally filtered by type.

hooks: dict[HookType, list[Hook]]#
register(name: str, hook_type: HookType, func: SessionStartHook | SessionEndHook | ToolExecuteHook | MessageProcessHook | LoopContinueHook | GenerationPreHook | FilePostSaveHook, 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

unregister(name: str, hook_type: HookType | None = None) → None#

Unregister a hook by name, optionally filtering by type.

class gptme.hooks.HookType#

Bases: str, Enum

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.LoopContinueHook#

Bases: Protocol

Hook called to decide whether to continue the chat loop.

Parameters:

  • manager – Conversation manager with log and workspace

  • interactive – Whether in interactive mode

  • prompt_queue – Queue of pending prompts

__init__(*args, **kwargs)#
class gptme.hooks.MessageProcessHook#

Bases: Protocol

Hook called before/after message processing.

Parameters:

manager – Conversation manager with log and workspace

__init__(*args, **kwargs)#
class gptme.hooks.SessionEndHook#

Bases: Protocol

Hook called at session end with logdir and manager.

__init__(*args, **kwargs)#
class gptme.hooks.SessionStartHook#

Bases: Protocol

Hook called at session start with logdir, workspace, and initial messages.

__init__(*args, **kwargs)#
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
class gptme.hooks.ToolExecuteHook#

Bases: Protocol

Hook called before/after tool execution.

Note: Receives log/workspace separately since manager isn’t available in ToolUse.execute() context.

Parameters:

  • log – The conversation log

  • workspace – Workspace directory path

  • tool_use – The tool being executed

__init__(*args, **kwargs)#
gptme.hooks.clear_hooks(hook_type: HookType | None = None) → None#

Clear all hooks, optionally filtered by type.

gptme.hooks.disable_hook(name: str, hook_type: HookType | None = None) → None#

Disable a hook.

gptme.hooks.enable_hook(name: str, hook_type: HookType | None = None) → None#

Enable a hook.

gptme.hooks.get_hooks(hook_type: HookType | None = None) → list[Hook]#

Get all registered hooks.

gptme.hooks.register_hook(name: str, hook_type: HookType, func: SessionStartHook | SessionEndHook | ToolExecuteHook | MessageProcessHook | LoopContinueHook | GenerationPreHook | FilePostSaveHook, priority: int = 0, enabled: bool = True) → None#

Register a hook with the global registry.

gptme.hooks.trigger_hook(hook_type: HookType, *args, **kwargs) → Generator[Message, None, None]#

Trigger hooks of a given type.

gptme.hooks.unregister_hook(name: str, hook_type: HookType | None = None) → None#

Unregister a hook from the global registry.

See Also#

Contents