Control and observability

Intercept and control agent behavior with hooks

Intercept and control agent behavior with hooks 这一页讲的，就是 Intercept and control agent behavior with hooks 这件事在 Claude Code 里到底怎么用。

页面信息

对应原页

页面性质

第三方中文解释页

使用建议

先看人话解释，再对照原页命令和代码

这页不是官方原文，而是顺着官方文档结构做的中文解释版。命令、参数、配置名这些硬东西尽量保留，解释部分则尽量讲成人能照着做的话。

如果你碰到特别敏感的配置、权限或企业环境差异，最好顺手点上面的“查看原始文档”再核一遍。

这一页先讲明白

这页主要讲 Intercept and control agent behavior with hooks：Intercept and customize agent behavior at key execution points with hooks

你可以把它当成"Control and observability"这块里专门管这一摊事的说明书。

你可以把"Intercept and control agent behavior with hooks"理解成 Control and observability 这一栏里的一把专门工具。这页不是让你背书，而是教你什么时候该把这把工具拿出来。

原文这页大多会按 How hooks work、Available hooks、Configure hooks、Matchers 这些环节往下讲。

翻成人话，大概就是：Callback functions

第一，先别一上来全开全配。先按最小一步试通，确认没跑偏，再继续往下加。

第二，命令、配置名、参数名这些硬东西尽量保留原样。人话解释是帮你听懂，不是帮你改关键字。

第三，照着原文这几个环节挨个过：How hooks work -> Available hooks -> Configure hooks -> Matchers。像下地先看水路、再试机器、再正式开干，一步一步最稳。

关键片段

原页关键片段：How hooks work

"How hooks work"这一段里最要紧的原始写法在下面，先看它怎么落地。

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


# Define a hook callback that receives tool call details
async def protect_env_files(input_data, tool_use_id, context):
    # Extract the file path from the tool's input arguments
    file_path = input_data["tool_input"].get("file_path", "")
    file_name = file_path.split("/")[-1]

    # Block the operation if targeting a .env file
    if file_name == ".env":
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": "Cannot modify .env files",
            }
        }

    # Return empty object to allow the operation
    return {}


async def main():
    options = ClaudeAgentOptions(
        hooks={
            # Register the hook for PreToolUse events
            # The matcher filters to only Write and Edit tool calls
            "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():
            # Filter for assistant and result messages
            if isinstance(message, (AssistantMessage, ResultMessage)):
                print(message)


asyncio.run(main())

关键片段

原页关键片段：Configure hooks

下面这块是这一段最值钱的原文样板，先对着看一眼。

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)

关键片段

原页关键片段：Callback functions

下面这块是这一段最值钱的原文样板，先对着看一眼。

async def async_hook(input_data, tool_use_id, context):
    # Start a background task, then return immediately
    asyncio.create_task(send_to_logging_service(input_data))
    return {"async_": True, "asyncTimeout": 30000}

关键片段

原页关键片段：Modify tool input

下面这块是这一段最值钱的原文样板，先对着看一眼。

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 {}

关键片段

原页关键片段：Add context and block a tool

先看下面这块原始片段，等会儿再回头看解释会顺得多。

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 {
            # Top-level field: inject guidance into the conversation
            "systemMessage": "Remember: system directories like /etc are protected.",
            # hookSpecificOutput: block the operation
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": "Writing to /etc is not allowed",
            },
        }
    return {}

关键片段

原页关键片段：Auto-approve specific tools

下面这块是这一段最值钱的原文样板，先对着看一眼。

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 {}

关键片段

原页关键片段：Register multiple hooks

"Register multiple hooks"这一段里最要紧的原始写法在下面，先看它怎么落地。

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

关键片段

原页关键片段：Filter with regex matchers

这一段要真抓重点，通常就抓下面这块原文。

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            # Match file modification tools
            HookMatcher(matcher="Write|Edit|Delete", hooks=[file_security_hook]),
            # Match all MCP tools
            HookMatcher(matcher="^mcp__", hooks=[mcp_audit_hook]),
            # Match everything (no matcher)
            HookMatcher(hooks=[global_logger]),
        ]
    }
)

关键片段

原页关键片段：Track subagent activity

先看下面这块原始片段，等会儿再回头看解释会顺得多。

async def subagent_tracker(input_data, tool_use_id, context):
    # Log subagent details when it finishes
    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])]}
)

关键片段

原页关键片段：Make HTTP requests from hooks

下面这块是这一段最值钱的原文样板，先对着看一眼。

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


