Claude Agent SDK 实战

[Chat API]
   │
   ▼
[Conversation Service] ──► [Session Store]
   │
   ▼
[Agent Loop] ──► [Tool Gateway] ──► [Order/Refund Service]
   │                 │                    │
   │                 ▼                    ▼
   │            [Approval Service]   [Idempotency Store]
   │                 │
   ▼                 ▼
[Audit/Event Log] ◄── [Policy/PII Boundary]

READ_TOOLS = [
    "mcp__shop__get_order",
    "mcp__shop__list_orders",
    "mcp__shop__refund_policy",
    "mcp__shop__get_customer",
]

options = ClaudeAgentOptions(
    system_prompt=SYSTEM,
    mcp_servers={"shop": desk.server()},
    allowed_tools=READ_TOOLS,          # 只读免审批
    permission_mode="default",
    can_use_tool=desk.approve,         # 写操作走这里
    hooks=audit_hooks(log),            # 审计 + 脱敏
    max_turns=10,
    max_budget_usd=1.0,
)

def refundable(oid: str, customer: str) -> tuple[bool, str]:
    o = ORDERS.get(oid)
    if o is None or o["customer"] != customer:
        return False, "订单不属于当前客户"
    if o["status"] != "已签收":
        return False, f"订单状态为{o['status']}, 签收后才能退款"
    days = (date.fromisoformat(TODAY) - date.fromisoformat(o["date"])).days
    return (days <= 30, f"签收后第 {days} 天" + ("" if days <= 30 else ", 超过 30 天"))

@dataclass
class Approval:
    id: str
    run_id: str
    tool: str
    args: dict
    state: str = "pending"   # pending/approved/denied/expired
    approver: str | None = None
    reason: str = ""

def decide(conn, approval_id: str, approver: str, approved: bool, reason: str) -> bool:
    cur = conn.execute(
        """UPDATE approvals
           SET state=?, approver=?, reason=?, decided_at=?
           WHERE id=? AND state='pending'""",
        ("approved" if approved else "denied", approver, reason, now(), approval_id),
    )
    conn.commit()
    return cur.rowcount == 1

def create_refund(conn, order_id: str, request_id: str) -> dict:
    cur = conn.execute(
        "INSERT OR IGNORE INTO refunds(order_id, request_id, status) VALUES(?,?,?)",
        (order_id, request_id, "created"),
    )
    conn.commit()
    if cur.rowcount == 0:
        # 已存在：返回上一次结果，不重复退
        row = conn.execute(
            "SELECT refund_id, status FROM refunds WHERE request_id=?",
            (request_id,),
        ).fetchone()
        return {"refund_id": row[0], "status": row[1], "deduped": True}
    ...

def mask(value):
    if isinstance(value, str):
        value = re.sub(r"(?<!\d)(1[3-9]\d)\d{4}(\d{4})(?!\d)", r"\1****\2", value)
        return value
    if isinstance(value, list):
        return [mask(v) for v in value]
    if isinstance(value, dict):
        return {k: mask(v) for k, v in value.items()}
    return value

async def on_post(inp, tool_use_id, ctx):
    raw = inp.get("tool_response")
    masked = mask(raw)
    if masked == raw:
        return {}
    return {
        "hookSpecificOutput": {
            "hookEventName": "PostToolUse",
            "updatedToolOutput": masked,   # 必须和原输出同形
        }
    }

async def main():
    client = ClaudeSDKClient(options=options)
    await client.connect()
    try:
        await client.query("你好，我是客户 c7。订单 a1001 能退吗？")
        async for msg in client.receive_response():
            ...
    finally:
        await client.disconnect()

options = ClaudeAgentOptions(..., resume=session_id)