AI Agent 开发教程：用 Python 写一个能搜索、调用工具、保存状态的 Agent

一个 Agent 就是三样东西：一个循环、几个工具、一份能落盘的状态。这篇教程用 Python 把这三样从零写出来，只依赖官方 anthropic SDK，不用任何 Agent 框架。写完你会得到一个能联网搜索、读写文件、跨运行记住事实、崩溃后从检查点续跑的 Agent，代码约 200 行。

文中所有输出都来自 2026 年 9 月 17 日的真实运行，token 数和耗时照抄日志，没有润色。完整代码可以直接下载：minimal_search_agent.py。

如果你想先看架构再动手，《AI Agent 开发实战》第 3 章"工具调用基础" 讲的是这里每一段代码背后的原理。这篇是那章的"能跑起来"版本。

先跑起来

环境要求：Python 3.10 以上，一个 Anthropic API key。

pip install "anthropic>=1.6"
export ANTHROPIC_API_KEY=sk-ant-...
curl -O https://waylandz.com/examples/minimal_search_agent.py

python minimal_search_agent.py --task "查一下 MCP 2026-07-28 规范里 Tasks 从核心变成了什么，把结论（3 句以内，附来源链接）写进 notes/mcp-tasks.md，并用 remember 记住当前规范版本号。"

跑完后当前目录下会多出一个 agent_workspace/，里面有 Agent 写的笔记、它记住的事实和整段对话的检查点。下面按"工具、循环、状态"三块拆开讲代码，然后看四次运行的结果。

第一块：工具

工具分两种，区别在于谁来执行。

服务端工具由 Anthropic 在自己的服务器上执行，结果直接出现在同一个响应里。联网搜索就是这种，声明只有一行：

WEB_SEARCH = {"type": "web_search_20260209", "name": "web_search", "max_uses": 5}

不需要搜索引擎的 API key，不需要你写任何搜索代码。max_uses 限制一轮里最多搜几次，这是防止失控最便宜的一道闸。

客户端工具由你的进程执行。模型只是提出"我想调用 write_file，参数是这些"，你的代码真正去写文件，再把结果交回去。这个 Agent 有三个客户端工具：

定义是一段 JSON Schema。三点值得注意：

{
    "name": "write_file",
    "description": "Create or overwrite a UTF-8 text file inside the workspace.",
    "input_schema": {
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "Relative path"},
            "content": {"type": "string"},
        },
        "required": ["path", "content"],
        "additionalProperties": False,
    },
    "strict": True,
}

1. description 是写给模型看的，它决定模型什么时候选这个工具。写得含糊，模型就会选错或漏掉参数。
2. strict: True 加上 additionalProperties: False，API 会保证模型给出的参数严格符合 schema，你的代码不用再做类型检查。
3. 路径都是相对工作区的。执行前用 safe_path() 把 ../ 这类逃逸拦下来，拦下来的错误会原样返回给模型（后面第四次运行会看到模型怎么处理）。

执行工具的代码就是一个 if 链：

def run_tool(name: str, args: dict) -> str:
    if name == "read_file":
        return safe_path(args["path"]).read_text(encoding="utf-8")
    if name == "write_file":
        target = safe_path(args["path"])
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(args["content"], encoding="utf-8")
        return f"wrote {len(args['content'])} chars to {args['path']}"
    if name == "remember":
        memory = load_memory()
        memory[args["key"]] = args["value"]
        MEMORY_FILE.write_text(json.dumps(memory, ensure_ascii=False, indent=2), encoding="utf-8")
        return f"remembered {args['key']}"
    raise ValueError(f"unknown tool {name}")

返回值是给模型看的字符串。写文件成功不要返回空，返回"写了多少字到哪里"，模型下一步才有依据。

第二块：循环

Agent 的核心就是这个 while True。每一圈：调一次模型，看它为什么停下来，按停下来的原因分支处理。

