Documentation Index
Fetch the complete documentation index at: https://polos.dev/docs/llms.txt
Use this file to discover all available pages before exploring further.
Lifecycle hooks let you customize agent behavior at key execution points. Use hooks for logging, validation, or modifying inputs/outputs.
What are lifecycle hooks?
Hooks are functions that run at specific stages of agent execution:
on_start - Before agent execution begins
on_end - After agent execution completes
on_agent_step_start - Before each agent reasoning step
on_agent_step_end - After each agent reasoning step
on_tool_start - Before each tool execution
on_tool_end - After each tool execution
Defining hooks
Create hooks using the @hook decorator in Python or the defineHook function in TypeScript:
from polos import hook, WorkflowContext, HookContext, HookResult
@hook
def log_hook(ctx: WorkflowContext, hook_context: HookContext) -> HookResult:
"""Log execution details."""
print(f"Workflow: {hook_context.workflow_id}")
print(f"Payload: {hook_context.current_payload}")
return HookResult.continue_with()
Hook signature:
ctx - WorkflowContext with execution metadata
hook_context - HookContext with current execution state
- Returns -
HookResult indicating what action to take
Hook results
Hooks return HookResult with three options:
1. Continue without changes
@hook
def simple_hook(ctx: WorkflowContext, hook_context: HookContext) -> HookResult:
# Just observe, don't modify
print("Hook executed")
return HookResult.continue_with()
2. Continue with modifications
import re
@hook
def redact_pii_hook(ctx: WorkflowContext, hook_context: HookContext) -> HookResult:
"""Redact sensitive information from payloads."""
if not hook_context.current_payload:
return HookResult.continue_with()
modified = hook_context.current_payload.copy()
# Redact emails
if isinstance(modified, dict):
for key, value in modified.items():
if isinstance(value, str):
modified[key] = re.sub(
r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
'[EMAIL_REDACTED]',
value
)
return HookResult.continue_with(modified_payload=modified)
3. Fail and stop execution
@hook
def validation_hook(ctx: WorkflowContext, hook_context: HookContext) -> HookResult:
"""Validate payload before execution."""
payload = hook_context.current_payload
if not payload or not payload.get("required_field"):
return HookResult.fail("Missing required field")
return HookResult.continue_with()
Using hooks
Attach hooks to agents:
from polos import Agent
tracked_agent = Agent(
id="tracked-agent",
provider="openai",
model="gpt-4o",
system_prompt="You are a helpful assistant.",
tools=[search, send_email],
# Workflow hooks
on_start=[log_hook, validate_input_hook],
on_end=[log_hook, save_results_hook],
# Step hooks
on_agent_step_start=[log_step_hook],
on_agent_step_end=[validate_step_hook],
# Tool hooks
on_tool_start=[log_tool_call_hook],
on_tool_end=[audit_tool_result_hook]
)
Hook context
Hooks receive HookContext with execution state:
@hook
def inspect_context_hook(ctx: WorkflowContext, hook_context: HookContext) -> HookResult:
# Workflow information
workflow_id = hook_context.workflow_id
# User context
user_id = hook_context.user_id
session_id = hook_context.session_id
# Execution state
current_payload = hook_context.current_payload
current_output = hook_context.current_output
current_tool = hook_context.current_tool
# Agent configuration
agent_config = hook_context.agent_config
# Step history
steps = hook_context.steps # List of completed steps
return HookResult.continue_with()
Multiple hooks
Hooks run in order. If any hook fails, execution stops:
agent = Agent(
id="multi-hook-agent",
provider="openai",
model="gpt-4o",
on_start=[
validate_input_hook, # Runs first
redact_pii_hook, # Runs second (on validated input)
add_metadata_hook # Runs third (on redacted input)
]
)
If validate_input_hook fails, redact_pii_hook and add_metadata_hook never run.
Hook execution flow
Complete agent execution with hooks: