Tool Call Hooks - Crewai

Tool Call Hooks provide fine-grained control over tool execution during agent operations. These hooks allow you to intercept tool calls, modify inputs, transform outputs, implement safety checks, and add comprehensive logging or monitoring.

Overview

Tool hooks are executed at two critical points:

Before Tool Call: Modify inputs, validate parameters, or block execution
After Tool Call: Transform results, sanitize outputs, or log execution details

Hook Types

Before Tool Call Hooks

Executed before every tool execution, these hooks can:

Inspect and modify tool inputs
Block tool execution based on conditions
Implement approval gates for dangerous operations
Validate parameters
Log tool invocations

Signature:

python

def before_hook(context: ToolCallHookContext) -> bool | None:
    # Return False to block execution
    # Return True or None to allow execution
    ...

After Tool Call Hooks

Executed after every tool execution, these hooks can:

Modify or sanitize tool results
Add metadata or formatting
Log execution results
Implement result validation
Transform output formats

Signature:

python

def after_hook(context: ToolCallHookContext) -> str | None:
    # Return modified result string
    # Return None to keep original result
    ...

Tool Hook Context

The ToolCallHookContext object provides comprehensive access to tool execution state:

python

class ToolCallHookContext:
    tool_name: str                    # Name of the tool being called
    tool_input: dict[str, Any]        # Mutable tool input parameters
    tool: CrewStructuredTool          # Tool instance reference
    agent: Agent | BaseAgent | None   # Agent executing the tool
    task: Task | None                 # Current task
    crew: Crew | None                 # Crew instance
    tool_result: str | None           # Tool result (after hooks only)

Modifying Tool Inputs

Important: Always modify tool inputs in-place:

python

# ✅ Correct - modify in-place
def sanitize_input(context: ToolCallHookContext) -> None:
    context.tool_input['query'] = context.tool_input['query'].lower()

# ❌ Wrong - replaces dict reference
def wrong_approach(context: ToolCallHookContext) -> None:
    context.tool_input = {'query': 'new query'}

Registration Methods

1. Global Hook Registration

python

from crewai.hooks import register_before_tool_call_hook, register_after_tool_call_hook

def log_tool_call(context):
    print(f"Tool: {context.tool_name}")
    print(f"Input: {context.tool_input}")
    return None  # Allow execution

register_before_tool_call_hook(log_tool_call)

2. Decorator-Based Registration

Use decorators for cleaner syntax:

python

from crewai.hooks import before_tool_call, after_tool_call

@before_tool_call
def block_dangerous_tools(context):
    dangerous_tools = ['delete_database', 'drop_table', 'rm_rf']
    if context.tool_name in dangerous_tools:
        print(f"⛔ Blocked dangerous tool: {context.tool_name}")
        return False  # Block execution
    return None

@after_tool_call
def sanitize_results(context):
    if context.tool_result and "password" in context.tool_result.lower():
        return context.tool_result.replace("password", "[REDACTED]")
    return None

3. Crew-Scoped Hooks

python

@CrewBase
class MyProjCrew:
    @before_tool_call_crew
    def validate_tool_inputs(self, context):
        # Only applies to this crew
        if context.tool_name == "web_search":
            if not context.tool_input.get('query'):
                print("❌ Invalid search query")
                return False
        return None

    @after_tool_call_crew
    def log_tool_results(self, context):
        # Crew-specific tool logging
        print(f"✅ {context.tool_name} completed")
        return None

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True
        )

Common Use Cases

1. Safety Guardrails

python

@before_tool_call
def safety_check(context: ToolCallHookContext) -> bool | None:
    # Block tools that could cause harm
    destructive_tools = [
        'delete_file',
        'drop_table',
        'remove_user',
        'system_shutdown'
    ]

    if context.tool_name in destructive_tools:
        print(f"🛑 Blocked destructive tool: {context.tool_name}")
        return False

    # Warn on sensitive operations
    sensitive_tools = ['send_email', 'post_to_social_media', 'charge_payment']
    if context.tool_name in sensitive_tools:
        print(f"⚠️  Executing sensitive tool: {context.tool_name}")

    return None

2. Human Approval Gate

python

@before_tool_call
def require_approval_for_actions(context: ToolCallHookContext) -> bool | None:
    approval_required = [
        'send_email',
        'make_purchase',
        'delete_file',
        'post_message'
    ]

    if context.tool_name in approval_required:
        response = context.request_human_input(
            prompt=f"Approve {context.tool_name}?",
            default_message=f"Input: {context.tool_input}\nType 'yes' to approve:"
        )

        if response.lower() != 'yes':
            print(f"❌ Tool execution denied: {context.tool_name}")
            return False

    return None

3. Input Validation and Sanitization

python

@before_tool_call
def validate_and_sanitize_inputs(context: ToolCallHookContext) -> bool | None:
    # Validate search queries
    if context.tool_name == 'web_search':
        query = context.tool_input.get('query', '')
        if len(query) < 3:
            print("❌ Search query too short")
            return False

        # Sanitize query
        context.tool_input['query'] = query.strip().lower()

    # Validate file paths
    if context.tool_name == 'read_file':
        path = context.tool_input.get('path', '')
        if '..' in path or path.startswith('/'):
            print("❌ Invalid file path")
            return False

    return None

