Agent Hooks

Intercept and customize agent behavior at key execution points

Hooks let you intercept agent execution at key points - before/after LLM calls, tool executions, and task completions.

Breaking change in PR #1558: execute_sync() cannot be called from within a running event loop. Use await runner.execute(event, input_data, target) instead in async contexts.

Error message: execute_sync() cannot be called from within a running event loop. Use 'await runner.execute(event, input_data, target)' instead in async contexts.

Quick Start

1. Register Hooks

   from praisonaiagents.hooks import add_hook, HookResult
   
   @add_hook('before_tool')
   def log_tool(event_data):
       print(f"Tool: {event_data.tool_name}")
       return HookResult.allow()
   
2. Create Agent

   from praisonaiagents import Agent
   
   agent = Agent(
       name="Hooked Agent",
       instructions="You are a helpful assistant"
   )
   # Hooks are automatically applied
   agent.start("Help me with a task")
   

Alternative: HooksConfig

from praisonaiagents import Agent, HooksConfig

def on_tool_call(tool_name, args):
    print(f"Tool: {tool_name}({args})")

agent = Agent(
    name="Hooked Agent",
    instructions="You are a helpful assistant",
    hooks=HooksConfig(
        on_tool_call=on_tool_call,
    )
)

---

Hook Types

Available Hooks

Hook	When Triggered	Use Case
BEFORE_LLM	Before LLM call	Modify prompt, log request
AFTER_LLM	After LLM response	Log response, track tokens
BEFORE_TOOL	Before tool execution	Validate args, log call
AFTER_TOOL	After tool execution	Log result, transform output
AFTER_AGENT	After agent completes	Cleanup, final logging

---

Configuration

from praisonaiagents import HooksConfig

config = HooksConfig(
    on_step=my_step_callback,      # Called on each step
    on_tool_call=my_tool_callback, # Called on tool execution
    middleware=[my_middleware],    # Request/response middleware
)

Option	Type	Description
on_step	Callable	Called on each execution step
on_tool_call	Callable	Called when tools are invoked
middleware	list	List of middleware functions

---

Common Patterns

Logging

import logging

def log_llm_call(event):
    logging.info(f"LLM: {event['model']} - {event['tokens']} tokens")

def log_tool_call(tool_name, args, result):
    logging.info(f"Tool: {tool_name}({args}) -> {result}")

agent = Agent(
    instructions="You are helpful",
    hooks=HooksConfig(
        on_step=log_llm_call,
        on_tool_call=log_tool_call,
    )
)

Cost Tracking

total_tokens = 0

def track_tokens(event):
    global total_tokens
    total_tokens += event.get('tokens', 0)
    print(f"Total tokens: {total_tokens}")

agent = Agent(
    instructions="You are helpful",
    hooks=HooksConfig(on_step=track_tokens)
)

Tool Validation

def validate_tool_call(tool_name, args):
    if tool_name == "delete_file":
        if not args.get("confirmed"):
            raise ValueError("Deletion requires confirmation")
    return True

agent = Agent(
    instructions="You manage files",
    hooks=HooksConfig(on_tool_call=validate_tool_call)
)

---

Multi-Agent Hooks

For multi-agent workflows, use MultiAgentHooksConfig:

from praisonaiagents import AgentTeam, MultiAgentHooksConfig

def on_task_start(task):
    print(f"Starting: {task.name}")

def on_task_complete(task, result):
    print(f"Completed: {task.name}")

team = AgentTeam(
    agents=[agent1, agent2],
    hooks=MultiAgentHooksConfig(
        on_task_start=on_task_start,
        on_task_complete=on_task_complete,
        completion_checker=my_checker,
    )
)

---

Middleware

Middleware wraps the entire request/response cycle:

def timing_middleware(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"Duration: {time.time() - start:.2f}s")
        return result
    return wrapper

agent = Agent(
    instructions="You are helpful",
    hooks=HooksConfig(middleware=[timing_middleware])
)

---

Async vs Sync Hook Execution

Choose the right execution method based on your context:

Migration Example

# Before (silently fire-and-forget, results dropped)
runner.execute_sync(event, input_data, target)

# After (in async context)
await runner.execute(event, input_data, target)

# Still works (in non-async context)
runner.execute_sync(event, input_data, target)

---

Best Practices

Keep hooks lightweight

Hooks run synchronously. Heavy operations should be queued for async processing.

Handle errors gracefully

Wrap hook code in try/except to prevent hook failures from crashing the agent.

Use middleware for cross-cutting concerns

Logging, timing, and authentication are good middleware candidates.

---

Related

Callbacks

Event-driven notifications

Guardrails

Output validation