Harness实战：给Agent装上文件操作工具

只有Bash的Agent像只有锤子的工人。给Agent加上read_file、write_file、edit_file三把精细工具，用路径安全函数守住工作区边界，再用分发表让循环自动路由——从"能跑命令"进化到"能操作代码"。

写在前面

上一篇《Harness实战：从零搭建Agent Loop》我们用不到 50 行 Python 搭了一个能跑的 Agent——模型决定调工具，Harness 执行，结果喂回来，循环直到搞定。

但那个 Agent 只有一把锤子：Bash。

Bash 当然能干很多事——cat 读文件、echo > 写文件、sed 做替换。但你有没有想过，为什么 Claude Code 这样的 coding agent 不把所有文件操作都交给 Bash，而是专门做了 Read、Write、Edit 三个独立工具？

今天我们就来回答这个问题，并且动手把这三个工具加到我们的 Agent 里。

为什么不全用 Bash？

三个原因：

1. 安全边界

Bash 是万能的，但"万能"也意味着"万能地搞破坏"。一条 cat /etc/passwd 或者 echo "oops" > ~/.zshrc 就能越过工作区。专用工具可以加一层路径检查，把文件操作锁在工作区内。

2. 语义清晰

模型看到 edit_file 就知道是精确替换一段文本。如果用 sed，模型得自己拼正则——正则写错了，文件就改坏了。工具越语义化，模型出错的概率越低。

3. 输出可控

cat 一个 5000 行的文件，整个上下文窗口就被撑爆了。专用的 read_file 可以限制行数、截断输出，保护 context window。

这也是 Claude Code 的设计思路：Bash 做通用执行，专用工具做精细操作。

专用工具 vs 万能 Bash

第一步：路径安全函数

所有文件操作共享一个安全检查——确保路径不会逃出工作区：

from pathlib import Path

WORKDIR = Path(os.getcwd())

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

两行代码，干了两件事：

resolve() 展开 .. 和软链接，把路径变成绝对路径
is_relative_to(WORKDIR) 检查最终路径是否还在工作区内

这一行就挡住了 ../../etc/passwd 这类路径穿越攻击。简单，但够用。

safe_path 路径安全

第二步：三个文件工具

read_file — 读取文件

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}"

两层保护：limit 参数控制读取行数，[:50000] 硬截断防止超大文件。模型可以先读前 10 行看看结构，再决定要不要读全文。

write_file — 写入文件

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}"

mkdir(parents=True) 自动创建中间目录。模型不需要先跑一个 mkdir -p，直接写就行。

edit_file — 精确替换

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}"

replace(..., 1) 只替换第一处匹配，避免误改。找不到目标文本会返回错误，模型可以据此调整策略——比如先 read_file 看看文件内容，再用正确的文本重试。

第三步：工具注册

将 4 个工具（bash + 三个文件工具）定义为 Anthropic API 要求的 JSON Schema 格式：

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"],
        },
    },
]

Schema 就是工具的"说明书"。模型通过 description 判断该用哪个工具，通过 properties 和 required 知道怎么传参。

第四步：工具分发表

上一篇 Bash 是唯一的工具，硬编码就行。现在有 4 个工具了，需要一个分发表：

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"]),
}

一个字典，key 是工具名，value 是处理函数。新增工具只需要加一行，不用改循环逻辑。

Agent Loop 的工具执行部分也要相应改造，从硬编码 Bash 变成查表分发：

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}:")
        print(output[:200])
        results.append({
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": output,
        })
messages.append({"role": "user", "content": results})

关键变化：TOOL_HANDLERS.get(block.name) 根据模型返回的工具名动态查找处理函数。未知工具名也不会崩溃，而是返回错误信息让模型自行调整。

这个模式很重要——Agent Loop 不需要知道每个工具的细节，只需要一张映射表。后面不管加多少工具，循环本身一行都不用改。

TOOL_HANDLERS 分发表

运行效果

案例 1：读文件（bash + read_file 组合）

❯ 读一下当前目录有哪些文件，然后读取 agentic-demo.py 的前 10 行
> bash:
agentic-demo.py
harness-agent-loop.md
harness-tools.md
harness.py
hello.txt
images
llm-claude.py
...
> read_file:
import json
import os

