Hooks

Intercept, log, or block agent actions at any lifecycle point — before tools run, after LLM responds, on error

Hooks intercept agent actions at lifecycle points so you can log, modify, or block them without changing agent code.

from praisonaiagents import Agent
from praisonaiagents.hooks import add_hook

@add_hook("before_tool")
def log_tools(event_data):
    print(f"Running tool: {event_data.tool_name}")

agent = Agent(name="SecureAssistant", instructions="You are a helpful assistant.")
agent.start("Help me organise my files")

The user sends a request; hooks intercept tools and lifecycle steps without changing agent code.

Quick Start

1. Simple Usage

   Register a hook with add_hook and any agent picks it up automatically:

   from praisonaiagents import Agent
   from praisonaiagents.hooks import add_hook
   
   @add_hook('before_tool')
   def log_tools(event_data):
       print(f"Running tool: {event_data.tool_name}")
   
   @add_hook('before_tool')
   def block_delete(event_data):
       if "delete" in event_data.tool_name.lower():
           return "Delete operations are blocked"
   
   agent = Agent(
       name="SecureAssistant",
       instructions="You are a helpful assistant."
   )
   
   agent.start("Help me organize my files")

   Hook return values:

   • None or no return → Allow
   • False → Deny
   • "reason" → Deny with custom message
2. Order with priority

   Set priority to control which hook runs first — lower runs earlier, default is 100:

   from praisonaiagents import Agent
   from praisonaiagents.hooks import add_hook
   
   @add_hook('after_llm', priority=10)
   def redact(event_data):
       return None  # runs first — allow
   
   @add_hook('after_llm', priority=90)
   def trace(event_data):
       print("emitted after redaction")  # runs after — sees redacted payload
   
   agent = Agent(name="Assistant", instructions="Help with anything.")
   agent.start("Say hi")

   Hooks in the same priority bucket keep registration order.

3. With HooksConfig

   Attach a HooksConfig to a specific agent for scoped hooks:

   from praisonaiagents import Agent, HooksConfig
   
   def log_step(event_data):
       print(f"Step: {event_data}")
   
   def on_tool(event_data):
       print(f"Tool called: {event_data.tool_name}")
   
   agent = Agent(
       name="MonitoredAgent",
       instructions="You are a helpful assistant.",
       hooks=HooksConfig(
           on_step=log_step,
           on_tool_call=on_tool,
       )
   )
   
   agent.start("Write a haiku about Python")

4. Per-agent event-hook isolation

   Pass a private HookRegistry() so an agent’s event hooks only fire on that agent:

   from praisonaiagents import Agent, HooksConfig
   from praisonaiagents.hooks import HookRegistry, HookEvent
   
   # Private registry — hooks registered here only fire on this agent.
   private = HookRegistry()
   
   @private.on(HookEvent.BEFORE_TOOL)
   def audit(event):
       print(f"[secure-agent] tool: {event.tool_name}")
       return event
   
   secure_agent = Agent(
       name="SecureAgent",
       instructions="Handle sensitive data.",
       hooks=HooksConfig(registry=private),
   )
   
   other_agent = Agent(name="OtherAgent", instructions="Help with coding.")
   # other_agent uses the shared default registry — audit() never fires here.

   Without registry=, every agent shares the process-wide default registry, so a BEFORE_TOOL hook registered for one agent fires on all of them. registry=HookRegistry() keeps them isolated. Available since PraisonAI PR #5247.

Per-agent event-hook isolation

Pick the widest scope that still keeps unrelated agents clean.

A “SecureAgent” audits every tool call while a plain “HelperAgent” runs untouched — the audit hook fires only on the agent that owns the private registry.

from praisonaiagents import Agent, HooksConfig
from praisonaiagents.hooks import HookRegistry, HookEvent

private = HookRegistry()
fired_on = []

@private.on(HookEvent.BEFORE_TOOL)
def audit(event):
    fired_on.append(event.tool_name)   # only records SecureAgent's tools
    return event

secure_agent = Agent(
    name="SecureAgent",
    instructions="Handle sensitive data.",
    hooks=HooksConfig(registry=private),
)
helper_agent = Agent(name="HelperAgent", instructions="Help with coding.")

secure_agent.start("Read the customer record")
helper_agent.start("Format this code snippet")
# audit() recorded only SecureAgent's tools — HelperAgent never triggered it.

API Surface

The simplified API covers registering, inspecting, and emitting hooks — all import from praisonaiagents.hooks.

Runtime components call fire_hook() at real state transitions, so subscribing with add_hook is all a plugin author needs. See fire_hook for the emission API.

Which Hook Point Should I Use?

Pick the lifecycle event that matches what you want to observe or block.

How It Works