def _send_webhook(tool_name):
    """Synchronous helper that POSTs tool usage data to an external webhook."""
    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):
    # Only fire after a tool completes (PostToolUse), not before
    if input_data["hook_event_name"] != "PostToolUse":
        return {}

    try:
        # Run the blocking HTTP call in a thread to avoid blocking the event loop
        await asyncio.to_thread(_send_webhook, input_data["tool_name"])
    except Exception as e:
        # Log the error but don't raise. A failed webhook shouldn't stop the agent
        print(f"Webhook request failed: {e}")

    return {}

关键片段

原页关键片段：Forward notifications to Slack

这一段要真抓重点，通常就抓下面这块原文。

import asyncio
import json
import urllib.request

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher


def _send_slack_notification(message):
    """Synchronous helper that sends a message to Slack via incoming webhook."""
    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:
        # Run the blocking HTTP call in a thread to avoid blocking the event loop
        await asyncio.to_thread(_send_slack_notification, input_data.get("message", ""))
    except Exception as e:
        print(f"Failed to send notification: {e}")

    # Return empty object. Notification hooks don't modify agent behavior
    return {}


async def main():
    options = ClaudeAgentOptions(
        hooks={
            # Register the hook for Notification events (no matcher needed)
            "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())

关键片段

原页关键片段：Matcher not filtering as expected

先看下面这块原始片段，等会儿再回头看解释会顺得多。

const myHook: HookCallback = async (input, toolUseID, { signal }) => {
  const preInput = input as PreToolUseHookInput;
  const toolInput = preInput.tool_input as Record<string, unknown>;
  const filePath = toolInput?.file_path as string;
  if (!filePath?.endsWith(".md")) return {}; // Skip non-markdown files
  // Process markdown files...
  return {};
};

预留广告位

正文中段响应式广告 等你后面真接 AdSense，这里再放正式广告。

Documentation Index

这里不是让你背"Documentation Index"这个词，而是让你看它真干活时怎么使。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

How hooks work

这一块主要是在说"How hooks work"真到手上该怎么用，哪里最容易踩坑。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

关键片段

How hooks work

"How hooks work"这一段里最要紧的原始写法在下面，先看它怎么落地。

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


# Define a hook callback that receives tool call details
async def protect_env_files(input_data, tool_use_id, context):
    # Extract the file path from the tool's input arguments
    file_path = input_data["tool_input"].get("file_path", "")
    file_name = file_path.split("/")[-1]

    # Block the operation if targeting a .env file
    if file_name == ".env":
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": "Cannot modify .env files",
            }
        }

    # Return empty object to allow the operation
    return {}


async def main():
    options = ClaudeAgentOptions(
        hooks={
            # Register the hook for PreToolUse events
            # The matcher filters to only Write and Edit tool calls
            "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():
            # Filter for assistant and result messages
            if isinstance(message, (AssistantMessage, ResultMessage)):
                print(message)


asyncio.run(main())

Available hooks

这一段不只是挂个标题，它是在说明"Available hooks"这一块到底负责什么。

如果你打算把外接能力往里挂，这里提到的 hooks、MCP、skills、memory 都要分清各自负责哪一摊。

Configure hooks

这里讲怎么把某个开关拧对。重点不是概念，而是你该在哪儿改、改完怎么确认生效。

看这段时要特别盯工具和权限边界，别为了省事一把全开。

关键片段

Configure hooks

下面这块是这一段最值钱的原文样板，先对着看一眼。

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)

Matchers

这一段是在说怎么用 matchers 去做 filter when your callbacks fire. The matcher field is a regex string that matches against a different value depending on the hook event type. For example, tool-based hooks match against the tool name, while Notification hooks match against the notification type. See the Claude Code hooks reference for the full list of matcher values for each event type.。看这种内容，光知道名字没用，还是得落到手上。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

Callback functions

这里不是让你背"Callback functions"这个词，而是让你看它真干活时怎么使。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

关键片段

Callback functions

下面这块是这一段最值钱的原文样板，先对着看一眼。

async def async_hook(input_data, tool_use_id, context):
    # Start a background task, then return immediately
    asyncio.create_task(send_to_logging_service(input_data))
    return {"async_": True, "asyncTimeout": 30000}

Examples

看到这里，就把"Examples"当成一件真要上手的活来看。

Modify tool input

这里主要是在交代"Modify tool input"这一块会碰到哪些事。

看这段时要特别盯工具和权限边界，别为了省事一把全开。

关键片段

Modify tool input

下面这块是这一段最值钱的原文样板，先对着看一眼。

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 {}

Add context and block a tool

别把这段只当成标题看，它其实是在给"Add context and block a tool"划边界。

看这段时要特别盯工具和权限边界，别为了省事一把全开。

关键片段

Add context and block a tool

先看下面这块原始片段，等会儿再回头看解释会顺得多。

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 {
            # Top-level field: inject guidance into the conversation
            "systemMessage": "Remember: system directories like /etc are protected.",
            # hookSpecificOutput: block the operation
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": "Writing to /etc is not allowed",
            },
        }
    return {}