while True:
    response = client.messages.create(
        model="claude-opus-5",
        max_tokens=16000,
        system=system,
        tools=[WEB_SEARCH, *CLIENT_TOOLS],
        messages=messages,
    )
    messages.append({"role": "assistant", "content": to_plain(response.content)})

    if response.stop_reason == "pause_turn":
        save_session(messages, "running", rounds)
        continue                      # 服务端搜索到了迭代上限，原样重发即可继续

    if response.stop_reason == "refusal":
        save_session(messages, "refused", rounds)
        return                        # 模型拒绝了，把原因打出来，不要重试

    if response.stop_reason == "max_tokens":
        save_session(messages, "truncated", rounds)
        return                        # 输出被截断，加大上限或拆任务

    if response.stop_reason != "tool_use":
        save_session(messages, "done", rounds)
        break                         # 没有工具调用了，就是最终回答

    results = []
    for block in response.content:
        if block.type != "tool_use":
            continue                  # server_tool_use 已经在服务端执行完了，跳过
        try:
            output = run_tool(block.name, block.input)
            results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})
        except Exception as exc:
            results.append({"type": "tool_result", "tool_use_id": block.id,
                            "content": f"Error: {exc}", "is_error": True})

    messages.append({"role": "user", "content": results})
    rounds += 1
    save_session(messages, "running", rounds)

stop_reason 是整个循环的开关，五个值各有含义：

初学者最常犯的错是只处理 tool_use 和 end_turn。pause_turn 只在用了服务端工具时出现，第一次遇到会以为 Agent"没说完就停了"。

另外两条规则：

• 一个响应里可能有多个 tool_use。模型会并行请求几个工具（第一次运行里它一次就发了写文件和记忆两个调用）。所有结果要放进同一条 user 消息里返回，拆成多条会让模型以后不再并行调用。
• 工具报错要返回给模型，不要在你这边崩掉。is_error: True 加上错误文本，模型会自己决定换路径还是放弃。

第三块：状态

状态有两层，寿命不同。

对话检查点（session.json）：每执行完一轮工具，就把整个 messages 列表和当前状态写盘。写法是先写临时文件再 os.replace() 覆盖，这样进程在写到一半时被杀，磁盘上要么是旧的完整检查点，要么是新的完整检查点，不会是半个。

def save_session(messages, status, rounds):
    tmp = SESSION_FILE.with_suffix(".tmp")
    tmp.write_text(json.dumps({"status": status, "rounds": rounds, "messages": messages}, ensure_ascii=False))
    os.replace(tmp, SESSION_FILE)

--resume 就是把这个文件读回来，从上次停下的地方接着调模型。

状态文件放在 agent_workspace/.agent/ 下，safe_path() 拒绝工具读写这个目录。不然模型一次误写 memory.json，下一次启动就会崩在读取记忆上。

跨运行记忆（memory.json）：remember 工具写的键值对。每次启动时读出来拼进 system prompt，这样新任务不用重新搜索已经知道的事。这是最简单的记忆实现，够用来理解原理；生产系统里的记忆架构见第 8 章。

为什么检查点存的是 messages 而不是"任务进度"？因为对模型来说，对话历史就是全部状态。工具结果在里面，它自己的中间判断也在里面。只要把这个列表原样送回去，模型就能从任何一步继续。这也是为什么保存前要用 model_dump() 把 SDK 对象转成普通 JSON。

四次真实运行

下面四次运行用的脚本版本，状态文件还直接放在工作区根目录；后来把它们移进了 .agent/ 并加了保留路径检查，循环逻辑没有变。

第一次：完成任务，然后故意崩溃

--crash-after 1 让进程在第一轮工具执行完之后直接 os._exit(1)，模拟断电或 OOM。