A hook that raises, times out, has no callable (a FunctionHook whose func is None), or — for command hooks — exits with an unexpected exit code (anything other than 0 = allow or 2 = blocking; e.g. 127 command-not-found, 126 not-executable) now counts as Deny, not Allow. This applies even when the command already printed {"decision": "allow"} on stdout — a gating hook that crashes after emitting allow-JSON has not actually approved the call, so the runner denies it. Every hook lifecycle path returns HookResult(decision="deny", reason=...) on error — a broken BEFORE_TOOL gate no longer lets the tool call through by accident.

Command Hook Exit Codes

A shell-command hook communicates its verdict through both its exit code and its stdout JSON — the runner requires them to agree on allow.

# Fail closed on a missing shell script
from praisonaiagents.hooks import HookRunner, HookRegistry

runner = HookRunner(HookRegistry(), cwd="/tmp")

# What happens when a command hook script is not on PATH
result = runner._parse_command_output(stdout="", stderr="not found", exit_code=127)
assert result.decision == "deny"
assert "127" in result.reason  # "Hook exited with unexpected code 127: not found"

# Fail closed on allow-JSON + unexpected exit — the allow is discarded
result = runner._parse_command_output(
    stdout='{"decision": "allow"}', stderr="", exit_code=127
)
assert result.decision == "deny"
assert "127" in result.reason

# Explicit deny-JSON still preserved even on unexpected exit
result = runner._parse_command_output(
    stdout='{"decision": "deny", "reason": "policy"}', stderr="", exit_code=127
)
assert result.decision == "deny"
assert result.reason == "policy"  # reason from JSON is respected

# FunctionHook with func=None fails closed
from praisonaiagents.hooks import FunctionHook, HookEvent

hook = FunctionHook(id="broken", name="broken", event=HookEvent.BEFORE_TOOL, func=None)
# When executed, the runner now returns:
#   output=HookResult(decision="deny", reason="Hook 'broken' has no callable")
# so is_blocked() returns True and the tool is refused.

Sequential vs Parallel Hooks

A hook’s execution mode decides whether it can rewrite the payload.

Sequential hooks run one after another and their modified_input is applied back to the payload. Parallel hooks run concurrently and cannot mutate the payload — use them for read-only observers (metrics, logging, tracing). If your hook needs to rewrite the request (redact secrets, inject headers, edit messages), register it with sequential=True.

priority applies before the sequential/parallel split, so a lower-priority sequential hook runs before a higher-priority parallel one — a redact (sequential=True, priority=10) now runs ahead of a trace (priority=20), closing the redact-then-trace leak the runner previously had. See Ordering hooks with priority below.

from praisonaiagents.hooks import HookRegistry, HookEvent

registry = HookRegistry()

# Read-only observer — parallel (default)
registry.register_function(
    event=HookEvent.BEFORE_LLM,
    func=log_prompt,
    name="my.logger",
)

# Rewriter — MUST be sequential to apply modified_input
registry.register_function(
    event=HookEvent.BEFORE_LLM,
    func=redact_pii,
    name="my.pii_redactor",
    sequential=True,   # <-- required for modified_input to take effect
)

Ordering hooks with priority

priority decides which hook on the same event runs first — lower runs earlier, and the default bucket is 100.

Ties fall back to deterministic registration order via a monotonic _seq counter, so ordering is reproducible across environments and processes. Every registration API accepts priority=: register_function(..., priority=100), register_command(..., priority=100), @registry.on(..., priority=100), and add_hook(event, callback, priority=100). The add_hook default is 100 — the same bucket as register_function and @on.

The runner honors priority across the parallel/sequential boundary. It walks the priority-sorted list and runs each contiguous run of same-mode hooks together, so a redact (sequential=True, priority=10) that must precede a trace (priority=20) still runs first. Adjacent parallel hooks are still gathered concurrently. A denial inside a sequential run breaks the whole chain — later groups do not run after a deny.

list_hooks() returns each hook dict with a "priority" key, and the list is already in effective execution order — an operator can see the exact sequence the runner will use.

from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult

registry = HookRegistry()

@registry.on(HookEvent.AFTER_LLM, priority=90)
def trace(event_data):
    return HookResult.allow()

@registry.on(HookEvent.AFTER_LLM, sequential=True, priority=10)
def redact(event_data):
    return HookResult.allow()

# Returned in execution order: redact (10) before trace (90)
for hook in registry.list_hooks()["after_llm"]:
    print(hook["name"], hook["priority"])

Thread Safety