4. Result Sanitization

python

@after_tool_call
def sanitize_sensitive_data(context: ToolCallHookContext) -> str | None:
    if not context.tool_result:
        return None

    import re
    result = context.tool_result

    # Remove API keys
    result = re.sub(
        r'(api[_-]?key|token)["\']?\s*[:=]\s*["\']?[\w-]+',
        r'\1: [REDACTED]',
        result,
        flags=re.IGNORECASE
    )

    # Remove email addresses
    result = re.sub(
        r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        '[EMAIL-REDACTED]',
        result
    )

    # Remove credit card numbers
    result = re.sub(
        r'\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b',
        '[CARD-REDACTED]',
        result
    )

    return result

5. Tool Usage Analytics

python

import time
from collections import defaultdict

tool_stats = defaultdict(lambda: {'count': 0, 'total_time': 0, 'failures': 0})

@before_tool_call
def start_timer(context: ToolCallHookContext) -> None:
    context.tool_input['_start_time'] = time.time()
    return None

@after_tool_call
def track_tool_usage(context: ToolCallHookContext) -> None:
    start_time = context.tool_input.get('_start_time', time.time())
    duration = time.time() - start_time

    tool_stats[context.tool_name]['count'] += 1
    tool_stats[context.tool_name]['total_time'] += duration

    if not context.tool_result or 'error' in context.tool_result.lower():
        tool_stats[context.tool_name]['failures'] += 1

    print(f"""
    📊 Tool Stats for {context.tool_name}:
    - Executions: {tool_stats[context.tool_name]['count']}
    - Avg Time: {tool_stats[context.tool_name]['total_time'] / tool_stats[context.tool_name]['count']:.2f}s
    - Failures: {tool_stats[context.tool_name]['failures']}
    """)

    return None

6. Rate Limiting

python

from collections import defaultdict
from datetime import datetime, timedelta

tool_call_history = defaultdict(list)

@before_tool_call
def rate_limit_tools(context: ToolCallHookContext) -> bool | None:
    tool_name = context.tool_name
    now = datetime.now()

    # Clean old entries (older than 1 minute)
    tool_call_history[tool_name] = [
        call_time for call_time in tool_call_history[tool_name]
        if now - call_time < timedelta(minutes=1)
    ]

    # Check rate limit (max 10 calls per minute)
    if len(tool_call_history[tool_name]) >= 10:
        print(f"🚫 Rate limit exceeded for {tool_name}")
        return False

    # Record this call
    tool_call_history[tool_name].append(now)
    return None

7. Caching Tool Results

python

import hashlib
import json

tool_cache = {}

def cache_key(tool_name: str, tool_input: dict) -> str:
    """Generate cache key from tool name and input."""
    input_str = json.dumps(tool_input, sort_keys=True)
    return hashlib.md5(f"{tool_name}:{input_str}".encode()).hexdigest()

@before_tool_call
def check_cache(context: ToolCallHookContext) -> bool | None:
    key = cache_key(context.tool_name, context.tool_input)
    if key in tool_cache:
        print(f"💾 Cache hit for {context.tool_name}")
        # Note: Can't return cached result from before hook
        # Would need to implement this differently
    return None

@after_tool_call
def cache_result(context: ToolCallHookContext) -> None:
    if context.tool_result:
        key = cache_key(context.tool_name, context.tool_input)
        tool_cache[key] = context.tool_result
        print(f"💾 Cached result for {context.tool_name}")
    return None

8. Debug Logging

python

@before_tool_call
def debug_tool_call(context: ToolCallHookContext) -> None:
    print(f"""
    🔍 Tool Call Debug:
    - Tool: {context.tool_name}
    - Agent: {context.agent.role if context.agent else 'Unknown'}
    - Task: {context.task.description[:50] if context.task else 'Unknown'}...
    - Input: {context.tool_input}
    """)
    return None

@after_tool_call
def debug_tool_result(context: ToolCallHookContext) -> None:
    if context.tool_result:
        result_preview = context.tool_result[:200]
        print(f"✅ Result Preview: {result_preview}...")
    else:
        print("⚠️  No result returned")
    return None

Hook Management

Unregistering Hooks

python

from crewai.hooks import (
    unregister_before_tool_call_hook,
    unregister_after_tool_call_hook
)

# Unregister specific hook
def my_hook(context):
    ...

register_before_tool_call_hook(my_hook)
# Later...
success = unregister_before_tool_call_hook(my_hook)
print(f"Unregistered: {success}")

Clearing Hooks

python

from crewai.hooks import (
    clear_before_tool_call_hooks,
    clear_after_tool_call_hooks,
    clear_all_tool_call_hooks
)

# Clear specific hook type
count = clear_before_tool_call_hooks()
print(f"Cleared {count} before hooks")

# Clear all tool hooks
before_count, after_count = clear_all_tool_call_hooks()
print(f"Cleared {before_count} before and {after_count} after hooks")

