1#!/usr/bin/env python3
  2"""
  3s02: Tool Use — 在 s01 基础上新增 4 个工具 + 分发映射。
  4
  5运行: python s02_tool_use/code.py
  6需要: pip install anthropic python-dotenv + .env 中配置 ANTHROPIC_API_KEY
  7
  8本文件 = s01 的全部代码 + 以下新增:
  9  + run_read / run_write / run_edit / run_glob 四个工具实现
 10  + TOOL_HANDLERS 分发映射（替代 s01 中硬编码的 run_bash 调用）
 11  + safe_path 路径安全校验
 12
 13循环本身（agent_loop）与 s01 完全一致。
 14"""
 15
 16import os, subprocess
 17from pathlib import Path
 18
 19try:
 20    import readline
 21    readline.parse_and_bind('set bind-tty-special-chars off')
 22    readline.parse_and_bind('set input-meta on')
 23    readline.parse_and_bind('set output-meta on')
 24    readline.parse_and_bind('set convert-meta off')
 25except ImportError:
 26    pass
 27
 28from anthropic import Anthropic
 29from dotenv import load_dotenv
 30
 31load_dotenv(override=True)
 32if os.getenv("ANTHROPIC_BASE_URL"):
 33    os.environ.pop("ANTHROPIC_AUTH_TOKEN", None)
 34
 35WORKDIR = Path.cwd()
 36client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))
 37MODEL = os.environ["MODEL_ID"]
 38
 39SYSTEM = f"You are a coding agent at {WORKDIR}. Use tools to solve tasks. Act, don't explain."
 40
 41
 42# ═══════════════════════════════════════════════════════════
 43#  FROM s01 (unchanged)
 44# ═══════════════════════════════════════════════════════════
 45
 46def run_bash(command: str) -> str:
 47    dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
 48    if any(d in command for d in dangerous):
 49        return "Error: Dangerous command blocked"
 50    try:
 51        r = subprocess.run(command, shell=True, cwd=WORKDIR,
 52                           capture_output=True, text=True,
 53                           encoding="utf-8", errors="replace", timeout=120)
 54        out = (r.stdout + r.stderr).strip()
 55        return out[:50000] if out else "(no output)"
 56    except subprocess.TimeoutExpired:
 57        return "Error: Timeout (120s)"
 58    except (FileNotFoundError, OSError) as e:
 59        return f"Error: {e}"
 60
 61
 62# ═══════════════════════════════════════════════════════════
 63#  NEW in s02: 4 个新工具
 64# ═══════════════════════════════════════════════════════════
 65
 66def safe_path(p: str) -> Path:
 67    path = (WORKDIR / p).resolve()
 68    if not path.is_relative_to(WORKDIR):
 69        raise ValueError(f"Path escapes workspace: {p}")
 70    return path
 71
 72
 73def run_read(path: str, limit: int | None = None) -> str:
 74    try:
 75        lines = safe_path(path).read_text().splitlines()
 76        if limit and limit < len(lines):
 77            lines = lines[:limit] + [f"... ({len(lines) - limit} more lines)"]
 78        return "\n".join(lines)
 79    except Exception as e:
 80        return f"Error: {e}"
 81
 82
 83def run_write(path: str, content: str) -> str:
 84    try:
 85        file_path = safe_path(path)
 86        file_path.parent.mkdir(parents=True, exist_ok=True)
 87        file_path.write_text(content)
 88        return f"Wrote {len(content)} bytes to {path}"
 89    except Exception as e:
 90        return f"Error: {e}"
 91
 92
 93def run_edit(path: str, old_text: str, new_text: str) -> str:
 94    try:
 95        file_path = safe_path(path)
 96        text = file_path.read_text()
 97        if old_text not in text:
 98            return f"Error: text not found in {path}"
 99        file_path.write_text(text.replace(old_text, new_text, 1))
