Hooks

Hook system for extending Arshai behavior.

Classes

class arshai.extensions.hooks.Hook(name, hook_type, callback, priority=0, enabled=True)

Bases: object

Represents a hook that can be registered in the system.

__init__(name, hook_type, callback, priority=0, enabled=True)

Initialize a hook.

Parameters:

• name (str) – Unique name for the hook

• hook_type (HookType) – Type of hook (when it should be called)

• callback (Callable) – Function to call when hook is triggered

• priority (int) – Priority for execution order (higher = earlier)

• enabled (bool) – Whether the hook is enabled

async execute(context)

Execute the hook callback.

Return type:

Any

class arshai.extensions.hooks.HookContext(hook_type, data, metadata=None)

Bases: object

Context passed to hook functions.

hook_type: HookType

data: Dict[str, Any]

metadata: Dict[str, Any] = None

__init__(hook_type, data, metadata=None)

class arshai.extensions.hooks.HookManager

Bases: object

Manages hooks for the Arshai framework.

__init__()

register_hook(hook)

Register a hook.

Parameters:

hook (Hook) – The hook to register

Return type:

None

unregister_hook(name, hook_type=None)

Unregister a hook.

Parameters:

• name (str) – Name of the hook to unregister

• hook_type (Optional[HookType]) – Type of hook (if None, removes from all types)

Return type:

None

async execute_hooks(hook_type, data, metadata=None)

Execute all hooks of a given type.

Parameters:

• hook_type (HookType) – Type of hooks to execute

• data (Dict[str, Any]) – Data to pass to hooks

• metadata (Optional[Dict[str, Any]]) – Additional metadata

Return type:

List[Any]

Returns:

List of results from hook executions

get_hooks(hook_type)

Get all hooks of a given type.

Return type:

List[Hook]

enable_hook(name)

Enable a hook by name.

Return type:

None

disable_hook(name)

Disable a hook by name.

Return type:

None

class arshai.extensions.hooks.HookType(value)

Bases: Enum

Types of hooks available in the system.

BEFORE_AGENT_PROCESS = 'before_agent_process'

AFTER_AGENT_PROCESS = 'after_agent_process'

BEFORE_WORKFLOW_START = 'before_workflow_start'

AFTER_WORKFLOW_END = 'after_workflow_end'

BEFORE_NODE_EXECUTE = 'before_node_execute'

AFTER_NODE_EXECUTE = 'after_node_execute'

BEFORE_MEMORY_SAVE = 'before_memory_save'

AFTER_MEMORY_SAVE = 'after_memory_save'

BEFORE_MEMORY_RETRIEVE = 'before_memory_retrieve'

AFTER_MEMORY_RETRIEVE = 'after_memory_retrieve'

BEFORE_TOOL_EXECUTE = 'before_tool_execute'

AFTER_TOOL_EXECUTE = 'after_tool_execute'

BEFORE_LLM_CALL = 'before_llm_call'

AFTER_LLM_CALL = 'after_llm_call'

Functions

arshai.extensions.hooks.get_hook_manager()

Get the global hook manager.

Return type:

HookManager

arshai.extensions.hooks.hook(hook_type, name=None, priority=0)

Decorator for registering a function as a hook.

Example

@hook(HookType.BEFORE_AGENT_PROCESS, priority=10) def my_hook(context: HookContext):

print(f”Processing: {context.data}”)