Listing Registered Hooks

python

from crewai.hooks import (
    get_before_tool_call_hooks,
    get_after_tool_call_hooks
)

# Get current hooks
before_hooks = get_before_tool_call_hooks()
after_hooks = get_after_tool_call_hooks()

print(f"Registered: {len(before_hooks)} before, {len(after_hooks)} after")

Advanced Patterns

Conditional Hook Execution

python

@before_tool_call
def conditional_blocking(context: ToolCallHookContext) -> bool | None:
    # Only block for specific agents
    if context.agent and context.agent.role == "junior_agent":
        if context.tool_name in ['delete_file', 'send_email']:
            print(f"❌ Junior agents cannot use {context.tool_name}")
            return False

    # Only block during specific tasks
    if context.task and "sensitive" in context.task.description.lower():
        if context.tool_name == 'web_search':
            print("❌ Web search blocked for sensitive tasks")
            return False

    return None

Context-Aware Input Modification

python

@before_tool_call
def enhance_tool_inputs(context: ToolCallHookContext) -> None:
    # Add context based on agent role
    if context.agent and context.agent.role == "researcher":
        if context.tool_name == 'web_search':
            # Add domain restrictions for researchers
            context.tool_input['domains'] = ['edu', 'gov', 'org']

    # Add context based on task
    if context.task and "urgent" in context.task.description.lower():
        if context.tool_name == 'send_email':
            context.tool_input['priority'] = 'high'

    return None

Tool Chain Monitoring

python

tool_call_chain = []

@before_tool_call
def track_tool_chain(context: ToolCallHookContext) -> None:
    tool_call_chain.append({
        'tool': context.tool_name,
        'timestamp': time.time(),
        'agent': context.agent.role if context.agent else 'Unknown'
    })

    # Detect potential infinite loops
    recent_calls = tool_call_chain[-5:]
    if len(recent_calls) == 5 and all(c['tool'] == context.tool_name for c in recent_calls):
        print(f"⚠️  Warning: {context.tool_name} called 5 times in a row")

    return None

Best Practices

Keep Hooks Focused: Each hook should have a single responsibility
Avoid Heavy Computation: Hooks execute on every tool call
Handle Errors Gracefully: Use try-except to prevent hook failures
Use Type Hints: Leverage ToolCallHookContext for better IDE support
Document Blocking Conditions: Make it clear when/why tools are blocked
Test Hooks Independently: Unit test hooks before using in production
Clear Hooks in Tests: Use clear_all_tool_call_hooks() between test runs
Modify In-Place: Always modify context.tool_input in-place, never replace
Log Important Decisions: Especially when blocking tool execution
Consider Performance: Cache expensive validations when possible

Error Handling

python

@before_tool_call
def safe_validation(context: ToolCallHookContext) -> bool | None:
    try:
        # Your validation logic
        if not validate_input(context.tool_input):
            return False
    except Exception as e:
        print(f"⚠️ Hook error: {e}")
        # Decide: allow or block on error
        return None  # Allow execution despite error

Type Safety

python

from crewai.hooks import ToolCallHookContext, BeforeToolCallHookType, AfterToolCallHookType

# Explicit type annotations
def my_before_hook(context: ToolCallHookContext) -> bool | None:
    return None

def my_after_hook(context: ToolCallHookContext) -> str | None:
    return None

# Type-safe registration
register_before_tool_call_hook(my_before_hook)
register_after_tool_call_hook(my_after_hook)

Integration with Existing Tools

Wrapping Existing Validation

python

def existing_validator(tool_name: str, inputs: dict) -> bool:
    """Your existing validation function."""
    # Your validation logic
    return True

@before_tool_call
def integrate_validator(context: ToolCallHookContext) -> bool | None:
    if not existing_validator(context.tool_name, context.tool_input):
        print(f"❌ Validation failed for {context.tool_name}")
        return False
    return None

Logging to External Systems

python

import logging

logger = logging.getLogger(__name__)

@before_tool_call
def log_to_external_system(context: ToolCallHookContext) -> None:
    logger.info(f"Tool call: {context.tool_name}", extra={
        'tool_name': context.tool_name,
        'tool_input': context.tool_input,
        'agent': context.agent.role if context.agent else None
    })
    return None

Troubleshooting

Hook Not Executing

Verify hook is registered before crew execution
Check if previous hook returned False (blocks execution and subsequent hooks)
Ensure hook signature matches expected type

Input Modifications Not Working

Use in-place modifications: context.tool_input['key'] = value
Don't replace the dict: context.tool_input = {}

Result Modifications Not Working

Return the modified string from after hooks
Returning None keeps the original result
Ensure the tool actually returned a result

Tool Blocked Unexpectedly

Check all before hooks for blocking conditions
Verify hook execution order
Add debug logging to identify which hook is blocking

Conclusion

Tool Call Hooks provide powerful capabilities for controlling and monitoring tool execution in CrewAI. Use them to implement safety guardrails, approval gates, input validation, result sanitization, logging, and analytics. Combined with proper error handling and type safety, hooks enable secure and production-ready agent systems with comprehensive observability.