Hooks

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")

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

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
)

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()

from praisonaiagents import Agent, HooksConfig

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

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")

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")

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")

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")

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")

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")