Lorenze/feat hooks

lib/crewai/src/crewai/agents/crew_agent_executor.py

@@ -23,6 +23,10 @@
     AgentLogsExecutionEvent,
     AgentLogsStartedEvent,
 )
+from crewai.hooks.llm_hooks import (
+    get_after_llm_call_hooks,
+    get_before_llm_call_hooks,
+)
 from crewai.utilities.agent_utils import (
     enforce_rpm_limit,
     format_message_for_llm,
@@ -38,10 +42,6 @@
 )
 from crewai.utilities.constants import TRAINING_DATA_FILE
 from crewai.utilities.i18n import I18N, get_i18n
-from crewai.utilities.llm_call_hooks import (
-    get_after_llm_call_hooks,
-    get_before_llm_call_hooks,
-)
 from crewai.utilities.printer import Printer
 from crewai.utilities.tool_utils import execute_tool_and_check_finality
 from crewai.utilities.training_handler import CrewTrainingHandler
@@ -263,6 +263,7 @@ def _invoke_loop(self) -> AgentFinish:
                         task=self.task,
                         agent=self.agent,
                         function_calling_llm=self.function_calling_llm,
+                        crew=self.crew,
                     )
                     formatted_answer = self._handle_agent_action(
                         formatted_answer, tool_result

lib/crewai/src/crewai/hooks/__init__.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+# LLM Hooks
+from crewai.hooks.llm_hooks import (
+    LLMCallHookContext,
+    get_after_llm_call_hooks,
+    get_before_llm_call_hooks,
+    register_after_llm_call_hook,
+    register_before_llm_call_hook,
+)
+
+# Tool Hooks
+from crewai.hooks.tool_hooks import (
+    ToolCallHookContext,
+    get_after_tool_call_hooks,
+    get_before_tool_call_hooks,
+    register_after_tool_call_hook,
+    register_before_tool_call_hook,
+)
+
+
+__all__ = [
+    "LLMCallHookContext",
+    "ToolCallHookContext",
+    "get_after_llm_call_hooks",
+    "get_after_tool_call_hooks",
+    "get_before_llm_call_hooks",
+    "get_before_tool_call_hooks",
+    "register_after_llm_call_hook",
+    "register_after_tool_call_hook",
+    "register_before_llm_call_hook",
+    "register_before_tool_call_hook",
+]

lib/crewai/src/crewai/hooks/base.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from typing import TypeVar
+
+
+# Type variable for hook context types
+HookContextT = TypeVar("HookContextT")
+
+
+def validate_hook_callable(hook: object, hook_type: str) -> None:
+    """Validate that a hook is callable.
+
+    Args:
+        hook: The hook object to validate
+        hook_type: Description of the hook type for error messages
+
+    Raises:
+        TypeError: If the hook is not callable
+    """
+    if not callable(hook):
+        raise TypeError(f"{hook_type} must be callable, got {type(hook).__name__}")

lib/crewai/src/crewai/utilities/llm_call_hooks.py → lib/crewai/src/crewai/hooks/llm_hooks.py

@@ -3,6 +3,9 @@
 from collections.abc import Callable
 from typing import TYPE_CHECKING
 
+from crewai.events.event_listener import event_listener
+from crewai.utilities.printer import Printer
+
 
 if TYPE_CHECKING:
     from crewai.agents.crew_agent_executor import CrewAgentExecutor
@@ -53,8 +56,51 @@ def __init__(
         self.iterations = executor.iterations
         self.response = response
 
+    def request_human_input(
+        self,
+        prompt: str,
+        default_message: str = "Press Enter to continue, or provide feedback:",
+    ) -> str:
+        """Request human input during LLM hook execution.
+
+        This method pauses live console updates, displays a prompt to the user,
+        waits for their input, and then resumes live updates. This is useful for
+        approval gates, debugging, or getting human feedback during execution.
+
+        Args:
+            prompt: Custom message to display to the user
+            default_message: Message shown after the prompt
+
+        Returns:
+            User's input as a string (empty string if just Enter pressed)
+
+        Example:
+            >>> def approval_hook(context: LLMCallHookContext) -> None:
+            ...     if context.iterations > 5:
+            ...         response = context.request_human_input(
+            ...             prompt="Allow this LLM call?",
+            ...             default_message="Type 'no' to skip, or press Enter:",
+            ...         )
+            ...         if response.lower() == "no":
+            ...             print("LLM call skipped by user")
+        """
+
+        printer = Printer()
+        event_listener.formatter.pause_live_updates()
+
+        try:
+            printer.print(content=f"\n{prompt}", color="bold_yellow")
+            printer.print(content=default_message, color="cyan")
+            response = input().strip()
+
+            if response:
+                printer.print(content="\nProcessing your input...", color="cyan")
+
+            return response
+        finally:
+            event_listener.formatter.resume_live_updates()
+
 
-# Global hook registries (optional convenience feature)
 _before_llm_call_hooks: list[Callable[[LLMCallHookContext], None]] = []
 _after_llm_call_hooks: list[Callable[[LLMCallHookContext], str | None]] = []
 
@@ -73,6 +119,13 @@ def register_before_llm_call_hook(
             context.messages directly. Should return None.
             IMPORTANT: Modify messages in-place (append, extend, remove items).
             Do NOT replace the list (context.messages = []), as this will break execution.
+
+    Example:
+        >>> def log_llm_calls(context: LLMCallHookContext) -> None:
+        ...     print(f"LLM call by {context.agent.role}")
+        ...     print(f"Messages: {len(context.messages)}")
+        >>>
+        >>> register_before_llm_call_hook(log_llm_calls)
     """
     _before_llm_call_hooks.append(hook)
 
@@ -93,6 +146,14 @@ def register_after_llm_call_hook(
             Both modifications are supported and can be used together.
             IMPORTANT: Modify messages in-place (append, extend, remove items).
             Do NOT replace the list (context.messages = []), as this will break execution.
+
+    Example:
+        >>> def sanitize_response(context: LLMCallHookContext) -> str | None:
+        ...     if context.response and "SECRET" in context.response:
+        ...         return context.response.replace("SECRET", "[REDACTED]")
+        ...     return None
+        >>>
+        >>> register_after_llm_call_hook(sanitize_response)
     """
     _after_llm_call_hooks.append(hook)
 

lib/crewai/src/crewai/hooks/tool_hooks.py

@@ -0,0 +1,202 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import TYPE_CHECKING, Any
+
+from crewai.events.event_listener import event_listener
+from crewai.utilities.printer import Printer
+
+
+if TYPE_CHECKING:
+    from crewai.agent import Agent
+    from crewai.agents.agent_builder.base_agent import BaseAgent
+    from crewai.crew import Crew
+    from crewai.task import Task
+    from crewai.tools.structured_tool import CrewStructuredTool
+
+
+class ToolCallHookContext:
+    """Context object passed to tool call hooks.
+
+    Provides hooks with access to the tool being called, its input,
+    the agent/task/crew context, and the result (for after hooks).
+
+    Attributes:
+        tool_name: Name of the tool being called
+        tool_input: Tool input parameters (mutable dict).
+            Can be modified in-place by before_tool_call hooks.
+            IMPORTANT: Modify in-place (e.g., context.tool_input['key'] = value).
+            Do NOT replace the dict (e.g., context.tool_input = {}), as this
+            will not affect the actual tool execution.
+        tool: Reference to the CrewStructuredTool instance
+        agent: Agent executing the tool (may be None)
+        task: Current task being executed (may be None)
+        crew: Crew instance (may be None)
+        tool_result: Tool execution result (only set for after_tool_call hooks).
+            Can be modified by returning a new string from after_tool_call hook.
+    """
+
+    def __init__(
+        self,
+        tool_name: str,
+        tool_input: dict[str, Any],
+        tool: CrewStructuredTool,
+        agent: Agent | BaseAgent | None = None,
+        task: Task | None = None,
+        crew: Crew | None = None,
+        tool_result: str | None = None,
+    ) -> None:
+        """Initialize tool call hook context.
+
+        Args:
+            tool_name: Name of the tool being called
+            tool_input: Tool input parameters (mutable)
+            tool: Tool instance reference
+            agent: Optional agent executing the tool
+            task: Optional current task
+            crew: Optional crew instance
+            tool_result: Optional tool result (for after hooks)
+        """
+        self.tool_name = tool_name
+        self.tool_input = tool_input
+        self.tool = tool
+        self.agent = agent
+        self.task = task
+        self.crew = crew
+        self.tool_result = tool_result
+
+    def request_human_input(
+        self,
+        prompt: str,
+        default_message: str = "Press Enter to continue, or provide feedback:",
+    ) -> str:
+        """Request human input during tool hook execution.
+
+        This method pauses live console updates, displays a prompt to the user,
+        waits for their input, and then resumes live updates. This is useful for
+        approval gates, reviewing tool results, or getting human feedback during execution.
+
+        Args:
+            prompt: Custom message to display to the user
+            default_message: Message shown after the prompt
+
+        Returns:
+            User's input as a string (empty string if just Enter pressed)
+
+        Example:
+            >>> def approval_hook(context: ToolCallHookContext) -> bool | None:
+            ...     if context.tool_name == "delete_file":
+            ...         response = context.request_human_input(
+            ...             prompt="Allow file deletion?",
+            ...             default_message="Type 'approve' to continue:",
+            ...         )
+            ...         if response.lower() != "approve":
+            ...             return False  # Block execution
+            ...     return None  # Allow execution
+        """
+
+        printer = Printer()
+        event_listener.formatter.pause_live_updates()
+
+        try:
+            printer.print(content=f"\n{prompt}", color="bold_yellow")
+            printer.print(content=default_message, color="cyan")
+            response = input().strip()
+
+            if response:
+                printer.print(content="\nProcessing your input...", color="cyan")
+
+            return response
+        finally:
+            event_listener.formatter.resume_live_updates()
+
+
+# Global hook registries
+_before_tool_call_hooks: list[Callable[[ToolCallHookContext], bool | None]] = []
+_after_tool_call_hooks: list[Callable[[ToolCallHookContext], str | None]] = []
+
+
+def register_before_tool_call_hook(
+    hook: Callable[[ToolCallHookContext], bool | None],
+) -> None:
+    """Register a global before_tool_call hook.
+
+    Global hooks are added to all tool executions automatically.
+    This is a convenience function for registering hooks that should
+    apply to all tool calls across all agents and crews.
+
+    Args:
+        hook: Function that receives ToolCallHookContext and can:
+            - Modify tool_input in-place
+            - Return False to block tool execution
+            - Return True or None to allow execution
+            IMPORTANT: Modify tool_input in-place (e.g., context.tool_input['key'] = value).
+            Do NOT replace the dict (context.tool_input = {}), as this will not affect
+            the actual tool execution.
+
+    Example:
+        >>> def log_tool_usage(context: ToolCallHookContext) -> None:
+        ...     print(f"Executing tool: {context.tool_name}")
+        ...     print(f"Input: {context.tool_input}")
+        ...     return None  # Allow execution
+        >>>
+        >>> register_before_tool_call_hook(log_tool_usage)
+
+        >>> def block_dangerous_tools(context: ToolCallHookContext) -> bool | None:
+        ...     if context.tool_name == "delete_database":
+        ...         print("Blocked dangerous tool execution!")
+        ...         return False  # Block execution
+        ...     return None  # Allow execution
+        >>>
+        >>> register_before_tool_call_hook(block_dangerous_tools)
+    """
+    _before_tool_call_hooks.append(hook)
+
+
+def register_after_tool_call_hook(
+    hook: Callable[[ToolCallHookContext], str | None],
+) -> None:
+    """Register a global after_tool_call hook.
+
+    Global hooks are added to all tool executions automatically.
+    This is a convenience function for registering hooks that should
+    apply to all tool calls across all agents and crews.
+
+    Args:
+        hook: Function that receives ToolCallHookContext and can modify
+            the tool result. Return modified result string or None to keep
+            the original result. The tool_result is available in context.tool_result.
+
+    Example:
+        >>> def sanitize_output(context: ToolCallHookContext) -> str | None:
+        ...     if context.tool_result and "SECRET_KEY" in context.tool_result:
+        ...         return context.tool_result.replace("SECRET_KEY=...", "[REDACTED]")
+        ...     return None  # Keep original result
+        >>>
+        >>> register_after_tool_call_hook(sanitize_output)
+
+        >>> def log_tool_results(context: ToolCallHookContext) -> None:
+        ...     print(f"Tool {context.tool_name} returned: {context.tool_result[:100]}")
+        ...     return None  # Keep original result
+        >>>
+        >>> register_after_tool_call_hook(log_tool_results)
+    """
+    _after_tool_call_hooks.append(hook)
+
+
+def get_before_tool_call_hooks() -> list[Callable[[ToolCallHookContext], bool | None]]:
+    """Get all registered global before_tool_call hooks.
+
+    Returns:
+        List of registered before hooks
+    """
+    return _before_tool_call_hooks.copy()
+
+
+def get_after_tool_call_hooks() -> list[Callable[[ToolCallHookContext], str | None]]:
+    """Get all registered global after_tool_call hooks.
+
+    Returns:
+        List of registered after hooks
+    """
+    return _after_tool_call_hooks.copy()