[call 1] stop_reason=tool_use in=22441 out=1391
  tool write_file({"path": "notes/mcp-tasks.md", "content": "# MCP 2026-07-28: Tasks 的定位变化\n\n在 20) ok
  tool remember({"key": "mcp_spec_version", "value": "当前 MCP 规范版本：2026-07-28（latest）；Tasks 已从核心移) ok
--- simulated crash after 1 tool round(s); run with --resume ---
exit=1

只调了一次模型。这一次调用里，服务端搜索已经跑完（in=22441，比其他调用大一倍以上，因为搜索结果被塞进了上下文），然后模型一次性发出了写文件和记忆两个客户端工具调用。两个工具都执行成功，检查点写盘，进程退出。

第二次：从检查点续跑

resuming after 1 tool round(s), 3 messages on disk
[call 1] stop_reason=end_turn in=9673 out=359

=== result ===
结论：在 2026-07-28 规范中，Tasks 从实验性的协议核心移出，成为正式扩展 `io.modelcontextprotocol/tasks`（SEP-2663），
采用轮询式 `tasks/get` 并新增 `tasks/update`……

calls=1 tool_rounds=1 searches=0 input_tokens=9673 output_tokens=359 cache_read=0 elapsed=8.3s

searches=0。续跑没有重新搜索，也没有重新写文件，因为搜索结果和工具结果都在检查点里。模型看到工具已经返回成功，直接给出最终总结。这次续跑省掉的是一次联网搜索和两次工具执行。

这个保证有边界。模拟的崩溃点在检查点写入之后，所以什么都没重做。检查点是整轮工具执行完才写盘的，如果崩溃发生在工具已经执行、检查点还没落盘的那一瞬间，续跑会把这一轮工具再执行一遍。这个入门脚本不保证任意崩溃时刻的副作用只执行一次。文件被覆盖问题不大，但如果工具是"发邮件"或"下单"，重来一次就是事故；要做到那一步，得在执行工具之前先记录意图、执行之后再记录结果；而且这两条日志仍然堵不住"外部操作已经成功、结果还没记下来"这个窗口，最后还得靠工具本身幂等，或者续跑前先核验外部结果。完整讨论在《长时间运行的 Agent Loop 里的中途检查点》。

顺带一提，笔记里的结论是对的。Tasks 在 2026-07-28 规范里确实从实验性核心移到了 io.modelcontextprotocol/tasks 扩展，可以对照官方发布说明。

第三次：新任务，用记忆不用搜索

[call 1] stop_reason=tool_use in=6431 out=100
  tool read_file({"path": "notes/mcp-tasks.md"}) ok
[call 2] stop_reason=end_turn in=6909 out=575

=== result ===
没有搜索，直接用记忆和本地文件回答。
上次记住的规范版本号：MCP 当前规范版本：2026-07-28（标记为 latest）……

calls=2 tool_rounds=1 searches=0 input_tokens=13340 output_tokens=675 cache_read=0 elapsed=12.0s

启动时 memory.json 里的内容被拼进了 system prompt，所以模型直接知道版本号。它只调了一次 read_file 把笔记读出来，没有联网。两次调用合计 13,340 输入 token，比第一次运行的一次调用还少，因为上下文里没有搜索结果。

第四次：工具报错，模型自己找路

任务故意要求写到工作区外面：

[call 1] stop_reason=tool_use in=6462 out=274
  tool read_file({"path": "notes/mcp-tasks.md"}) ok
[call 2] stop_reason=tool_use in=8624 out=620
  tool write_file failed: path '../backup/mcp-tasks.md' is outside the workspace
[call 3] stop_reason=tool_use in=9281 out=627
  tool write_file({"path": "backup/mcp-tasks.md", ...}) ok
[call 4] stop_reason=tool_use in=9934 out=58
  tool read_file({"path": "backup/mcp-tasks.md"}) ok
[call 5] stop_reason=tool_use in=10370 out=150
  tool remember({"key": "workspace_write_boundary", "value": "write_file 无法写工作区之外……"}) ok
[call 6] stop_reason=end_turn in=10537 out=365

calls=6 tool_rounds=5 searches=0 input_tokens=55208 output_tokens=2094 cache_read=0 elapsed=38.8s

第 2 次调用被 safe_path() 拦下，错误以 is_error: True 返回。模型没有反复重试同一个路径，而是改写到工作区内的 backup/，回读校验内容一致，然后主动 remember 了这条边界，最后在回答里明确说"没做到的部分"是什么、为什么。这是你希望 Agent 具备的错误行为：一次失败，一次绕行，把限制记下来，如实汇报。

代价是六次模型调用、55,208 输入 token，是第三次运行的四倍。错误恢复不是免费的，每绕一次路上下文就长一截。这也是为什么工具的错误信息要写得具体：path is outside the workspace 让模型一次就明白该怎么改，一句 Error 可能要它试三四次。

一个没料到的发现：搜索工具带了一个沙箱

看第四次运行的检查点，第 1 次调用里除了 read_file，还有一个我没定义过的调用：

SERVER_TOOL_USE: bash_code_execution
  {"command": "pwd; ls -la; ls -la notes 2>/dev/null; ls -la .. 2>&1 | head -20"}

web_search_20260209 这个版本的搜索工具内置了"动态过滤"，实现方式是在服务端给模型一个代码执行沙箱来筛选搜索结果。模型知道自己有这个沙箱，于是在处理"能不能写到 ../backup"时，先去沙箱里 ls 了一圈，然后在最终回答里说"bash 沙箱里根本看不到 notes/ 目录"。

沙箱当然看不到，那是 Anthropic 服务器上的一个容器，不是你的工作区。模型的结论没错（它最后正确判断了限制来自 write_file），但它多花了一次调用去探一个和任务无关的环境。

教训是：你给模型的工具面，比你自己写的那几个工具要大。声明一个服务端工具，可能连带引入它的附属能力。写 system prompt 时要把这件事说清楚（比如"沙箱里的文件系统与工作区无关"），或者在不需要搜索的任务里干脆不挂搜索工具。

这几次运行花了多少

按 2026 年 9 月 Claude Opus 5 的公开价格（输入 $5 / 百万 token，输出 $25 / 百万 token，联网搜索另按次计费）粗算：

四次合计不到一美元。注意 cache_read=0：这个教程没有开 prompt cache，因为对话都很短。一旦 system prompt 和工具定义超过一两千 token，就应该在它们后面加一个 cache_control 断点，后续调用的输入费用能降一个量级。做法见第 39 章"Prompt Cache 稳定性"。

这 200 行没有覆盖的东西

按重要性排：

1. 上下文压缩。对话长了以后 messages 会超过窗口，需要摘要或截断旧的工具结果。原理和取舍在第 35 章和第 36 章。
2. 超时与卡循环检测。这个循环没有上限，模型如果反复调同一个工具会一直烧钱。生产版本至少要有最大轮数和"连续三次相同调用就停"的守卫，见第 42 章和第 43 章。
3. 拒绝后的备选模型。API 支持在 refusal 时自动切换到备用模型（fallbacks 参数），这里没开，因为教程要让你看到 refusal 分支本身。
4. 工具从哪来。这里的三个工具是手写的。真实项目里工具可能来自 MCP server、命令行程序或 Skills 文件，三种方式各有代价，我用同一个任务做了对比：《MCP、WebMCP、CLI、Skills：同一个任务该选哪种工具？》。

常见问题

能换成别的模型吗？ 可以，改 MODEL 常量。但 stop_reason 的取值、strict 模式和服务端搜索工具都是 Anthropic API 的特性，换供应商需要改循环里的分支逻辑，不只是换模型名。

为什么不用 SDK 自带的 tool runner？ SDK 有一个 tool_runner 能帮你跑这个循环。教程手写是为了让你看清每个分支；看懂之后用 runner 少写 30 行。在我核对的 SDK 文档里，runner 不会自动续跑 pause_turn，这个分支要自己留着。

session.json 会不会无限变大？ 会。每轮工具结果都留在里面。教程里的任务短，几十 KB 就结束；长任务要么压缩，要么换成按轮次追加的日志。

搜索结果能存下来复用吗？ 检查点里存的是搜索结果的加密引用，续跑同一个对话可以复用，但不能拿出来单独读。要保留原文，让模型用 write_file 写下来。