claude code解析系列之(二)工具使用机制:让 AI 真正动起来-夜雨聆风

claude code解析系列之(二)工具使用机制:让 AI 真正动起来

系列文章第二篇 – 上一篇我们学会了让 AI “思考”，但这还不够。今天我们来给 AI 装上”手脚”——工具（Tools）。有了工具，AI 才能真正与真实世界交互，而不仅仅是”纸上谈兵”。

一、为什么需要工具？

大模型本身只是”脑子”，它能思考、能分析，但它不能：

执行命令
读写文件
访问网络
调用其他系统

工具就是 AI 的”手”，让它能够真正去做事情。

二、工具是如何工作的？

核心机制：工具调度（Tool Dispatch）

当 AI 决定使用工具时，系统需要：

识别 – AI 想用哪个工具？
提取 – 参数是什么？
执行 – 运行对应的处理函数
返回 – 把结果告诉 AI

这就像一个调度中心，AI 说”我要用 bash 工具执行 ls 命令”，调度中心就找到负责 bash 的操作员，执行命令，然后回报结果。

三、代码实现：处理器映射模式

看看我们是怎么实现工具调度的：

# 工具处理器映射表TOOL_HANDLERS = {    "bash":       lambda **kw: _run_bash(kw["command"]),    "read_file":  lambda **kw: _run_read(kw["path"], kw.get("limit")),    "write_file": lambda **kw: _run_write(kw["path"], kw["content"]),    "edit_file":  lambda **kw: _run_edit(kw["path"], kw["old_text"], kw["new_text"]),}# 工具执行循环for block in response.content:    if block.type == "tool_use":        handler = TOOL_HANDLERS.get(block.name)        if handler:            output = handler(**block.input)        else:            output = f"Unknown tool: {block.name}"        results.append({            "type": "tool_result",            "tool_use_id": block.id,            "content": str(output),        })

关键设计点：

1. 映射表（Handler Map）用一个字典把工具名称映射到对应的处理函数。这种设计非常灵活，添加新工具只需要加一行代码。

2. 参数解包（**kw）AI 传来的参数是一个字典，用 **block.input 可以直接解包成函数参数。这就像把打包好的行李箱打开，把里面的东西拿出来。

3. 优雅降级如果 AI 想用一个不存在的工具，我们不会崩溃，而是返回一个友好的错误信息。

四、四个基础工具详解

1. bash – 执行命令

def _run_bash(command: str) -> str:    # 安全检查    dangerous = ["rm -rf /", "sudo", "shutdown", "reboot"]    if any(d in command for d in dangerous):        return "Error: Dangerous command blocked"    # 执行命令    r = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=120)    return (r.stdout + r.stderr).strip()[:50000]

安全设计：阻止危险命令，限制输出长度，设置超时时间。这不是多此一举，而是防止 AI “手滑”删库跑路。

2. read_file – 读取文件

def _run_read(path: str, limit: int = None) -> str:    lines = _safe_path(path).read_text().splitlines()    if limit and limit < len(lines):        lines = lines[:limit] + [f"... ({len(lines) - limit} more)"]    return "\n".join(lines)[:50000]

智能截断：文件太大时，只显示前面部分，并告诉你还有多少行没显示。

3. write_file – 写入文件

def _run_write(path: str, content: str) -> str:    fp = _safe_path(path)    fp.parent.mkdir(parents=True, exist_ok=True)    fp.write_text(content)    return f"Wrote {len(content)} bytes"

自动创建目录：如果目标目录不存在，会自动创建，不会报错。

4. edit_file – 编辑文件

def _run_edit(path: str, old_text: str, new_text: str) -> str:    fp = _safe_path(path)    c = fp.read_text()    if old_text not in c:        return f"Error: Text not found in {path}"    fp.write_text(c.replace(old_text, new_text, 1))    return f"Edited {path}"

精确替换：只替换第一次出现的文本，避免误伤其他地方。

五、工具定义：告诉 AI 怎么用

光有工具实现还不够，我们还得告诉 AI 这些工具是做什么的、需要什么参数：