100        return f"Edited {path}"
101    except Exception as e:
102        return f"Error: {e}"
103
104
105def run_glob(pattern: str) -> str:
106    import glob as g
107    try:
108        results = []
109        for match in g.glob(pattern, root_dir=WORKDIR):
110            if (WORKDIR / match).resolve().is_relative_to(WORKDIR):
111                results.append(match)
112        return "\n".join(results) if results else "(no matches)"
113    except Exception as e:
114        return f"Error: {e}"
115
116
117# ═══════════════════════════════════════════════════════════
118#  NEW in s02: 工具定义（s01 只有一个 bash，现在扩展到 5 个）
119# ═══════════════════════════════════════════════════════════
120
121TOOLS = [
122    {"name": "bash", "description": "Run a shell command.",
123     "input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}},
124    {"name": "read_file", "description": "Read file contents.",
125     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}},
126    {"name": "write_file", "description": "Write content to a file.",
127     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},
128    {"name": "edit_file", "description": "Replace exact text in a file once.",
129     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}},
130    {"name": "glob", "description": "Find files matching a glob pattern.",
131     "input_schema": {"type": "object", "properties": {"pattern": {"type": "string"}}, "required": ["pattern"]}},
132]
133
134# ═══════════════════════════════════════════════════════════
135#  NEW in s02: 工具分发映射（s01 是硬编码 run_bash，现在改为查表）
136# ═══════════════════════════════════════════════════════════
137
138TOOL_HANDLERS = {
139    "bash": run_bash, "read_file": run_read, "write_file": run_write,
140    "edit_file": run_edit, "glob": run_glob,
141}
142
143# my code in s03
144# -----------------------------------------------------------
145
146# Gate 1: Hard deny list — always forbidden
147DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"]
148def check_deny_list(command: str) -> bool:
149    if any(d in command for d in DENY_LIST):
150        return "Error: Dangerous command blocked"
151    return None
152
153# Gate 2: Rule matching — context-dependent checks
154PERMISSION_RULES = [
155    {
156        "tools": ["write_file", "edit_file"], # 针对的工具
157        "check": lambda args: not (WORKDIR / args.get("path", "")).resolve().is_relative_to(WORKDIR), # lambda方法
158        "message": "Writing outside workspace" # 提供给用户的核验消息
159    },
160    {
161        "tools": ["bash"],
162        "check": lambda args: any(kw in args.get("command", "") for kw in ["rm ", "> /etc/", "chmod 777"]),
163        "message": "Potentially destructive command"
164    },
165]
166
167def check_rules(tool_name: str, args: dict) -> str | None:
168    for rule in PERMISSION_RULES:
169        # 如果在检测名单中 并且触发了危险信号 提交人工审核
170        if tool_name in rule["tools"] and rule["check"](args):
171            return f"Warning: {rule['message']}"
172    return None
173
174# Gate 3: User approval — wait for confirmation after rule match
175def ask_user(tool_name: str, args: dict, reason: str) -> str:
176    print(f"\n\033[33m⚠  {reason}\033[0m")
177    print(f"   Tool: {tool_name}({args})")
178    choice = input("   Allow? [y/N] ").strip().lower()
179    return "allow" if choice in ("y", "yes") else "deny"
180
181# pipeline
182def check_permission_pipeline(block) -> str:
183    if block.name == 'bash':
184        reason = check_deny_list(block.input.get("command", ""))
185        if reason:
186            print(f"\n\033[31m⛔ {reason}\033[0m")
187            return False
188    reason = check_rules(block.name, block.input)
189    if reason:
190        decision = ask_user(block.name, block.input, reason)
191        if decision == "deny":
192            print(f"\n\033[31m⛔ User denied permission for {block.name}\033[0m")
193            return False
194    return True
195
196
197# -----------------------------------------------------------
198# ═══════════════════════════════════════════════════════════
199#  agent_loop — 与 s01 结构完全一致，只改了工具执行那部分
200#  s01: output = run_bash(block.input["command"])
201#  s02: output = TOOL_HANDLERS[block.name](**block.input)
202# ═══════════════════════════════════════════════════════════
203
204def agent_loop(messages: list):
205    while True:
206        response = client.messages.create(
207            model=MODEL, system=SYSTEM, messages=messages,
208            tools=TOOLS, max_tokens=8000,
209        )
210        messages.append({"role": "assistant", "content": response.content})
211
212        if response.stop_reason != "tool_use":
213            return
214
215        results = []
216        for block in response.content:
217            if block.type == "tool_use":
218                
219                print(f"\033[33m> {block.name}({block.input})\033[0m")
220                # 权限核验
221                if not check_permission_pipeline(block):
222                    results.append({"type": "tool_result", "tool_use_id": block.id, "content": "Permission denied"})
223                    print(f"\033[31m⛔ Permission denied for {block.name}\033[0m")
224                    continue
225
226                print(f"\033[33m> {block.name}\033[0m")
227                handler = TOOL_HANDLERS.get(block.name)
228                output = handler(**block.input) if handler else f"Unknown: {block.name}"
229                print(str(output)[:200])
230                results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})
231
232        messages.append({"role": "user", "content": results})
233
234
235if __name__ == "__main__":
236    print("s02: Tool Use — 在 s01 基础上加了 4 个工具")
237    print("输入问题，回车发送。输入 q 退出。\n")
238
239    history = []
240    while True:
241        try:
242            query = input("\033[36ms02 >> \033[0m")
243        except (EOFError, KeyboardInterrupt):
244            break
245        if query.strip().lower() in ("q", "exit", ""):
246            break
247        history.append({"role": "user", "content": query})
248        agent_loop(history)
249        for block in history[-1]["content"]:
250            if getattr(block, "type", None) == "text":
251                print(block.text)
252        print()

1{ toolName: "Bash", ruleBehavior: "deny", ruleContent: "npm publish:*" }