
5편에서 can_use_tool 콜백으로 도구 호출을 허용하거나 거부하는 방법을 다뤘습니다. 하지만 이건 "문 앞의 경비원"에 가깝습니다 — 들어오는 것만 제어할 뿐, 안에서 무슨 일이 일어나는지는 모릅니다.
훅(Hook) 시스템은 다릅니다. 에이전트의 동작 전후에 이벤트 리스너를 달아서:
이 글에서는 훅 시스템의 전체 구조를 분석하고, 실전 패턴을 정리합니다.
can_use_tool (5편) | 훅 시스템 (이 글) | |
|---|---|---|
| 역할 | 도구 사용 허용/거부 | 동작 관찰/개입/로깅 |
| 시점 | 도구 실행 전만 | 실행 전, 후, 실패, 종료 등 10가지 시점 |
| 반환 | Allow / Deny | 실행 계속/중단, 입력 수정, 컨텍스트 추가 등 |
| 적용 대상 | 도구 호출만 | 프롬프트 제출, 에이전트 시작/종료, 컨텍스트 압축 등 |
| 비유 | 문 앞의 경비원 | 건물 전체의 CCTV + 방송 시스템 |
둘을 함께 사용할 수 있습니다. can_use_tool로 거부/허용을 결정하고, 훅으로 모든 과정을 로깅하는 식입니다.
사용자 프롬프트 제출
│
├── UserPromptSubmit ──────── 프롬프트가 Claude에 전달되기 전
│
▼
Claude가 응답 생성
│
├── PermissionRequest ─────── 권한 확인이 필요할 때
│
├── PreToolUse ────────────── 도구 실행 직전
│ │
│ ├── (도구 실행 성공)
│ │ └── PostToolUse ─── 도구 실행 직후
│ │
│ └── (도구 실행 실패)
│ └── PostToolUseFailure ─── 도구 실행 실패 시
│
├── SubagentStart ─────────── 서브 에이전트 시작
│ └── SubagentStop ────── 서브 에이전트 종료
│
├── PreCompact ────────────── 컨텍스트 압축 직전
│
├── Notification ──────────── 알림 발생 시
│
└── Stop ──────────────────── 에이전트 종료
CLI가 훅 이벤트 발생
│
▼
CLI → SDK: control_request (hook_callback)
│ callback_id, input, tool_use_id 전달
│
▼
SDK: HookMatcher에서 매칭되는 콜백 함수 찾기
│
▼
SDK: 콜백 함수 실행 (await)
│
▼
SDK → CLI: control_response (HookJSONOutput)
│ continue_, hookSpecificOutput 등 반환
│
▼
CLI: 훅 결과에 따라 동작 결정
(계속 실행 / 중단 / 입력 수정 등)
@dataclass
class HookMatcher:
matcher: str | None = None # 매칭 패턴 (None이면 모든 이벤트)
hooks: list[HookCallback] = [] # 콜백 함수 리스트
timeout: float | None = None # 타임아웃 (초, 기본 60초)
| 패턴 | 의미 |
|---|---|
None | 해당 이벤트의 모든 호출에 매칭 |
"Bash" | Bash 도구에만 매칭 |
"Write\|Edit" | Write 또는 Edit 도구에 매칭 |
"Read" | Read 도구에만 매칭 |
matcher는 도구 이름 기반이므로, PreToolUse/PostToolUse/PostToolUseFailure 이벤트에서만 의미가 있습니다. 다른 이벤트(Stop, UserPromptSubmit 등)에서는 None을 사용합니다.
options = ClaudeAgentOptions(
hooks={
"PreToolUse": [
HookMatcher(
matcher="Bash", # Bash 도구에만
hooks=[my_bash_guard], # 이 함수를 실행
timeout=10, # 10초 타임아웃
),
HookMatcher(
matcher=None, # 모든 도구에
hooks=[my_logger], # 이 함수를 실행
),
],
"PostToolUse": [
HookMatcher(
matcher=None,
hooks=[my_result_logger],
),
],
"Stop": [
HookMatcher(
matcher=None,
hooks=[my_cleanup],
),
],
}
)
HookCallback = Callable[
[HookInput, str | None, HookContext],
Awaitable[HookJSONOutput]
]
| 인자 | 타입 | 설명 |
|---|---|---|
input | HookInput | 이벤트별 입력 데이터 |
tool_use_id | str \| None | 도구 호출 ID (도구 관련 이벤트에서만) |
context | HookContext | 컨텍스트 (현재 signal: None 고정) |
async def my_hook(
input: HookInput,
tool_use_id: str | None,
context: HookContext,
) -> HookJSONOutput:
# input의 타입은 이벤트에 따라 다름
print(f"이벤트: {input['hook_event_name']}")
return {} # 빈 dict = 기본 동작 (계속 진행)
모든 입력에는 공통 필드가 있습니다:
class BaseHookInput(TypedDict):
session_id: str # 세션 ID
transcript_path: str # 대화 기록 파일 경로
cwd: str # 작업 디렉토리
permission_mode: str # 현재 권한 모드 (선택적)
class PreToolUseHookInput(BaseHookInput):
hook_event_name: Literal["PreToolUse"]
tool_name: str # 도구 이름
tool_input: dict[str, Any] # 도구 입력값
tool_use_id: str # 도구 호출 고유 ID
agent_id: str # 서브 에이전트 ID (선택적)
agent_type: str # 에이전트 타입 (선택적)
class PostToolUseHookInput(BaseHookInput):
hook_event_name: Literal["PostToolUse"]
tool_name: str
tool_input: dict[str, Any]
tool_response: Any # 도구 실행 결과
tool_use_id: str
agent_id: str # (선택적)
agent_type: str # (선택적)
class PostToolUseFailureHookInput(BaseHookInput):
hook_event_name: Literal["PostToolUseFailure"]
tool_name: str
tool_input: dict[str, Any]
tool_use_id: str
error: str # 에러 메시지
is_interrupt: bool # 인터럽트 여부 (선택적)
class UserPromptSubmitHookInput(BaseHookInput):
hook_event_name: Literal["UserPromptSubmit"]
prompt: str # 사용자 프롬프트
class StopHookInput(BaseHookInput):
hook_event_name: Literal["Stop"]
stop_hook_active: bool # 다른 Stop 훅이 실행 중인지
class SubagentStartHookInput(BaseHookInput):
hook_event_name: Literal["SubagentStart"]
agent_id: str # 서브 에이전트 ID
agent_type: str # 에이전트 타입
class SubagentStopHookInput(BaseHookInput):
hook_event_name: Literal["SubagentStop"]
stop_hook_active: bool
agent_id: str
agent_transcript_path: str # 서브 에이전트 대화 기록
agent_type: str
class PreCompactHookInput(BaseHookInput):
hook_event_name: Literal["PreCompact"]
trigger: Literal["manual", "auto"] # 수동/자동 압축
custom_instructions: str | None # 압축 시 커스텀 지시
class NotificationHookInput(BaseHookInput):
hook_event_name: Literal["Notification"]
message: str # 알림 내용
title: str # 제목 (선택적)
notification_type: str # 알림 타입
class PermissionRequestHookInput(BaseHookInput):
hook_event_name: Literal["PermissionRequest"]
tool_name: str
tool_input: dict[str, Any]
permission_suggestions: list[Any] # (선택적)
훅 함수가 반환하는 값으로, 에이전트의 동작을 제어합니다. 동기(Sync)와 비동기(Async) 두 가지 모드가 있습니다.
class SyncHookJSONOutput(TypedDict):
# 실행 제어
continue_: bool # False면 실행 중단 (기본 True)
suppressOutput: bool # stdout 숨김 (기본 False)
stopReason: str # continue_=False일 때 중단 사유
# 차단 결정
decision: Literal["block"] # "block"으로 차단
systemMessage: str # 사용자에게 표시할 메시지
reason: str # Claude에게 전달할 사유
# 이벤트별 세부 제어
hookSpecificOutput: HookSpecificOutput
주의: Python에서 continue와 async는 예약어입니다. SDK에서는 언더스코어 버전 (continue_, async_)을 사용하고, CLI에 전송할 때 자동으로 변환됩니다.
class AsyncHookJSONOutput(TypedDict):
async_: Literal[True] # 비동기 실행 표시
asyncTimeout: int # 타임아웃 (ms, 선택적)
비동기 훅은 "이 작업은 시간이 걸리니 나중에 결과를 확인해"라고 CLI에 알립니다. 외부 API 호출이나 승인 대기 같은 상황에서 사용합니다.
class PreToolUseHookSpecificOutput(TypedDict):
hookEventName: Literal["PreToolUse"]
permissionDecision: Literal["allow", "deny", "ask"] # 허용/거부/확인
permissionDecisionReason: str # 결정 사유
updatedInput: dict[str, Any] # 수정된 입력값
additionalContext: str # Claude에게 추가 컨텍스트
이것이 훅 시스템의 가장 강력한 기능입니다:
permissionDecision: "deny" — 도구 실행을 차단updatedInput — 도구 입력값을 수정additionalContext — Claude에게 추가 정보 제공 ("이 파일은 프로덕션 코드입니다. 신중하게 수정하세요.")class PostToolUseHookSpecificOutput(TypedDict):
hookEventName: Literal["PostToolUse"]
additionalContext: str # Claude에게 추가 컨텍스트
updatedMCPToolOutput: Any # MCP 도구 결과 수정
도구 실행 결과를 보고 Claude에게 추가 정보를 제공할 수 있습니다.
대부분의 이벤트는 additionalContext만 지원합니다:
# PostToolUseFailure, UserPromptSubmit, Notification, SubagentStart 등
hookSpecificOutput: {
"hookEventName": "PostToolUseFailure",
"additionalContext": "이 에러는 알려진 이슈입니다. 무시해도 됩니다.",
}
class PermissionRequestHookSpecificOutput(TypedDict):
hookEventName: Literal["PermissionRequest"]
decision: dict[str, Any] # 권한 결정 (allow/deny 등)
모든 도구 호출의 전후를 기록합니다.
import json
from datetime import datetime
log_entries = []
async def pre_tool_logger(input, tool_use_id, context):
entry = {
"time": datetime.now().isoformat(),
"event": "pre",
"tool": input["tool_name"],
"input_keys": list(input["tool_input"].keys()),
"tool_use_id": tool_use_id,
}
log_entries.append(entry)
print(f"[PRE] {input['tool_name']}: {list(input['tool_input'].keys())}")
return {}
async def post_tool_logger(input, tool_use_id, context):
response = input.get("tool_response")
response_preview = str(response)[:100] if response else "N/A"
entry = {
"time": datetime.now().isoformat(),
"event": "post",
"tool": input["tool_name"],
"response_preview": response_preview,
}
log_entries.append(entry)
print(f"[POST] {input['tool_name']}: {response_preview}")
return {}
async def failure_logger(input, tool_use_id, context):
print(f"[FAIL] {input['tool_name']}: {input['error']}")
return {}
options = ClaudeAgentOptions(
hooks={
"PreToolUse": [HookMatcher(matcher=None, hooks=[pre_tool_logger])],
"PostToolUse": [HookMatcher(matcher=None, hooks=[post_tool_logger])],
"PostToolUseFailure": [HookMatcher(matcher=None, hooks=[failure_logger])],
}
)
PreToolUse 훅에서 permissionDecision: "deny"로 위험한 명령을 차단합니다.
BLOCKED_PATTERNS = ["rm -rf", "mkfs", "dd if=", "> /dev/"]
async def bash_guard(input, tool_use_id, context):
if input["hook_event_name"] != "PreToolUse":
return {}
tool_name = input["tool_name"]
if tool_name != "Bash":
return {}
command = input["tool_input"].get("command", "")
for pattern in BLOCKED_PATTERNS:
if pattern in command:
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": f"보안 정책 위반: {pattern}",
}
}
return {}
options = ClaudeAgentOptions(
hooks={
"PreToolUse": [
HookMatcher(matcher="Bash", hooks=[bash_guard]),
],
}
)
PreToolUse 훅에서 updatedInput으로 파일 경로를 workspace 안으로 리다이렉트합니다.
WORKSPACE = "/app/workspace"
async def path_rewriter(input, tool_use_id, context):
tool_name = input["tool_name"]
tool_input = input["tool_input"]
path_field = {"Read": "file_path", "Write": "file_path", "Edit": "file_path"}.get(tool_name)
if path_field and path_field in tool_input:
original = tool_input[path_field]
if not original.startswith(WORKSPACE):
safe_path = f"{WORKSPACE}/{original.lstrip('/')}"
print(f"[경로 변환] {original} → {safe_path}")
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"updatedInput": {**tool_input, path_field: safe_path},
}
}
return {}
options = ClaudeAgentOptions(
hooks={
"PreToolUse": [
HookMatcher(matcher="Read|Write|Edit", hooks=[path_rewriter]),
],
}
)
도구 실행 전에 Claude가 알아야 할 정보를 additionalContext로 전달합니다.
PROTECTED_FILES = {
"main.py": "이 파일은 프로덕션 서버의 진입점입니다. 수정 시 서비스 중단 가능성이 있습니다.",
".env": "이 파일에는 시크릿이 포함되어 있습니다. 절대 읽거나 수정하지 마세요.",
}
async def context_injector(input, tool_use_id, context):
tool_input = input["tool_input"]
file_path = tool_input.get("file_path", "")
for protected, warning in PROTECTED_FILES.items():
if file_path.endswith(protected):
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"additionalContext": f"[경고] {warning}",
}
}
return {}
options = ClaudeAgentOptions(
hooks={
"PreToolUse": [
HookMatcher(matcher="Read|Write|Edit", hooks=[context_injector]),
],
}
)
특정 조건에서 에이전트 실행을 완전히 중단합니다.
MAX_TOOL_CALLS = 50
tool_call_count = 0
async def rate_limiter(input, tool_use_id, context):
global tool_call_count
tool_call_count += 1
if tool_call_count > MAX_TOOL_CALLS:
return {
"continue_": False,
"stopReason": f"도구 호출 횟수가 {MAX_TOOL_CALLS}회를 초과했습니다.",
}
return {}
async def cleanup_on_stop(input, tool_use_id, context):
print(f"[Stop] 에이전트 종료. 세션: {input['session_id']}")
print(f"[Stop] 작업 디렉토리: {input['cwd']}")
# 임시 파일 정리, 로그 저장, 알림 전송 등
if log_entries:
import json
with open("/tmp/claude-audit-log.json", "w") as f:
json.dump(log_entries, f, ensure_ascii=False, indent=2)
print(f"[Stop] 감사 로그 저장: {len(log_entries)}건")
return {}
options = ClaudeAgentOptions(
hooks={
"Stop": [HookMatcher(matcher=None, hooks=[cleanup_on_stop])],
}
)
active_agents = {}
async def track_agent_start(input, tool_use_id, context):
agent_id = input["agent_id"]
agent_type = input["agent_type"]
active_agents[agent_id] = agent_type
print(f"[Agent 시작] {agent_type} (id={agent_id})")
return {}
async def track_agent_stop(input, tool_use_id, context):
agent_id = input["agent_id"]
agent_type = input["agent_type"]
active_agents.pop(agent_id, None)
print(f"[Agent 종료] {agent_type} (id={agent_id})")
print(f" 기록: {input['agent_transcript_path']}")
return {}
options = ClaudeAgentOptions(
hooks={
"SubagentStart": [HookMatcher(matcher=None, hooks=[track_agent_start])],
"SubagentStop": [HookMatcher(matcher=None, hooks=[track_agent_stop])],
}
)
import asyncio
from claude_agent_sdk import (
ClaudeSDKClient,
ClaudeAgentOptions,
HookMatcher,
AssistantMessage,
ResultMessage,
TextBlock,
)
# ── 훅 함수들 ──
async def log_all_tools(input, tool_use_id, context):
"""모든 도구 호출 로깅"""
print(f" [LOG] {input['tool_name']} → {list(input['tool_input'].keys())}")
return {}
async def guard_bash(input, tool_use_id, context):
"""위험한 Bash 명령 차단"""
command = input["tool_input"].get("command", "")
if "rm -rf" in command:
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": "rm -rf 는 금지입니다.",
}
}
return {}
async def log_results(input, tool_use_id, context):
"""도구 실행 결과 로깅"""
response = str(input.get("tool_response", ""))[:80]
print(f" [RESULT] {input['tool_name']} → {response}")
return {}
async def on_stop(input, tool_use_id, context):
"""에이전트 종료 시 실행"""
print(f" [STOP] 세션 종료: {input['session_id']}")
return {}
# ── 메인 ──
async def main():
options = ClaudeAgentOptions(
permission_mode="bypassPermissions",
allowed_tools=["Read", "Write", "Bash", "Glob"],
max_turns=10,
hooks={
"PreToolUse": [
HookMatcher(matcher="Bash", hooks=[guard_bash]),
HookMatcher(matcher=None, hooks=[log_all_tools]),
],
"PostToolUse": [
HookMatcher(matcher=None, hooks=[log_results]),
],
"Stop": [
HookMatcher(matcher=None, hooks=[on_stop]),
],
},
)
async with ClaudeSDKClient(options=options) as client:
await client.query(
"현재 디렉토리의 파일 목록을 확인하고, "
"hook_test.txt 파일을 만들어줘."
)
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
elif isinstance(message, ResultMessage):
print(f"\n완료: {message.num_turns}턴, ${message.total_cost_usd or 0:.4f}")
if __name__ == "__main__":
asyncio.run(main())
| 필드 | 타입 | 설명 | 기본값 |
|---|---|---|---|
continue_ | bool | False면 실행 중단 | True |
stopReason | str | 중단 시 표시할 사유 | — |
suppressOutput | bool | stdout 출력 숨김 | False |
decision | "block" | 차단 결정 | — |
systemMessage | str | 사용자에게 표시할 메시지 | — |
reason | str | Claude에게 전달할 사유 | — |
hookSpecificOutput | dict | 이벤트별 세부 제어 | — |
async_ | True | 비동기 실행 표시 | — |
asyncTimeout | int | 비동기 타임아웃 (ms) | — |
훅 시스템의 핵심은 3가지 능력입니다:
PostToolUse로 도구 실행 결과를 로깅PreToolUse의 permissionDecision/updatedInput으로 실행 전 제어continue_: False로 에이전트 실행을 완전히 멈춤5편의 can_use_tool과 함께 사용하면, 프로덕션 환경에서 Claude Agent를 안전하고 투명하게 운영할 수 있습니다.
전체 소스 코드는 GitHub에서 확인할 수 있습니다:
👉 GitHub 레포지토리 링크