1  #!/usr/bin/env python3
  2  """
  3  Basic Hooks Example for PraisonAI Agents.
  4  
  5  This example demonstrates the simplified hook API using add_hook():
  6  1. Log tool calls before execution
  7  2. Block dangerous operations
  8  3. Add context to agent responses
  9  4. Monitor session lifecycle
 10  
 11  For the simplest example, see simple_hooks.py
 12  
 13  Usage:
 14      python basic_hooks.py
 15  """
 16  
 17  import asyncio
 18  from praisonaiagents.hooks import (
 19      add_hook, remove_hook, has_hook, get_default_registry,
 20      HookRunner, HookEvent, HookResult,
 21      BeforeToolInput, AfterToolInput
 22  )
 23  
 24  
 25  def main():
 26      # Clear any existing hooks from previous runs
 27      get_default_registry().clear()
 28      
 29      # ==========================================================================
 30      # Hook 1: Log all tool calls (using string event name)
 31      # ==========================================================================
 32      @add_hook('before_tool')
 33      def log_tool_calls(event_data: BeforeToolInput) -> HookResult:
 34          """Log every tool call before execution."""
 35          print(f"\n📝 [LOG] Tool: {event_data.tool_name}")
 36          print(f"   Input: {event_data.tool_input}")
 37          return HookResult.allow()
 38      
 39      # ==========================================================================
 40      # Hook 2: Block dangerous file operations
 41      # ==========================================================================
 42      @add_hook('before_tool')
 43      def block_delete_operations(event_data: BeforeToolInput) -> HookResult:
 44          """Block any delete or remove operations."""
 45          dangerous_keywords = ['delete', 'remove', 'rm', 'unlink']
 46          tool_lower = event_data.tool_name.lower()
 47          
 48          if any(kw in tool_lower for kw in dangerous_keywords):
 49              print(f"\n🚫 [BLOCKED] Dangerous operation: {event_data.tool_name}")
 50              return HookResult.deny(
 51                  f"Operation '{event_data.tool_name}' is blocked by security policy"
 52              )
 53          return HookResult.allow()
 54      
 55      # ==========================================================================
 56      # Hook 3: Require confirmation for write operations
 57      # ==========================================================================
 58      @add_hook('before_tool')
 59      def confirm_write_operations(event_data: BeforeToolInput) -> HookResult:
 60          """Request confirmation for write operations."""
 61          write_keywords = ['write', 'save', 'create', 'update']
 62          tool_lower = event_data.tool_name.lower()
 63          
 64          if any(kw in tool_lower for kw in write_keywords):
 65              print(f"\n⚠️  [CONFIRM] Write operation: {event_data.tool_name}")
 66              # In a real scenario, you might prompt the user here
 67              return HookResult.allow("Auto-approved for demo")
 68          return HookResult.allow()
 69      
 70      # ==========================================================================
 71      # Hook 4: Add timing information after tool execution
 72      # ==========================================================================
 73      @add_hook('after_tool')
 74      def log_tool_completion(event_data: AfterToolInput) -> HookResult:
 75          """Log tool completion with timing."""
 76          print(f"\n✅ [DONE] Tool: {event_data.tool_name}")
 77          print(f"   Duration: {event_data.execution_time_ms:.2f}ms")
 78          if event_data.tool_error:
 79              print(f"   Error: {event_data.tool_error}")
 80          return HookResult.allow()
 81      
 82      # ==========================================================================
 83      # Test the hooks
 84      # ==========================================================================
 85      print("=" * 60)
 86      print("Testing Hooks System (using add_hook API)")
 87      print("=" * 60)
 88      
 89      # Check hooks are registered
 90      print(f"\nHooks registered for 'before_tool': {has_hook('before_tool')}")
 91      print(f"Hooks registered for 'after_tool': {has_hook('after_tool')}")
 92      
 93      runner = HookRunner(get_default_registry())
 94      
 95      # Test 1: Normal tool call (should be logged and allowed)
 96      print("\n--- Test 1: Normal read operation ---")
 97      input1 = BeforeToolInput(
 98          session_id="test",
 99          cwd="/tmp",
100          event_name="before_tool",
101          timestamp="2024-01-01T00:00:00",
102          tool_name="read_file",
103          tool_input={"path": "/tmp/test.txt"}
104      )
105      results = asyncio.run(runner.execute(HookEvent.BEFORE_TOOL, input1))
106      print(f"Result: {'ALLOWED' if not HookRunner.is_blocked(results) else 'BLOCKED'}")
107      
108      # Test 2: Delete operation (should be blocked)
109      print("\n--- Test 2: Delete operation ---")
110      input2 = BeforeToolInput(
111          session_id="test",
112          cwd="/tmp",
113          event_name="before_tool",
114          timestamp="2024-01-01T00:00:00",
115          tool_name="delete_file",
116          tool_input={"path": "/important/data.txt"}
117      )
118      results = asyncio.run(runner.execute(HookEvent.BEFORE_TOOL, input2))
119      print(f"Result: {'ALLOWED' if not HookRunner.is_blocked(results) else 'BLOCKED'}")
120      if HookRunner.is_blocked(results):
121          print(f"Reason: {HookRunner.get_blocking_reason(results)}")
122      
123      # Test 3: Write operation (should be confirmed)
124      print("\n--- Test 3: Write operation ---")
125      input3 = BeforeToolInput(
126          session_id="test",
127          cwd="/tmp",
128          event_name="before_tool",
129          timestamp="2024-01-01T00:00:00",
130          tool_name="write_file",
131          tool_input={"path": "/tmp/output.txt", "content": "Hello"}
132      )
133      results = asyncio.run(runner.execute(HookEvent.BEFORE_TOOL, input3))
134      print(f"Result: {'ALLOWED' if not HookRunner.is_blocked(results) else 'BLOCKED'}")
135      
136      # Show registered hooks
137      print("\n" + "=" * 60)
138      print("Registered Hooks:")
139      print("=" * 60)
140      for event, hooks in get_default_registry().list_hooks().items():
141          print(f"\n{event}:")
142          for hook in hooks:
143              print(f"  - {hook['name']} (matcher: {hook['matcher'] or '*'})")
144  
145  
146  if __name__ == "__main__":
147      main()