ba1b5b76c6
Add comprehensive documentation for Claude Agent SDK including usage patterns, test examples, and development guide. Includes: - CLAUDE.md with project structure and quick reference - SDK test examples for query, client, and hooks - Shared test utilities for auth and tmpdir management - Detailed SDK overview and capabilities documentation
112 lines
3.2 KiB
Python
112 lines
3.2 KiB
Python
"""Example: Hooks — audit log + PreToolUse deny.
|
|
|
|
Demonstrates:
|
|
- PostToolUse hook for audit logging
|
|
- PreToolUse hook to deny dangerous commands
|
|
- HookMatcher pattern matching by tool name
|
|
- Hook callback signature and return format
|
|
|
|
Usage:
|
|
cd docs/claude
|
|
../../.venv/Scripts/python test_hooks_audit_deny.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
from _sdk_test_common import auth_env, make_tmpdir, remove_tmpdir, setup_auth
|
|
|
|
audit_log: list[dict] = []
|
|
|
|
|
|
async def audit_hook(input_data, tool_use_id, context):
|
|
"""PostToolUse: record every tool invocation."""
|
|
audit_log.append({
|
|
"tool": input_data.get("tool_name"),
|
|
"input": input_data.get("tool_input"),
|
|
})
|
|
return {}
|
|
|
|
|
|
async def deny_rm_hook(input_data, tool_use_id, context):
|
|
"""PreToolUse: block 'rm' commands."""
|
|
if input_data.get("tool_name") != "Bash":
|
|
return {}
|
|
command = input_data.get("tool_input", {}).get("command", "")
|
|
if "rm " in command:
|
|
return {
|
|
"hookSpecificOutput": {
|
|
"hookEventName": "PreToolUse",
|
|
"permissionDecision": "deny",
|
|
"permissionDecisionReason": "rm commands are blocked by policy",
|
|
}
|
|
}
|
|
return {}
|
|
|
|
|
|
async def main() -> int:
|
|
ok, msg = setup_auth()
|
|
print(msg)
|
|
if not ok:
|
|
return 1
|
|
|
|
from claude_agent_sdk import (
|
|
AssistantMessage,
|
|
ClaudeAgentOptions,
|
|
HookMatcher,
|
|
ResultMessage,
|
|
TextBlock,
|
|
query,
|
|
)
|
|
|
|
tmpdir = make_tmpdir("hooks-")
|
|
try:
|
|
(tmpdir / "data.txt").write_text("important data\n", encoding="utf-8")
|
|
|
|
print("--- Query with hooks ---")
|
|
opts = ClaudeAgentOptions(
|
|
cwd=str(tmpdir),
|
|
allowed_tools=["Bash", "Read"],
|
|
permission_mode="acceptEdits",
|
|
max_turns=4,
|
|
env=auth_env(),
|
|
hooks={
|
|
"PostToolUse": [HookMatcher(matcher="Bash|Read", hooks=[audit_hook])],
|
|
"PreToolUse": [HookMatcher(matcher="Bash", hooks=[deny_rm_hook])],
|
|
},
|
|
)
|
|
|
|
async for msg in query(
|
|
prompt=(
|
|
"Do these steps in order:\n"
|
|
"1. Read data.txt\n"
|
|
"2. Run 'echo hello'\n"
|
|
"3. Run 'rm data.txt'\n"
|
|
"Report what happened for each step."
|
|
),
|
|
options=opts,
|
|
):
|
|
if isinstance(msg, AssistantMessage):
|
|
for block in msg.content:
|
|
if isinstance(block, TextBlock):
|
|
print(f" Claude: {block.text[:300]}")
|
|
elif isinstance(msg, ResultMessage):
|
|
print(f" Result: {msg.result!r:.300}")
|
|
|
|
print(f"\nAudit log ({len(audit_log)} entries):")
|
|
for entry in audit_log:
|
|
print(f" {entry['tool']}: {entry['input']}")
|
|
|
|
file_exists = (tmpdir / "data.txt").exists()
|
|
print(f"\ndata.txt still exists: {file_exists}")
|
|
print("PASS" if file_exists else "FAIL: rm was not blocked")
|
|
return 0 if file_exists else 2
|
|
finally:
|
|
remove_tmpdir(tmpdir)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(asyncio.run(main()))
|