훅으로 에이전트 동작 가로채기 및 제어

import asyncio
from claude_agent_sdk import (
    AssistantMessage,
    ClaudeSDKClient,
    ClaudeAgentOptions,
    HookMatcher,
    ResultMessage,
)


# 도구 호출 세부 정보를 받는 훅 콜백을 정의합니다
async def protect_env_files(input_data, tool_use_id, context):
    # 도구의 입력 인수에서 파일 경로를 추출합니다
    file_path = input_data["tool_input"].get("file_path", "")
    file_name = file_path.split("/")[-1]

    # .env 파일을 대상으로 하는 경우 작업을 차단합니다
    if file_name == ".env":
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": "Cannot modify .env files",
            }
        }

    # 빈 객체를 반환하여 작업을 허용합니다
    return {}


async def main():
    options = ClaudeAgentOptions(
        hooks={
            # PreToolUse 이벤트에 대한 훅을 등록합니다
            # 매처는 Write 및 Edit 도구 호출만 필터링합니다
            "PreToolUse": [HookMatcher(matcher="Write|Edit", hooks=[protect_env_files])]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Update the database configuration")
        async for message in client.receive_response():
            # 어시스턴트 및 결과 메시지를 필터링합니다
            if isinstance(message, (AssistantMessage, ResultMessage)):
                print(message)


asyncio.run(main())

options = ClaudeAgentOptions(
    hooks={"PreToolUse": [HookMatcher(matcher="Bash", hooks=[my_callback])]}
)

async with ClaudeSDKClient(options=options) as client:
    await client.query("Your prompt")
    async for message in client.receive_response():
        print(message)

async def async_hook(input_data, tool_use_id, context):
    # 백그라운드 작업을 시작한 후 즉시 반환합니다
    asyncio.create_task(send_to_logging_service(input_data))
    return {"async_": True, "asyncTimeout": 30000}

async def redirect_to_sandbox(input_data, tool_use_id, context):
    if input_data["hook_event_name"] != "PreToolUse":
        return {}

    if input_data["tool_name"] == "Write":
        original_path = input_data["tool_input"].get("file_path", "")
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "allow",
                "updatedInput": {
                    **input_data["tool_input"],
                    "file_path": f"/sandbox{original_path}",
                },
            }
        }
    return {}

async def block_etc_writes(input_data, tool_use_id, context):
    file_path = input_data["tool_input"].get("file_path", "")

    if file_path.startswith("/etc"):
        return {
            # 최상위 필드: 사용자에게 표시되는 메시지
            "systemMessage": "Remember: system directories like /etc are protected.",
            # hookSpecificOutput: 작업을 차단합니다
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": "Writing to /etc is not allowed",
            },
        }
    return {}

async def auto_approve_read_only(input_data, tool_use_id, context):
    if input_data["hook_event_name"] != "PreToolUse":
        return {}

    read_only_tools = ["Read", "Glob", "Grep"]
    if input_data["tool_name"] in read_only_tools:
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "allow",
                "permissionDecisionReason": "Read-only tool auto-approved",
            }
        }
    return {}

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(hooks=[authorization_check]),
            HookMatcher(hooks=[input_validator]),
            HookMatcher(hooks=[audit_logger]),
        ]
    }
)

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            # 파일 수정 도구 일치
            HookMatcher(matcher="Write|Edit|Delete", hooks=[file_security_hook]),
            # 모든 MCP 도구 일치
            HookMatcher(matcher="^mcp__", hooks=[mcp_audit_hook]),
            # 모든 것 일치(매처 없음)
            HookMatcher(hooks=[global_logger]),
        ]
    }
)

async def subagent_tracker(input_data, tool_use_id, context):
    # 서브에이전트가 완료될 때 세부 정보를 기록합니다
    print(f"[SUBAGENT] Completed: {input_data['agent_id']}")
    print(f"  Transcript: {input_data['agent_transcript_path']}")
    print(f"  Tool use ID: {tool_use_id}")
    print(f"  Stop hook active: {input_data.get('stop_hook_active')}")
    return {}


options = ClaudeAgentOptions(
    hooks={"SubagentStop": [HookMatcher(hooks=[subagent_tracker])]}
)

import asyncio
import json
import urllib.request
from datetime import datetime


def _send_webhook(tool_name):
    """도구 사용 데이터를 외부 웹훅에 POST하는 동기 헬퍼입니다."""
    data = json.dumps(
        {
            "tool": tool_name,
            "timestamp": datetime.now().isoformat(),
        }
    ).encode()
    req = urllib.request.Request(
        "https://api.example.com/webhook",
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    urllib.request.urlopen(req)


async def webhook_notifier(input_data, tool_use_id, context):
    # 도구가 완료된 후(PostToolUse)에만 발생하고, 전에는 아닙니다
    if input_data["hook_event_name"] != "PostToolUse":
        return {}

    try:
        # 이벤트 루프를 차단하지 않도록 스레드에서 차단 HTTP 호출을 실행합니다
        await asyncio.to_thread(_send_webhook, input_data["tool_name"])
    except Exception as e:
        # 오류를 기록하지만 발생시키지 않습니다. 실패한 웹훅이 에이전트를 중단해서는 안 됩니다
        print(f"Webhook request failed: {e}")

    return {}

import asyncio
import json
import urllib.request

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher


def _send_slack_notification(message):
    """수신 웹훅을 통해 Slack에 메시지를 보내는 동기 헬퍼입니다."""
    data = json.dumps({"text": f"Agent status: {message}"}).encode()
    req = urllib.request.Request(
        "https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    urllib.request.urlopen(req)


async def notification_handler(input_data, tool_use_id, context):
    try:
        # 이벤트 루프를 차단하지 않도록 스레드에서 차단 HTTP 호출을 실행합니다
        await asyncio.to_thread(_send_slack_notification, input_data.get("message", ""))
    except Exception as e:
        print(f"Failed to send notification: {e}")

    # 빈 객체를 반환합니다. 알림 훅은 에이전트 동작을 수정하지 않습니다
    return {}


async def main():
    options = ClaudeAgentOptions(
        hooks={
            # 알림 이벤트에 대한 훅을 등록합니다(매처 필요 없음)
            "Notification": [HookMatcher(hooks=[notification_handler])],
        },
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Analyze this codebase")
        async for message in client.receive_response():
            print(message)


asyncio.run(main())

options = ClaudeAgentOptions(
    setting_sources=["project"],  # .claude/settings.json을 로드합니다(훅 포함)
)