도구 호출 훅(Tool Call Hooks)

에이전트가 도구를 부르기 직전이나 직후에 무언가 끼어들고 싶을 때가 있습니다. 예컨대 위험한 파일 삭제 호출을 막거나, 검색 입력을 소문자로 정리하거나, 민감 정보가 결과에 섞여 나오지 않게 치고, 도구 사용량을 기록하는 식이죠. CrewAI의 도구 호출 훅이 바로 그 자리를 잡아 줍니다. 이 글에서는 @on 데코레이터로 훅을 등록하고 제어하는 방법을 정리합니다.

출처: 공식문서

본문

도구 호출 훅(Tool Call Hooks)은 에이전트 작업 중 도구 실행을 세밀하게 제어할 수 있게 해 줍니다. 이 훅들로 도구 호출을 가로채고, 입력을 수정하고, 출력을 변환하고, 안전 검사를 구현하고, 포괄적인 로깅·모니터링을 추가할 수 있어요.

개요

도구 훅은 두 개의 가로채기 시점(interception point)에서 실행됩니다.

시점	시기	훅이 받는 것
`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)를 노출하는데, 라이브 콘솔 업데이트를 멈추고 터미널에서 입력을 수집합니다 — 승인 게이트에 유용합니다.

도구 입력 수정하기

중요: 도구 입력은 항상 제자리에서(in-place) 수정하세요.

# ✅ 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. 전역 훅(Global Hooks)

모든 크루의 모든 도구 호출에 적용됩니다. 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. 크루 범위 훅(Crew-Scoped Hooks)

@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. 안전 가드레일(Safety Guardrails)

@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. 인간 승인 게이트(Human Approval Gate)

@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. 입력 검증·정화(Input Validation and Sanitization)

@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. 결과 정화(Result Sanitization)

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. 도구 사용 분석(Tool Usage Analytics)

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. 속도 제한(Rate Limiting)

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)

훅 관리(Hook Management)

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과의 차이점:

차단은 before 훅에서 return False로 합니다 — HookAborted를 던지는 것과 같지만, 텔레메트리용 커스텀 이유·소스가 없습니다. 에이전트는 같은 "Tool execution blocked by hook" 메시지를 봅니다.
시그니처는 시점별로 다릅니다: before 훅은 bool | None을, after 훅은 str | None을 반환합니다. 컨텍스트 객체는 같은 ToolCallHookContext예요.
필터와 크루 범위는 같은 방식으로 동작합니다: @before_tool_call(tools=[...], agents=[...]), 그리고 @CrewBase 메서드에 데코레이터를 적용하면 해당 크루로 범위가 한정됩니다.

새 코드에서는 @on을 선호하고, 이미 쓰는 곳에서는 레거시 스타일을 유지하세요 — 동작상 패널티는 없습니다.

모범 사례

훅을 가볍고 빠르게 유지 — 매 도구 호출마다 실행됩니다.
제자리 수정 — 항상 ctx.tool_input을 변경하고, 딕셔너리를 교체하지 마세요.
조건문보다 필터 선호 — tools= / agents=로 훅 본문을 작게 유지하세요.
크게 중단 — 의미 있는 이유와 소스를 담아 HookAborted를 던지세요. 다른 예외는 삼켜집니다(fail-open).
타입 힌트 사용 — IDE 지원을 위해 ToolCallHookContext로 어노테이션하세요.
테스트에서 훅 정리 — 테스트 실행 사이에 clear_all_hooks()를 호출하세요.

문제 해결

훅이 실행되지 않음

크루 실행 전에 훅이 등록됐는지 확인하세요.
이전 훅이 호출을 차단했는지 확인하세요(이후 pre 훅은 실행되지 않습니다).
tools= / agents= 필터를 실제 도구 이름과 에이전트 역할에 맞춰 확인하세요.

입력 수정이 동작하지 않음

제자리 수정을 쓰세요: ctx.tool_input['key'] = value
딕셔너리를 교체하지 마세요: ctx.tool_input = {}

결과 수정이 동작하지 않음

POST_TOOL_CALL 훅에서 수정된 문자열을 반환하세요.
None을 반환하면 원래 결과가 유지됩니다.

도구가 예상치 못하게 차단됨

모든 pre 훅에서 HookAborted / return False 조건을 확인하세요.
중단 이유와 소스는 HookDispatchedEvent 텔레메트리에 나타납니다.

도구 호출 훅(Tool Call Hooks)