HookRegistry guards its per-event hook lists with a re-entrant lock, so register / unregister / clear / enable_hook / disable_hook and get_hooks are safe under concurrent access from multiple threads. get_hooks snapshots the list under the lock before filtering, so a concurrent unregister / clear on another thread can no longer raise RuntimeError: list changed size during iteration or skip / duplicate a hook mid-iteration. The lock is an RLock, so a hook callback that registers or unregisters other hooks (a legitimate re-entrant pattern) still works.

This matters most for the process-wide default registry (the one you get via get_default_registry() or the module-level add_hook-style helpers) — it is shared across every agent in the process, so plugins registering hooks at import time can race an agent iterating hooks on another thread. As of PraisonAI PR #4634 that race is fixed. fire_hook() targets that default registry too; pass registry= for a scoped one.

from praisonaiagents.hooks import HookRegistry, HookEvent
import threading

registry = HookRegistry()

# Safe to register / unregister from any thread while another thread reads:
def worker():
    hook_id = registry.register_function(
        event=HookEvent.BEFORE_TOOL,
        func=lambda data: data,
        name="my.hook",
    )                                              # under RLock
    registry.get_hooks(HookEvent.BEFORE_TOOL)      # snapshots under RLock
    registry.unregister(hook_id)                   # under RLock

threading.Thread(target=worker).start()

Available Hook Events

The events most agents will ever need are the agent / tool / LLM / error / session ones — start here.

on_retry is emitted once per retryable error, just before the back-off sleep. It runs whether the LLM call is agent.chat(...) (sync) or await agent.achat(...) (async). Sync-registered callbacks on the async path are run in a thread executor — they cannot block the event loop.

Configuration Options

from praisonaiagents import Agent, HooksConfig

agent = Agent(
    instructions="...",
    hooks=HooksConfig(
        on_step=my_step_callback,
        on_tool_call=my_tool_callback,
        middleware=[my_middleware],
    )
)

Common Patterns

Security Filtering

from praisonaiagents import Agent
from praisonaiagents.hooks import add_hook

BLOCKED_TOOLS = {"delete_file", "execute_command", "drop_table"}

@add_hook('before_tool')
def security_filter(event_data):
    if event_data.tool_name in BLOCKED_TOOLS:
        return f"Tool '{event_data.tool_name}' is not allowed in this environment"

agent = Agent(
    name="SafeAgent",
    instructions="Help users manage their documents safely."
)

agent.start("Clean up the old log files")

Guardrails before tracing

Order a redaction hook ahead of an observer so the tracer only ever sees the scrubbed payload — set the redactor to a lower priority.

from praisonaiagents import Agent
from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult

registry = HookRegistry()

# Runs FIRST — redacts secrets in-place, sequential to apply modified_input
@registry.on(HookEvent.AFTER_LLM, sequential=True, priority=10)
def redact_pii(event_data):
    return HookResult.allow()

# Runs AFTER redaction — cheap observer, parallel
@registry.on(HookEvent.AFTER_LLM, priority=90)
def add_trace_span(event_data):
    return HookResult.allow()

agent = Agent(
    name="SafeAgent",
    instructions="Answer using the fetch tool.",
    hooks=registry,
)
agent.start("Fetch the user profile")

Audit Logging

import json
from datetime import datetime
from praisonaiagents import Agent
from praisonaiagents.hooks import add_hook

@add_hook('before_tool')
def audit_before(event_data):
    print(json.dumps({
        "ts": datetime.utcnow().isoformat(),
        "event": "before_tool",
        "tool": event_data.tool_name,
    }))

@add_hook('after_tool')
def audit_after(event_data):
    print(json.dumps({
        "ts": datetime.utcnow().isoformat(),
        "event": "after_tool",
        "tool": event_data.tool_name,
    }))

agent = Agent(
    name="AuditAgent",
    instructions="Process user requests with full audit trail."
)

agent.start("Search for the latest AI research papers")

Redact Secrets from Tool Output

Rewrite the tool result the model sees — scrub API keys before they ever reach the LLM. Returning a value from an after_tool hook replaces event_data.tool_output.

import re
from praisonaiagents import Agent
from praisonaiagents.hooks import add_hook

SECRET_RE = re.compile(r"sk-[A-Za-z0-9]{20,}")

@add_hook('after_tool')
def redact_secrets(event_data):
    """Rewrite the tool result the model sees — never leak API keys."""
    result = event_data.tool_output
    if isinstance(result, str) and SECRET_RE.search(result):
        return SECRET_RE.sub("[REDACTED-SECRET]", result)
    return None  # unchanged

agent = Agent(name="Assistant", instructions="Help with logs.")
agent.start("Read /var/log/api.log and summarise the errors")

Block a Tool Result via GuardrailBlocked

Raise GuardrailBlocked inside after_tool to stop a result reaching the model — mirrors the block path already available on before_tool.

from praisonaiagents import Agent
from praisonaiagents.hooks import add_hook
from praisonaiagents.plugins import GuardrailBlocked

