第 6 课：安全与权限管控

客服系统为什么需要安全管控？

想象一下： - AI 客服不小心把客户 A 的手机号告诉了客户 B - AI 客服直接帮客户退了 10 万的订单，没有任何审批 - AI 客服执行了一个删除数据库的命令

这些都是灾难。AI 强大的同时，也需要"围栏"。

Claude Agent SDK 提供两套安全机制： 1. Hook（钩子）：在工具执行前后拦截，做检查 2. Permission Callback（权限回调）：精细控制每个工具的使用权限

Hook 基础：PreToolUse 和 PostToolUse

场景一：PII 脱敏

客户个人信息（PII = Personally Identifiable Information）不能随便暴露。

code
import re
from claude_agent_sdk.types import HookInput, HookContext, HookJSONOutput

async def pii_filter_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """工具返回结果后，自动脱敏 PII"""

    tool_response = str(input_data.get("tool_response", ""))

    # 手机号脱敏: 13812345678 → 138****5678
    masked = re.sub(
        r'(\d{3})\d{4}(\d{4})',
        r'\1****\2',
        tool_response
    )

    # 邮箱脱敏: zhangsan@example.com → z***n@example.com
    def mask_email(match):
        email = match.group(0)
        name, domain = email.split("@")
        if len(name) <= 2:
            masked_name = name[0] + "***"
        else:
            masked_name = name[0] + "***" + name[-1]
        return f"{masked_name}@{domain}"

    masked = re.sub(
        r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
        mask_email,
        masked
    )

    # 身份证脱敏: 110101199001011234 → 1101***********1234
    masked = re.sub(
        r'(\d{4})\d{10}(\d{4})',
        r'\1**********\2',
        masked
    )

    # 如果做了脱敏，更新工具输出
    if masked != tool_response:
        return {
            "hookSpecificOutput": {
                "hookEventName": "PostToolUse",
                "updatedMCPToolOutput": masked,
                "additionalContext": "注意：返回数据中的个人信息已自动脱敏。请勿尝试还原。",
            }
        }

    return {}

效果：

code

原始数据: {"name": "张三", "phone": "13812345678", "email": "zhangsan@example.com"}
脱敏后:   {"name": "张三", "phone": "138****5678", "email": "z***n@example.com"}

AI 拿到的就是脱敏后的数据，绝不会泄露给客户。

场景二：敏感操作拦截

退款、删除、修改这类操作，需要拦截确认：

code
# 需要审批的操作
SENSITIVE_TOOLS = [
    "mcp__support__update_ticket_status",
    "mcp__support__process_refund",
    "mcp__support__delete_ticket",
]

async def sensitive_operation_gate(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """拦截敏感操作，要求人工确认"""

    tool_name = input_data["tool_name"]

    if tool_name in SENSITIVE_TOOLS:
        tool_input = input_data.get("tool_input", {})
        print(f"\n⚠️ 敏感操作需要确认:")
        print(f"   工具: {tool_name}")
        print(f"   参数: {tool_input}")

        confirm = input("   确认执行？(y/N): ").strip().lower()
        if confirm != 'y':
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": "人工审核未通过",
                }
            }

        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "allow",
                "permissionDecisionReason": "人工审核通过",
            }
        }

    return {}

场景三：操作审计日志

每一步操作都要记录下来，出了问题能追溯：

code
import datetime

audit_records: list[dict] = []

