工具调用钩子

了解如何使用工具调用钩子来拦截、修改和控制 CrewAI 中的工具执行

工具调用钩子在智能体运行期间为工具执行提供细粒度控制。这些钩子允许你拦截工具调用、修改输入、转换输出、实现安全检查，以及添加全面的日志记录或监控。

概述

工具钩子在两个拦截点执行：

拦截点	时机	钩子接收到的内容
PRE_TOOL_CALL	每次工具执行之前	ToolCallHookContext
POST_TOOL_CALL	每次工具执行之后	ToolCallHookContext（已设置结果）

使用 @on 装饰器 编写它们。旧版 @before_tool_call / @after_tool_call 装饰器 仍可照常使用——两种风格注册到同一引擎，并在同一条有序链中运行。

钩子签名

from crewai.hooks import on, HookAborted, InterceptionPoint, ToolCallHookContext

@on(InterceptionPoint.PRE_TOOL_CALL)
def before_hook(ctx: ToolCallHookContext) -> None:
    # Mutate ctx.tool_input in place, or
    # raise HookAborted(reason, source) to block the call
    ...

@on(InterceptionPoint.POST_TOOL_CALL)
def after_hook(ctx: ToolCallHookContext) -> str | None:
    # Return a string to replace ctx.tool_result
    # Return None to keep the original result
    ...

与边界点和步骤点不同，工具调用点直接将丰富的 ToolCallHookContext 作为钩子参数传入（没有单独的 ctx.payload）：在调用前就地修改 ctx.tool_input，并在调用后返回字符串以替换结果。

当调用被阻止时，工具不会运行，智能体收到 "Tool execution blocked by hook. Tool: <name>" 作为结果——运行继续进行。POST_TOOL_CALL 钩子在调用被阻止时仍会触发，因此监控钩子能看到每一次尝试。

工具钩子上下文

ToolCallHookContext 对象提供对工具执行状态的全面访问：

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           # Agent-facing result string (POST_TOOL_CALL only)
    raw_tool_result: Any | None       # Raw Python result (POST_TOOL_CALL only)

对于带类型的工具输出，tool_result 是智能体看到的字符串。默认情况下为 JSON。如果工具使用自定义格式，则可能是 Markdown 或其他字符串。当钩子需要类型化对象或字典时，请使用 raw_tool_result；它不受结果替换影响。

该上下文还提供 request_human_input(prompt, default_message)，它会暂停实时控制台更新并从终端收集输入——适用于审批关卡。

修改工具输入

重要： 始终就地修改工具输入：

# ✅ Correct - modify in-place
@on(InterceptionPoint.PRE_TOOL_CALL)
def sanitize_input(ctx: ToolCallHookContext) -> None:
    ctx.tool_input['query'] = ctx.tool_input['query'].lower()

# ❌ Wrong - replaces dict reference; the tool never sees it
@on(InterceptionPoint.PRE_TOOL_CALL)
def wrong_approach(ctx: ToolCallHookContext) -> None:
    ctx.tool_input = {'query': 'new query'}

注册方式

1. 全局钩子

应用于所有团队中的所有工具调用。使用 tools= / agents= 过滤器限定钩子作用范围：

from crewai.hooks import on, HookAborted, InterceptionPoint

@on(InterceptionPoint.PRE_TOOL_CALL)
def log_tool_call(ctx):
    print(f"Tool: {ctx.tool_name}, input: {ctx.tool_input}")

@on(InterceptionPoint.PRE_TOOL_CALL, tools=["delete_file", "drop_table"])
def block_destructive(ctx):
    raise HookAborted(reason=f"{ctx.tool_name} is not allowed", source="safety-policy")

@on(InterceptionPoint.POST_TOOL_CALL, tools=["web_search"], agents=["Researcher"])
def log_search_results(ctx):
    print(f"search returned {len(ctx.tool_result or '')} chars")

2. 团队作用域钩子

在 @CrewBase 类内部对方法应用相同的装饰器，即可将钩子限定为该团队范围：

from crewai.hooks import on, InterceptionPoint

@CrewBase
class MyProjCrew:
    @on(InterceptionPoint.PRE_TOOL_CALL)
    def validate_tool_inputs(self, ctx):
        # Only applies to this crew
        if ctx.tool_name == "web_search" and not ctx.tool_input.get("query"):
            raise HookAborted(reason="empty search query", source="input-validation")

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

常见用例

1. 安全护栏

@on(InterceptionPoint.PRE_TOOL_CALL)
def safety_check(ctx: ToolCallHookContext) -> None:
    destructive = {'delete_file', 'drop_table', 'remove_user', 'system_shutdown'}
    if ctx.tool_name in destructive:
        raise HookAborted(reason=f"{ctx.tool_name} is destructive", source="safety-policy")

2. 人工审批关卡

@on(InterceptionPoint.PRE_TOOL_CALL, tools=["send_email", "make_purchase", "delete_file"])
def require_approval(ctx: ToolCallHookContext) -> None:
    response = ctx.request_human_input(
        prompt=f"Approve {ctx.tool_name}?",
        default_message=f"Input: {ctx.tool_input}\nType 'yes' to approve:",
    )
    if response.lower() != 'yes':
        raise HookAborted(reason="denied by operator", source="approval-gate")

3. 输入校验与清理