Auto-approve specific tools

这里不是让你背"Auto-approve specific tools"这个词，而是让你看它真干活时怎么使。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

关键片段

Auto-approve specific tools

下面这块是这一段最值钱的原文样板，先对着看一眼。

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 {}

这一段更像在讲判断条件，什么时候该上，什么时候先别急。把触发条件看清，比背标题更重要。

看这段时要特别盯工具和权限边界，别为了省事一把全开。

关键片段

Register multiple hooks

"Register multiple hooks"这一段里最要紧的原始写法在下面，先看它怎么落地。

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

Filter with regex matchers

这一段是在说怎么用 regex patterns 去做 match multiple tools. This example registers three matchers with different scopes: the first triggers file_security_hook only for file modification tools, the second triggers mcp_audit_hook for any MCP tool (tools whose names start with mcp__), and the third triggers global_logger for every tool call regardless of name:。看这种内容，光知道名字没用，还是得落到手上。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

关键片段

Filter with regex matchers

这一段要真抓重点，通常就抓下面这块原文。

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            # Match file modification tools
            HookMatcher(matcher="Write|Edit|Delete", hooks=[file_security_hook]),
            # Match all MCP tools
            HookMatcher(matcher="^mcp__", hooks=[mcp_audit_hook]),
            # Match everything (no matcher)
            HookMatcher(hooks=[global_logger]),
        ]
    }
)

Track subagent activity

这一段是在说怎么用 SubagentStop hooks 去做 monitor when subagents finish their work. See the full input type in the TypeScript and Python SDK references. This example logs a summary each time a subagent completes:。看这种内容，光知道名字没用，还是得落到手上。

如果你打算把外接能力往里挂，这里提到的 hooks、MCP、skills、memory 都要分清各自负责哪一摊。

关键片段

Track subagent activity

先看下面这块原始片段，等会儿再回头看解释会顺得多。

async def subagent_tracker(input_data, tool_use_id, context):
    # Log subagent details when it finishes
    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])]}
)

Make HTTP requests from hooks

这里不是让你背"Make HTTP requests from hooks"这个词，而是让你看它真干活时怎么使。

看这段时要特别盯工具和权限边界，别为了省事一把全开。

关键片段

Make HTTP requests from hooks

下面这块是这一段最值钱的原文样板，先对着看一眼。

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


def _send_webhook(tool_name):
    """Synchronous helper that POSTs tool usage data to an external webhook."""
    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):
    # Only fire after a tool completes (PostToolUse), not before
    if input_data["hook_event_name"] != "PostToolUse":
        return {}

    try:
        # Run the blocking HTTP call in a thread to avoid blocking the event loop
        await asyncio.to_thread(_send_webhook, input_data["tool_name"])
    except Exception as e:
        # Log the error but don't raise. A failed webhook shouldn't stop the agent
        print(f"Webhook request failed: {e}")

    return {}

Forward notifications to Slack

这一段是在说怎么用 Notification hooks to receive system notifications from the agent and forward them 去做 external services. Notifications fire for specific event types: permission_prompt (Claude needs permission), idle_prompt (Claude is waiting for input), auth_success (authentication completed), and elicitation_dialog (Claude is prompting the user). Each notification includes a message field with a human-readable description and optionally a title.。看这种内容，光知道名字没用，还是得落到手上。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

关键片段

Forward notifications to Slack

这一段要真抓重点，通常就抓下面这块原文。

import asyncio
import json
import urllib.request

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher


def _send_slack_notification(message):
    """Synchronous helper that sends a message to Slack via incoming webhook."""
    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:
        # Run the blocking HTTP call in a thread to avoid blocking the event loop
        await asyncio.to_thread(_send_slack_notification, input_data.get("message", ""))
    except Exception as e:
        print(f"Failed to send notification: {e}")

    # Return empty object. Notification hooks don't modify agent behavior
    return {}