async def audit_trail(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """记录每次工具调用的审计日志"""

    record = {
        "timestamp": datetime.datetime.now().isoformat(),
        "tool": input_data.get("tool_name", "unknown"),
        "input_summary": str(input_data.get("tool_input", {}))[:200],
        "response_summary": str(input_data.get("tool_response", ""))[:200],
    }
    audit_records.append(record)

    # 同时打印到控制台
    print(f"  📝 审计: {record['tool']} @ {record['timestamp'][:19]}")

    return {}


def print_audit_summary():
    """打印审计摘要"""
    print(f"\n📋 审计摘要: 共 {len(audit_records)} 次操作")
    for i, r in enumerate(audit_records, 1):
        print(f"  {i}. [{r['timestamp'][:19]}] {r['tool']}")

场景四：用 Permission Callback 做精细控制

Hook 是"全局拦截"，Permission Callback 是"逐个审批"：

code
from claude_agent_sdk import (
    PermissionResultAllow, PermissionResultDeny,
    ToolPermissionContext,
)

async def support_permission_callback(
    tool_name: str,
    input_data: dict,
    context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
    """客服系统的权限控制"""

    # 查询类工具：自动允许
    read_tools = [
        "mcp__support__search_faq",
        "mcp__support__get_ticket",
        "mcp__support__get_order",
        "Read", "Grep", "Glob",
    ]
    if tool_name in read_tools:
        return PermissionResultAllow()

    # 创建工单：允许，但限制优先级
    if tool_name == "mcp__support__create_ticket":
        # AI 不能自己创建 "critical" 优先级的工单
        if input_data.get("priority") == "critical":
            modified = input_data.copy()
            modified["priority"] = "high"  # 降级为 high
            return PermissionResultAllow(updated_input=modified)
        return PermissionResultAllow()

    # 退款操作：拒绝超过 5000 元的
    if tool_name == "mcp__support__process_refund":
        amount = input_data.get("amount", 0)
        if amount > 5000:
            return PermissionResultDeny(
                message=f"退款金额 ¥{amount} 超过 ¥5000 限额，需要主管审批。"
            )
        return PermissionResultAllow()

    # 删除操作：全部拒绝
    if "delete" in tool_name.lower():
        return PermissionResultDeny(
            message="AI 客服无权执行删除操作。"
        )

    # 其他：默认允许
    return PermissionResultAllow()

组装：带安全的完整客服

code
import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    AssistantMessage, TextBlock, ToolUseBlock,
)
from claude_agent_sdk.types import HookMatcher

# 导入上面写的 Hook 和工具
# from hooks.safety import ...
# from tools.support_tools import ...

async def main():
    options = ClaudeAgentOptions(
        system_prompt="你是客服助手小智。用中文回复。",
        mcp_servers={"support": support_server},
        allowed_tools=[
            "mcp__support__search_faq",
            "mcp__support__get_order",
            "mcp__support__create_ticket",
            "mcp__support__check_refund_eligibility",
        ],
        # Hook 配置
        hooks={
            "PreToolUse": [
                # 敏感操作拦截
                HookMatcher(
                    matcher=None,
                    hooks=[sensitive_operation_gate],
                ),
            ],
            "PostToolUse": [
                # PII 脱敏
                HookMatcher(
                    matcher=None,
                    hooks=[pii_filter_hook],
                ),
                # 审计日志
                HookMatcher(
                    matcher=None,
                    hooks=[audit_trail],
                ),
            ],
        },
        # 权限回调
        can_use_tool=support_permission_callback,
    )

    async with ClaudeSDKClient(options=options) as client:
        while True:
            user_input = input("客户: ").strip()
            if user_input.lower() == 'quit':
                print_audit_summary()
                break

            await client.query(user_input)
            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(f"🤖 {block.text}")
                        elif isinstance(block, ToolUseBlock):
                            print(f"  🔧 {block.name}")
            print()

asyncio.run(main)

Hook 和 Permission Callback 的区别

特性	Hook	Permission Callback
作用时机	PreToolUse / PostToolUse	工具执行前
能修改输入	❌	✅ (updated_input)
能修改输出	✅ (updatedMCPToolOutput)	❌
能拒绝执行	✅ (permissionDecision: deny)	✅ (PermissionResultDeny)
能加审计	✅	✅
适合场景	全局拦截、脱敏、日志	逐工具精细控制

简单说：Hook 管大方向，Callback 管细节。两个可以组合使用。

安全最佳实践

code

1. 最小权限
   └── AI 只给它需要的工具，不多给

2. 分层防御
   ├── Hook: 全局拦截危险操作
   ├── Callback: 精细控制每个工具
   └── 业务逻辑: 工具函数内部也要检查

3. PII 保护
   ├── PostToolUse 自动脱敏
   └── 日志中也要脱敏

4. 操作审计
   └── 每次工具调用都记录，出问题能追溯

5. 金额限制
   └── 退款、支付等操作设定上限

6. 人工兜底
   └── 超出 AI 能力范围的，自动转人工

本课小结

Hook 系统提供 PreToolUse（执行前拦截）和 PostToolUse（执行后处理）
PII 脱敏用 PostToolUse Hook，自动替换敏感信息
敏感操作拦截用 PreToolUse Hook，需要人工确认
Permission Callback 提供更精细的逐工具控制
审计日志记录所有操作，方便追溯

课后练习

给 PII 脱敏加银行卡号的匹配规则
实现一个"操作频率限制"Hook：同一个工具 1 分钟内最多调用 10 次
把审计日志保存到文件，每天一个文件