@on(InterceptionPoint.PRE_TOOL_CALL, tools=["web_search"])
def validate_query(ctx: ToolCallHookContext) -> None:
    query = ctx.tool_input.get('query', '')
    if len(query) < 3:
        raise HookAborted(reason="search query too short", source="input-validation")
    ctx.tool_input['query'] = query.strip().lower()

@on(InterceptionPoint.PRE_TOOL_CALL, tools=["read_file"])
def validate_path(ctx: ToolCallHookContext) -> None:
    path = ctx.tool_input.get('path', '')
    if '..' in path or path.startswith('/'):
        raise HookAborted(reason="invalid file path", source="input-validation")

4. 结果清理

import re

@on(InterceptionPoint.POST_TOOL_CALL)
def sanitize_sensitive_data(ctx: ToolCallHookContext) -> str | None:
    if not ctx.tool_result:
        return None
    result = re.sub(
        r'(api[_-]?key|token)["\']?\s*[:=]\s*["\']?[\w-]+',
        r'\1: [REDACTED]',
        ctx.tool_result,
        flags=re.IGNORECASE,
    )
    return re.sub(
        r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        '[EMAIL-REDACTED]',
        result,
    )

5. 工具使用分析

import time
from collections import defaultdict

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

@on(InterceptionPoint.PRE_TOOL_CALL)
def start_timer(ctx: ToolCallHookContext) -> None:
    ctx.tool_input['_start_time'] = time.time()

@on(InterceptionPoint.POST_TOOL_CALL)
def track_tool_usage(ctx: ToolCallHookContext) -> None:
    start_time = ctx.tool_input.pop('_start_time', time.time())
    tool_stats[ctx.tool_name]['count'] += 1
    tool_stats[ctx.tool_name]['total_time'] += time.time() - start_time

6. 限流

from collections import defaultdict
from datetime import datetime, timedelta

tool_call_history = defaultdict(list)

@on(InterceptionPoint.PRE_TOOL_CALL)
def rate_limit_tools(ctx: ToolCallHookContext) -> None:
    now = datetime.now()
    history = tool_call_history[ctx.tool_name]
    history[:] = [t for t in history if now - t < timedelta(minutes=1)]
    if len(history) >= 10:
        raise HookAborted(reason=f"rate limit exceeded for {ctx.tool_name}",
                          source="rate-limiter")
    history.append(now)

钩子管理

from crewai.hooks import (
    InterceptionPoint,
    clear_all_hooks,
    clear_hooks,
    get_hooks,
    unregister_hook,
)

# Unregister a specific hook
unregister_hook(InterceptionPoint.PRE_TOOL_CALL, my_hook)

# Clear one point, or everything (e.g. between tests)
clear_hooks(InterceptionPoint.POST_TOOL_CALL)
clear_all_hooks()

# Inspect what's registered
print(len(get_hooks(InterceptionPoint.PRE_TOOL_CALL)))

旧版管理 API（register_before_tool_call_hook、unregister_before_tool_call_hook、clear_before_tool_call_hooks、clear_all_tool_call_hooks、get_before_tool_call_hooks 及其对应的 after_ 版本）操作相同的底层注册表，因此任一 API 都能管理另一 API 注册的钩子。

旧版装饰器

原有的按拦截点装饰器照常工作，并与 @on 钩子在同一注册顺序链中运行：

from crewai.hooks import before_tool_call, after_tool_call

@before_tool_call
def block_dangerous_tools(context):
    if context.tool_name in ('delete_database', 'drop_table'):
        return False  # Block execution
    return None

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

与 @on 的区别：

• 阻止调用是在前置钩子中 return False——等价于抛出 HookAborted，但无需为遥测提供自定义原因或来源。智能体 看到的仍是相同的 "Tool execution blocked by hook" 消息。
• 签名与拦截点相关：前置钩子返回 bool | None，后置 钩子返回 str | None。上下文对象同样是 ToolCallHookContext。
• 过滤器和团队作用域的工作方式相同： @before_tool_call(tools=[...], agents=[...])，将装饰器应用于 @CrewBase 方法即可将其限定为该团队。

新代码优先使用 @on；已在使用的旧版风格可继续保留——行为上没有差别。

最佳实践

1. 保持钩子专注且快速——它们会在每次工具调用时运行
2. 就地修改——始终修改 ctx.tool_input，不要替换整个 dict
3. 优先使用过滤器而非条件判断——tools= / agents= 可让钩子主体保持精简
4. 显式中止——抛出带有明确原因和来源的 HookAborted； 其他任何异常都会被吞掉（fail-open）
5. 使用类型提示——用 ToolCallHookContext 标注以获得 IDE 支持
6. 在测试中清理钩子——在测试运行之间调用 clear_all_hooks()

故障排查

钩子未执行

• 确认钩子在团队执行之前已注册
• 检查是否有更早的钩子阻止了调用（后续前置钩子不会运行）
• 对照实际工具名称和智能体角色检查 tools= / agents= 过滤器

输入修改不生效

• 使用就地修改：ctx.tool_input['key'] = value
• 不要替换整个 dict：ctx.tool_input="true"

结果修改不生效

• 从 POST_TOOL_CALL 钩子返回修改后的字符串
• 返回 None 则保留原始结果

工具被意外阻止

• 检查所有前置钩子中的 HookAborted / return False 条件
• 中止原因和来源会出现在 HookDispatchedEvent 遥测中

相关文档

• 执行钩子概览 →
• LLM 调用钩子 →
• 执行边界钩子 →
• 步骤钩子 →