import anthropic
import requests
import subprocess
from dotenv import load_dotenv
from pathlib import Path
load_dotenv()

... (147 more lines)

模型自动选择了两个工具：先用 bash 执行 ls 列出文件，再用 read_file 读取指定文件的前 10 行。

没人告诉它先列目录再读文件——模型自己规划了调用顺序。这就是 Agent Loop 的核心能力。

案例 2：写文件 + 编辑（write_file + edit_file 链）

❯ 在当前目录创建 hello.py，内容是 print("hello world")，再把 hello 改成 hi
> write_file:
Wrote 21 bytes to hello.py
> edit_file:
Edited hello.py

两步完成：先 write_file 创建文件，再 edit_file 精确替换。模型知道用 edit_file 而不是重新 write_file 整个文件——因为 Schema 里的 description 告诉了它 edit_file 是"替换文本"的工具。

案例 3：创建并运行（write_file + bash）

❯ 创建 fizzbuzz.py，实现 fizzbuzz(1..20)，然后运行它
> write_file:
Wrote 182 bytes to fizzbuzz.py
> bash:
1
2
Fizz
4
Buzz
Fizz
7
8
Fizz
Buzz
11
Fizz
13
14
FizzBuzz
16
17
Fizz
19
Buzz

先写代码，再执行——这已经是一个完整的 "写代码 → 执行 → 看结果" 循环了。距离真正的 coding agent 只差一个自动修 bug 的能力。

案例 4：路径安全拦截

这个最有意思：

❯ 创建一个 /tmp/hello.py，内容是 print("hello world")，然后把 hello 改成 hi
> write_file:
Error: Path escapes workspace: /tmp/hello.py
> bash:
print("hello world")
> bash:
print("hi world")

write_file 被 safe_path 拦截了——/tmp 不在工作区内。但模型没有卡死，而是立刻切换策略改用 Bash 完成任务。

这体现了两个关键能力：

安全边界生效 — 文件工具只能操作工作区内的文件
模型的自适应能力 — 工具报错后自动回退到 Bash，不需要人类干预

完整代码

import json
import os
import subprocess

import anthropic
from dotenv import load_dotenv
from pathlib import Path

load_dotenv()

client = anthropic.Anthropic(
    api_key=os.environ["API_KEY"],
    base_url=os.environ["BASE_URL"],
)
MODEL = "claude-sonnet-4.6"
WORKDIR = Path(os.getcwd())
SYSTEM = f"You are a coding agent at {WORKDIR}. Use tools to solve tasks. Act, don't explain."


# ── 工具实现 ──

def 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=os.getcwd(),
                           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)"
    except (FileNotFoundError, OSError) as e:
        return f"Error: {e}"


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


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}"


# ── 工具注册 & 分发 ──

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"]}},
]

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"]),
}


# ── Agent Loop ──

def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL,
            system=SYSTEM,
            messages=messages,
            tools=TOOLS,
            max_tokens=1000,
        )
        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}:")
                print(output[:200])
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        messages.append({"role": "user", "content": results})


BANNER = """\
\033[1;35m ▐▛███▜▌\033[0m   \033[1mHuanCode Harness s01\033[0m
\033[1;35m▝▜█████▛▘\033[0m  Agent Loop · 最小 Agent 循环
\033[1;35m  ▘▘ ▝▝\033[0m    输入 q 退出

\033[90m─────────────────────────────────────────────\033[0m"""

if __name__ == "__main__":
    print(BANNER)
    history = []
    while True:
        try:
            query = input("\033[1;35m❯\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()

小结

这一篇我们做了三件事：

增加了 3 个文件工具 — read_file、write_file、edit_file，都经过 safe_path 路径安全检查
引入了工具分发表 — TOOL_HANDLERS 字典让 Agent Loop 可以动态路由到任意工具，新增工具只需加一行
验证了模型的自适应能力 — 工具报错时模型能自动切换策略，不会卡死

到这里，我们的 Agent 已经具备了 coding agent 的基本能力：执行命令、读写文件、精确编辑代码。

但它还缺少几个关键能力：不会拆分任务、不会跟踪进度、对话太长会撑爆 context window。下一篇，我们继续进化。

Harness实战：给Agent装上文件操作工具

On this page