TOOLS = [    {        "name": "bash",        "description": "Run a shell command.",        "input_schema": {            "type": "object",            "properties": {                "command": {"type": "string"}            },            "required": ["command"]        }    },    {        "name": "read_file",        "description": "Read file contents.",        "input_schema": {            "type": "object",            "properties": {                "path": {"type": "string"},                "limit": {"type": "integer"}            },            "required": ["path"]        }    },    # ... 其他工具定义]

input_schema 是一个 JSON Schema，它定义了：

参数的类型（string、integer、boolean 等）
哪些参数是必需的
参数的约束条件

六、安全机制：路径沙箱

WORKDIR = Path.cwd()def _safe_path(p: str) -> Path:    path = (WORKDIR / p).resolve()    if not path.is_relative_to(WORKDIR):        raise ValueError(f"Path escapes workspace: {p}")    return path

沙箱机制：所有文件操作都被限制在工作目录内。AI 想读取 ../../etc/passwd？门都没有！

七、完整的工作流程

     用户请求        │        ▼    ┌─────────────────────────────────────┐    │         调用大模型                   │    │   把 TOOLS 定义一起发给 AI           │    └─────────────┬───────────────────────┘                  │                  ▼    ┌─────────────────────────────────────┐    │         AI 分析并决定                │    │   "我需要用 read_file 工具"          │    │   参数: {"path": "data.txt"}        │    └─────────────┬───────────────────────┘                  │                  ▼    ┌─────────────────────────────────────┐    │         工具调度                     │    │   handler = TOOL_HANDLERS["read_file"]│    │   output = handler(path="data.txt")  │    └─────────────┬───────────────────────┘                  │                  ▼    ┌─────────────────────────────────────┐    │         返回结果                     │    │   把文件内容告诉 AI                  │    └─────────────┬───────────────────────┘                  │                  ▼    ┌─────────────────────────────────────┐    │         AI 继续处理                  │    │   根据文件内容做下一步决策            │    └─────────────────────────────────────┘

八、对比：添加工具前后的变化

对比项	没有工具	有工具
能力范围	只能聊天	能执行真实操作
信息来源	训练数据	实时获取
可靠性	可能编造	基于事实
实用性	咨询建议	自动执行

九、设计原则总结

添加工具的原则：

单一职责

– 每个工具只做一件事

清晰描述

– 让 AI 准确理解工具用途

安全第一

– 限制危险操作，防止误操作

优雅降级

– 出错时给出有用的反馈

十、关键洞察

核心洞察：循环本身没有变化。我只是添加了工具。这就是工具机制的美妙之处——它是一种扩展机制，而不是改变核心逻辑。你可以在不修改 Agent Loop 的情况下，不断增加新的能力。

十一、总结

这一章我们学会了：

工具是 AI 与外界交互的”手”
使用处理器映射模式优雅地调度工具
四个基础工具：bash、read、write、edit
安全设计：沙箱机制、危险命令过滤

有了工具，AI 才能真正”干活”。但这还不够——下一个问题来了：AI 怎么知道自己要做什么？

下一篇，我们将学习 TodoWrite 机制，让 AI 能够跟踪自己的任务进度。

十二、附源码

#!/usr/bin/env python3"""s02_tool_use.py - ToolsThe agent loop from s01 didn't change. We just added tools to the arrayand a dispatch map to route calls.    +----------+      +-------+      +------------------+    |   User   | ---> |  LLM  | ---> | Tool Dispatch    |    |  prompt  |      |       |      | {                |    +----------+      +---+---+      |   bash: run_bash |                          ^          |   read: run_read |                          |          |   write: run_wr  |                          +----------+   edit: run_edit |                          tool_result| }                |                                     +------------------+Key insight: "The loop didn't change at all. I just added tools.""""import osimport subprocessfrom pathlib import Pathfrom anthropic import Anthropicfrom dotenv import load_dotenvload_dotenv(override=True)if os.getenv("ANTHROPIC_BASE_URL"):    os.environ.pop("ANTHROPIC_AUTH_TOKEN", None)WORKDIR = Path.cwd()client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))MODEL = os.environ["MODEL_ID"]SYSTEM = f"You are a coding agent at {WORKDIR}. Use tools to solve tasks. Act, don't explain."def safe_path(p: str) -> Path:    path = (WORKDIR / p).resolve()    if not path.is_relative_to(WORKDIR):        raise ValueError(f"Path escapes workspace: {p}")    return pathdef run_bash(command: str) -> str:    dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]    if any(d in command for d in dangerous):        return "Error: Dangerous command blocked"    try:        r = subprocess.run(command, shell=True, cwd=WORKDIR,                           capture_output=True, text=True, timeout=120)        out = (r.stdout + r.stderr).strip()        return out[:50000] if out else "(no output)"    except subprocess.TimeoutExpired:        return "Error: Timeout (120s)"def run_read(path: str, limit: int = None) -> str:    try:        text = safe_path(path).read_text()        lines = text.splitlines()        if limit and limit < len(lines):            lines = lines[:limit] + [f"... ({len(lines) - limit} more lines)"]        return "\n".join(lines)[:50000]    except Exception as e:        return f"Error: {e}"def run_write(path: str, content: str) -> str:    try:        fp = safe_path(path)        fp.parent.mkdir(parents=True, exist_ok=True)        fp.write_text(content)        return f"Wrote {len(content)} bytes to {path}"    except Exception as e:        return f"Error: {e}"def run_edit(path: str, old_text: str, new_text: str) -> str:    try:        fp = safe_path(path)        content = fp.read_text()        if old_text not in content:            return f"Error: Text not found in {path}"        fp.write_text(content.replace(old_text, new_text, 1))        return f"Edited {path}"    except Exception as e:        return f"Error: {e}"# -- The dispatch map: {tool_name: handler} --TOOL_HANDLERS = {    "bash":       lambda **kw: run_bash(kw["command"]),    "read_file":  lambda **kw: run_read(kw["path"], kw.get("limit")),    "write_file": lambda **kw: run_write(kw["path"], kw["content"]),    "edit_file":  lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),}TOOLS = [    {"name": "bash", "description": "Run a shell command.",     "input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}},    {"name": "read_file", "description": "Read file contents.",     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}},    {"name": "write_file", "description": "Write content to file.",     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},    {"name": "edit_file", "description": "Replace exact text in file.",     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}},]def agent_loop(messages: list):    while True:        response = client.messages.create(            model=MODEL, system=SYSTEM, messages=messages,            tools=TOOLS, max_tokens=8000,        )        messages.append({"role": "assistant", "content": response.content})        if response.stop_reason != "tool_use":            return        results = []        for block in response.content:            if block.type == "tool_use":                handler = TOOL_HANDLERS.get(block.name)                output = handler(**block.input) if handler else f"Unknown tool: {block.name}"                print(f"> {block.name}: {output[:200]}")                results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})        messages.append({"role": "user", "content": results})if __name__ == "__main__":    history = []    while True:        try:            query = input("\033[36ms02 >> \033[0m")        except (EOFError, KeyboardInterrupt):            break        if query.strip().lower() in ("q", "exit", ""):            break        history.append({"role": "user", "content": query})        agent_loop(history)        response_content = history[-1]["content"]        if isinstance(response_content, list):            for block in response_content:                if hasattr(block, "text"):                    print(block.text)        print()