@add_hook('after_tool')
def block_pii(event_data):
    if "ssn=" in str(event_data.tool_output).lower():
        raise GuardrailBlocked("Tool result contained PII; blocked before reaching model.")

agent = Agent(name="Assistant", instructions="Help with customer records.")
agent.start("Look up the account details for order 4821")

Tool Matching with HookRegistry

from praisonaiagents import Agent
from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult

registry = HookRegistry()

@registry.on(HookEvent.BEFORE_TOOL)
def file_guard(event_data) -> HookResult:
    if event_data.tool_name.startswith("write_"):
        print(f"Write operation: {event_data.tool_name}")
    return HookResult.allow()

agent = Agent(
    name="GuardedAgent",
    instructions="You are a helpful assistant.",
    hooks=registry
)

agent.start("Create a summary and save it to output.txt")

Redact a Tool Result

Rewrite event_data.tool_output in place to scrub a secret, or return HookResult.block(reason) to suppress the result entirely.

from praisonaiagents import Agent
from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult

registry = HookRegistry()

@registry.on(HookEvent.AFTER_TOOL)
def redact_secrets(event_data):
    scrubbed = str(event_data.tool_output).replace("sk-SECRET123", "[REDACTED]")
    event_data.tool_output = scrubbed        # rewrite in place
    return HookResult.allow()

@registry.on(HookEvent.AFTER_TOOL)
def block_on_pii(event_data):
    if "ssn=" in str(event_data.tool_output):
        return HookResult.block("PII detected in tool output")
    return HookResult.allow()

agent = Agent(
    name="SafeAgent",
    instructions="Answer using the fetch tool.",
    tools=["fetch"],
    hooks=registry,
)
agent.start("Fetch the user profile")

See Redact / Block Tool Output for the plugin-style equivalent.

Testing with hooks (pytest-xdist safety)

Pass a private HookRegistry() in tests so a hook registered by one test can’t fire on an Agent in another.

from praisonaiagents import Agent
from praisonaiagents.hooks import HookRegistry

def test_my_agent_runs_tools():
    agent = Agent(
        instructions="Test tools",
        tools=[my_tool],
        hooks=HookRegistry(),   # private registry — no global hooks leak in
    )
    assert agent.start("run the tool") is not None

A global BEFORE_TOOL deny-hook registered by another test will silently block your tools and flake only under parallel runners. A private registry gives each Agent zero shared state, so test order and worker assignment stop mattering.

Share one fixture so every test gets a fresh registry:

import pytest
from praisonaiagents import Agent
from praisonaiagents.hooks import HookRegistry

@pytest.fixture
def fresh_hooks():
    return HookRegistry()

def test_tool_runs_clean(fresh_hooks):
    agent = Agent(instructions="Test tools", tools=[my_tool], hooks=fresh_hooks)
    assert agent.start("run the tool") is not None

This is the same mechanism as Per-agent event-hook isolation — a private registry keeps one Agent’s hooks from leaking onto another. See PraisonAI PR #5700 for a real-world fix that isolates hooks in the tool-executor concurrency tests.

Best Practices

from praisonaiagents import Agent, HooksConfig
from praisonaiagents.hooks import before_tool, HookResult

@before_tool
def block_delete(request):
    if request.tool_name == "delete_file":
        return HookResult.block("delete_file is not allowed")
    return request

def log_tool(request):            # observer — return ignored
    print(f"tool: {request.tool_name} args={request.arguments}")

agent = Agent(
    name="Assistant",
    instructions="Manage files safely.",
    hooks=HooksConfig(
        on_tool_call=log_tool,    # logging
        middleware=[block_delete] # control (runs first)
    ),
)
agent.start("Tidy up the workspace")

from praisonaiagents import Agent, HooksConfig

async def log_tool(request):          # observer — awaited, return ignored
    print(f"→ {request.tool_name}({request.arguments})")

agent = Agent(
    instructions="Help with tasks.",
    hooks=HooksConfig(
        on_tool_call=log_tool,        # logging
        middleware=[my_validator],    # control (returns HookResult)
    ),
)

from praisonaiagents import Agent
from praisonaiagents.hooks import add_hook

# No before_agent / after_agent hook registered anywhere in the process.
# The runtime skips input construction entirely — no os.getcwd(), no tools list build.
agent = Agent(name="Fast", instructions="Be helpful.")
agent.start("Say hi")

# Only pay the cost when you actually opt in:
@add_hook("after_agent")
def log_turn(event_data):
    print(f"Turn completed by {event_data.agent_name} in {event_data.execution_time_ms:.0f}ms")

agent.start("Say hi again")   # Now the input is built and after_agent fires.

Related