async def main():
    options = ClaudeAgentOptions(
        hooks={
            # Register the hook for Notification events (no matcher needed)
            "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())

Fix common issues

遇到这种内容，别急着大拆大改，先按它给的路子把问题缩小。

Hook not firing

这一块主要是在说"Hook not firing"真到手上该怎么用，哪里最容易踩坑。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

Matcher not filtering as expected

看到这里，就把"Matcher not filtering as expected"当成一件真要上手的活来看。

看这段时要特别盯工具和权限边界，别为了省事一把全开。

关键片段

Matcher not filtering as expected

先看下面这块原始片段，等会儿再回头看解释会顺得多。

const myHook: HookCallback = async (input, toolUseID, { signal }) => {
  const preInput = input as PreToolUseHookInput;
  const toolInput = preInput.tool_input as Record<string, unknown>;
  const filePath = toolInput?.file_path as string;
  if (!filePath?.endsWith(".md")) return {}; // Skip non-markdown files
  // Process markdown files...
  return {};
};

Hook timeout

这一段主要是在把"Hook timeout"讲实，不是只摆个标题给你看。

如果你打算把外接能力往里挂，这里提到的 hooks、MCP、skills、memory 都要分清各自负责哪一摊。

Tool blocked unexpectedly

这一段主要是在把"Tool blocked unexpectedly"讲实，不是只摆个标题给你看。

看这段时要特别盯工具和权限边界，别为了省事一把全开。

Modified input not applied

这一块主要是在说"Modified input not applied"真到手上该怎么用，哪里最容易踩坑。

看这段时要特别盯工具和权限边界，别为了省事一把全开。

关键片段

Modified input not applied

"Modified input not applied"这一段里最要紧的原始写法在下面，先看它怎么落地。

return {
  hookSpecificOutput: {
    hookEventName: "PreToolUse",
    permissionDecision: "allow",
    updatedInput: { command: "new command" }
  }
};

Session hooks not available in Python

看到这里，就把"Session hooks not available in Python"当成一件真要上手的活来看。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

改配置

Session hooks not available in Python

这一段说完，最后还得写到配置里才算真的生效。

options = ClaudeAgentOptions(
    setting_sources=["project"],  # Loads .claude/settings.json including hooks
)

Subagent permission prompts multiplying

这一段更像在讲判断条件，什么时候该上，什么时候先别急。把触发条件看清，比背标题更重要。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

Recursive hook loops with subagents

这里不是让你背"Recursive hook loops with subagents"这个词，而是让你看它真干活时怎么使。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

systemMessage not appearing in output

这里主要是在交代"systemMessage not appearing in output"这一块会碰到哪些事。

如果你打算把外接能力往里挂，这里提到的 hooks、MCP、skills、memory 都要分清各自负责哪一摊。

Related resources

这一段主要是在把"Related resources"讲实，不是只摆个标题给你看。

看这段时要特别盯工具和权限边界，别为了省事一把全开。

照着做一遍

如果你不想来回翻，就先照这几步顺着做。

每做完一步就看一下结果，再决定要不要继续往下。

关键片段

第 1 步：How hooks work

"How hooks work"这一段里最要紧的原始写法在下面，先看它怎么落地。

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


# Define a hook callback that receives tool call details
async def protect_env_files(input_data, tool_use_id, context):
    # Extract the file path from the tool's input arguments
    file_path = input_data["tool_input"].get("file_path", "")
    file_name = file_path.split("/")[-1]

    # Block the operation if targeting a .env file
    if file_name == ".env":
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": "Cannot modify .env files",
            }
        }

    # Return empty object to allow the operation
    return {}


async def main():
    options = ClaudeAgentOptions(
        hooks={
            # Register the hook for PreToolUse events
            # The matcher filters to only Write and Edit tool calls
            "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():
            # Filter for assistant and result messages
            if isinstance(message, (AssistantMessage, ResultMessage)):
                print(message)


asyncio.run(main())

关键片段

第 2 步：Configure hooks

下面这块是这一段最值钱的原文样板，先对着看一眼。

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)

关键片段

第 3 步：Callback functions

下面这块是这一段最值钱的原文样板，先对着看一眼。

async def async_hook(input_data, tool_use_id, context):
    # Start a background task, then return immediately
    asyncio.create_task(send_to_logging_service(input_data))
    return {"async_": True, "asyncTimeout": 30000}

关键片段

第 4 步：Modify tool input

下面这块是这一段最值钱的原文样板，先对着看一眼。

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 {}

一眼看懂这一页

先把这页到底在讲什么看明白，再去碰具体命令和配置，最不容易绕晕。

Intercept and control agent behavior with hooks
   |
   v
这是 Control and observability 里的一摊要紧活
   |
   v
先弄懂，再下手

文末提醒

这站会按官方 docs 的导航和内容变化继续重生成，原站加页、删页、改页时，这里会跟着更新。

人话解释会尽量顺着原页往下讲，但命令、参数名、配置名这些硬东西还是保留原样，免得你抄过去跑不起来。