10_permissions.py

"""10: Runtime approval with can_use_tool. Allows file creation, denies deletion, rewrites paths."""

import asyncio

from claude_agent_sdk import (
    ClaudeAgentOptions,
    HookMatcher,
    PermissionResultAllow,
    PermissionResultDeny,
    ResultMessage,
    ToolPermissionContext,
    UserMessage,
    ToolResultBlock,
    query,
)
from demo_util import PROJECT

decisions = []


async def can_use_tool(tool_name: str, input_data: dict, context: ToolPermissionContext):
    """Stands in for a UI prompt. Policy: never delete, sandbox all writes under scratch/."""
    if tool_name == "Bash" and "rm " in input_data.get("command", ""):
        decisions.append(("deny", tool_name, input_data["command"]))
        return PermissionResultDeny(message="User policy: deletion is not allowed. Leave the file in place.")

    if tool_name == "Write":
        path = input_data["file_path"]
        if "/scratch/" not in path:
            new_path = path.rsplit("/", 1)[0] + "/scratch/" + path.rsplit("/", 1)[1]
            decisions.append(("allow+rewrite", tool_name, f"{path} -> {new_path}"))
            return PermissionResultAllow(updated_input={**input_data, "file_path": new_path})

    decisions.append(("allow", tool_name, str(input_data)[:60]))
    return PermissionResultAllow(updated_input=input_data)


# Required workaround: a (no-op) hook keeps the control stream open for can_use_tool
async def noop_hook(input_data, tool_use_id, context):
    return {"continue_": True}


async def prompt_stream():
    yield {"type": "user", "message": {"role": "user",
           "content": "Create a file named notes.txt containing 'hello', then delete it with rm."}}


async def main():
    async for message in query(
        prompt=prompt_stream(),
        options=ClaudeAgentOptions(
            cwd=PROJECT,
            permission_mode="default",
            can_use_tool=can_use_tool,
            hooks={"PreToolUse": [HookMatcher(matcher=None, hooks=[noop_hook])]},
            max_turns=6,
        ),
    ):
        if isinstance(message, UserMessage) and isinstance(message.content, list):
            for block in message.content:
                if isinstance(block, ToolResultBlock):
                    print(f"[tool_result]{' ERROR' if block.is_error else ''} {str(block.content)[:160]}")
        elif isinstance(message, ResultMessage):
            print(f"\n[result] {message.subtype}: {message.result}")
            print("[result] permission_denials:", message.permission_denials)

    print("\nDecisions made by can_use_tool:")
    for d in decisions:
        print("  ", d)


asyncio.run(main())