Agent开发|从0实现Agent(一):50行代码实现MiniClaudeCode(工具与执行篇)

admin 2026-03-27 03:43:54 网络安全文章 来源:ZONE.CI 全球网 0 阅读模式

文章总结: 本文是一个关于从零开始开发AI智能体(Agent)的技术系列文章。它通过复现learn-claude-code项目,逐步讲解如何手动实现一个类似ClaudeCode的智能体。内容涵盖了从基础的循环与工具使用,到规划与知识管理、任务系统、后台任务处理,再到高级的自治执行与团队协作等多个阶段,旨在帮助读者深入理解智能体的内部工作原理和开发流程。 综合评分: 85 文章分类: AI安全,CTF,二进制安全,WEB安全,技术标准


cover_image

Agent开发|从0实现Agent(一):50行代码实现Mini Claude Code(工具与执行篇)

原创

Real返璞归真 Real返璞归真

Real返璞归真

2026年3月18日 20:26 山东

公众号

欢迎关注公众号【Real返璞归真】,我们将不定期分享CTF竞赛、二进制安全、JS/安卓逆向、AI安全等领域的前沿知识与技术内容。

前言

简介

Agent = LLM + 工具 + 循环

模型就是智能体,我们的工作就是给它工具,然后让开。

本系列文章将复现learn-claude-code项目,从0构建nano Claude Code-like agent

手动实现一个类似Claude Code的Agent,学习Agent的内部工作原理和开发流程。

学习路线

学习路线:

第一阶段: 循环                       第二阶段: 规划与知识
==================                   ==============================
s01  Agent 循环              [1]     s03  TodoWrite               [5]
     while + stop_reason                  TodoManager + nag 提醒
     |                                    |
     +-> s02  Tool Use            [4]     s04  子智能体             [5]
              dispatch map: name->handler     每个子智能体独立 messages[]
                                              |
                                         s05  Skills               [5]
                                              SKILL.md 通过 tool_result 注入
                                              |
                                         s06  Context Compact      [5]
                                              三层上下文压缩

第三阶段: 持久化                     第四阶段: 团队
==================                   =====================
s07  任务系统                [8]     s09  智能体团队             [9]
     文件持久化 CRUD + 依赖图             队友 + JSONL 邮箱
     |                                    |
s08  后台任务                [6]     s10  团队协议               [12]
     守护线程 + 通知队列                  关机 + 计划审批 FSM
                                          |
                                     s11  自治智能体             [14]
                                          空闲轮询 + 自动认领
                                     |
                                     s12  Worktree 隔离          [16]
                                          任务协调 + 按需隔离执行通道

                                     [N] = 工具数量

课程目录:

| 课程 | 主题 | 格言 | | — | — | — | | s01 | Agent 循环 | One loop & Bash is all you need | | s02 | Tool Use | 加一个工具, 只加一个 handler | | s03 | TodoWrite | 没有计划的 agent 走哪算哪 | | s04 | 子智能体 | 大任务拆小, 每个小任务干净的上下文 | | s05 | Skills | 用到什么知识, 临时加载什么知识 | | s06 | Context Compact | 上下文总会满, 要有办法腾地方 | | s07 | 任务系统 | 大目标要拆成小任务, 排好序, 记在磁盘上 | | s08 | 后台任务 | 慢操作丢后台, agent 继续想下一步 | | s09 | 智能体团队 | 任务太大一个人干不完, 要能分给队友 | | s10 | 团队协议 | 队友之间要有统一的沟通规矩 | | s11 | 自治智能体 | 队友自己看看板, 有活就认领 | | s12 | Worktree + 任务隔离 | 各干各的目录, 互不干扰 |

参考资料

learn-claude-code:https://github.com/shareAI-lab/learn-claude-code

12 个递进式课程, 从简单循环到隔离化的自治执行。每个课程添加一个机制。每个机制有一句格言。

s01“One loop & Bash is all you need” — 一个工具 + 一个循环 = 一个智能体

s02“加一个工具, 只加一个 handler” — 循环不用动, 新工具注册进 dispatch map 就行

s03“没有计划的 agent 走哪算哪” — 先列步骤再动手, 完成率翻倍

s04“大任务拆小, 每个小任务干净的上下文” — 子智能体用独立 messages[], 不污染主对话

s05“用到什么知识, 临时加载什么知识” — 通过 tool_result 注入, 不塞 system prompt

s06“上下文总会满, 要有办法腾地方” — 三层压缩策略, 换来无限会话

s07“大目标要拆成小任务, 排好序, 记在磁盘上” — 文件持久化的任务图, 为多 agent 协作打基础

s08“慢操作丢后台, agent 继续想下一步” — 后台线程跑命令, 完成后注入通知

s09“任务太大一个人干不完, 要能分给队友” — 持久化队友 + 异步邮箱

s10“队友之间要有统一的沟通规矩” — 一个 request-response 模式驱动所有协商

s11“队友自己看看板, 有活就认领” — 不需要领导逐个分配, 自组织

s12“各干各的目录, 互不干扰” — 任务管目标, worktree 管目录, 按 ID 绑定

下载项目与环境安装

Ubuntu22.04

在VMware虚拟机中安装Ubuntu22.04操作系统,作为基础环境。

基础环境安装

sudo apt update

sudo apt install net-tools
sudo apt install vim
sudo apt install curl
sudo apt install git
sudo apt install python3-pip

curl -fsSL https://deb.nodesource.com/setup_20.x | sudo -E bash -
sudo apt-get install -y nodejs

下载项目

git clone https://github.com/shareAI-lab/learn-claude-code
cd learn-claude-code
pip install -r requirements.txt

配置大模型API

配置大模型API作为agent的基座:

cp .env.example .env

编辑.env文件:

# API Key (required)
# Get yours at: https://console.anthropic.com/
ANTHROPIC_API_KEY=sk-ant-xxx

# Model ID (required)
MODEL_ID=claude-sonnet-4-6

# Base URL (optional, for Anthropic-compatible providers)
# ANTHROPIC_BASE_URL=https://api.anthropic.com

# =============================================================================
#  Anthropic-compatible providers
#
#  Provider         MODEL_ID              SWE-bench  TB2     Base URL
#  ---------------  --------------------  ---------  ------  -------------------
#  Anthropic        claude-sonnet-4-6     79.6%      59.1%   (default)
#  MiniMax          MiniMax-M2.5          80.2%        -     see below
#  GLM (Zhipu)      glm-5                77.8%        -     see below
#  Kimi (Moonshot)  kimi-k2.5            76.8%        -     see below
#  DeepSeek         deepseek-chat        73.0%        -     see below
#                   (V3.2)
#
#  SWE-bench = SWE-bench Verified (Feb 2026)
#  TB2       = Terminal-Bench 2.0 (Feb 2026)
# =============================================================================

# ---- International ----

# MiniMax          https://www.minimax.io
# ANTHROPIC_BASE_URL=https://api.minimax.io/anthropic
# MODEL_ID=MiniMax-M2.5

# GLM (Zhipu)      https://z.ai
# ANTHROPIC_BASE_URL=https://api.z.ai/api/anthropic
# MODEL_ID=glm-5

# Kimi (Moonshot)  https://platform.moonshot.ai
# ANTHROPIC_BASE_URL=https://api.moonshot.ai/anthropic
# MODEL_ID=kimi-k2.5

# DeepSeek         https://platform.deepseek.com
# ANTHROPIC_BASE_URL=https://api.deepseek.com/anthropic
# MODEL_ID=deepseek-chat

# ---- China mainland ----

# MiniMax          https://platform.minimax.io
# ANTHROPIC_BASE_URL=https://api.minimaxi.com/anthropic
# MODEL_ID=MiniMax-M2.5

# GLM (Zhipu)      https://open.bigmodel.cn
# ANTHROPIC_BASE_URL=https://open.bigmodel.cn/api/anthropic
# MODEL_ID=glm-5

# Kimi (Moonshot)  https://platform.moonshot.cn
# ANTHROPIC_BASE_URL=https://api.moonshot.cn/anthropic
# MODEL_ID=kimi-k2.5

# DeepSeek (no regional split, same endpoint globally)
# ANTHROPIC_BASE_URL=https://api.deepseek.com/anthropic
# MODEL_ID=deepseek-chat

可以根据实际使用的模型填写,国内推荐使用DeepSeek API

# API Key (required)
# Get yours at: https://console.anthropic.com/
ANTHROPIC_API_KEY=sk-xxxxxxxxxxxxxxxxxx

# Model ID (required)
MODEL_ID=deepseek-chat

# Base URL (optional, for Anthropic-compatible providers)
ANTHROPIC_BASE_URL=https://api.deepseek.com/anthropic

查看课程文档

cd web && npm install && npm run dev   # http://localhost:3000

访问http://localhost:3000

即可打开官方课程文档。

基础知识

核心模式

所有 Agent 共享同一个循环:调用模型、执行工具、回传结果。生产级系统会在其上叠加策略、权限和生命周期层。

核心模式代码:

while True:
    response = client.messages.create(messages=messages, tools=tools)
    if response.stop_reason != "tool_use":
        break
    for tool_call in response.content:
        result = execute_tool(tool_call.name, tool_call.input)
        messages.append(result)

Anthropic API

用户(user)请求格式:

response = client.messages.create(
    model=MODEL,           # 1. 指定模型
    system=SYSTEM,         # 2. 设置“系统提示词”
    messages=messages,     # 3. 提供“对话历史”
    tools=TOOLS,           # 4. (可选)定义“可用工具”
    max_tokens=8000,       # 5. 设置“最大回复长度”
)

它会构造如下请求:

{
  "model": "deepseek-chat",
  "system": "你是一个有用的AI助手,可以调用工具。",  // 系统提示词,设定AI的身份和规则
  "messages": [          // 对话历史,必须交替出现 user 和 assistant 的消息 [citation:8]
    {"role": "user", "content": "帮我看看当前目录有什么文件?"}
  ],
  "tools": [             // 可供AI使用的工具列表
    {
      "name": "bash",
      "description": "运行一个bash命令",
      "input_schema": {  // 工具的参数格式要求
        "type": "object",
        "properties": {
          "command": {"type": "string"}
        },
        "required": ["command"]
      }
    }
  ],
  "max_tokens": 8000     // 告诉AI最多可以说多少个词
}

LLM响应格式:

{
  "id": "msg_123",
  "type": "message",
  "role": "assistant",
  "content": [           // ← 重点!这是一个内容块数组
    {
      "type": "text",
      "text": "好的,我来看看当前目录下有什么文件。需要运行 `ls` 命令。"
    },
    {
      "type": "tool_use",  // ← AI决定使用工具!
      "id": "toolu_111",
      "name": "bash",
      "input": {
        "command": "ls -la"
      }
    }
  ],
  "stop_reason": "tool_use",  // ← AI暂停的原因:因为要用工具
  "stop_sequence": null,
  "usage": {                   // ← 本次对话消耗的token数量
    "input_tokens": 120,
    "output_tokens": 45
  }
}

s01:Agent循环 – Bash is All You Need

The minimal agent kernel is a while loop + one tool

最小化的agent内核是while循环 + 1个工具

问题

语言模型能推理代码,但碰不到真实世界—不能读文件、跑测试、看报错。没有循环,每次工具调用你都得手动把结果粘回去。你自己就是那个循环。

解决方案

+--------+      +-------+      +---------+
|  User  | ---> |  LLM  | ---> |  Tool   |
| prompt |      |       |      | execute |
+--------+      +---+---+      +----+----+
                    ^                |
                    |   tool_result  |
                    +----------------+
                    (loop until stop_reason != "tool_use")

一个退出条件控制整个流程。循环持续运行,直到模型不再调用工具。

工作原理

完整代码(不到 30 行, 这就是整个智能体):

def agent_loop(query):
    messages = [{"role": "user", "content": query}]
    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":
                output = run_bash(block.input["command"])
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        messages.append({"role": "user", "content": results})

工作过程:

  1. 用户 prompt 作为第一条消息:
   messages.append({"role": "user", "content": query})
  1. 将消息和工具定义一起发给 LLM:
   response = client.messages.create(
       model=MODEL, system=SYSTEM, messages=messages,
       tools=TOOLS, max_tokens=8000,
   )
  1. 保存LLM的响应内容到对话上下文,通过 stop_reason 检查模型是否调用工具:
   messages.append({"role": "assistant", "content": response.content})
   if response.stop_reason != "tool_use":
       return
  1. 执行每个工具调用,收集结果,追加到 user 消息。回到第 2 步。
   results = []
   for block in response.content:
       if block.type == "tool_use":
           output = run_bash(block.input["command"])
           results.append({
               "type": "tool_result",
               "tool_use_id": block.id,
               "content": output,
           })
   messages.append({"role": "user", "content": results})

完整代码

#!/usr/bin/env python3

"""

s01_agent_loop.py - The Agent Loop

The entire secret of an AI coding agent in one pattern:

    while stop_reason == "tool_use":

        response = LLM(messages, tools)

        execute tools

        append results

    +----------+      +-------+      +---------+

    |   User   | ---> |  LLM  | ---> |  Tool   |

    |  prompt  |      |       |      | execute |

    +----------+      +---+---+      +----+----+

                          ^               |

                          |   tool_result |

                          +---------------+

                          (loop continues)

This is the core loop: feed tool results back to the model

until the model decides to stop. Production agents layer

policy, hooks, and lifecycle controls on top.

"""

import os

import subprocess

from anthropic import Anthropic

from dotenv import load_dotenv

load_dotenv(override=True)

if os.getenv("ANTHROPIC_BASE_URL"):

    os.environ.pop("ANTHROPIC_AUTH_TOKEN", None)

client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))

MODEL = os.environ["MODEL_ID"]

SYSTEM = f"You are a coding agent at {os.getcwd()}. Use bash to solve tasks. Act, don't explain."

TOOLS = [{

    "name": "bash",

    "description": "Run a shell command.",

    "input_schema": {

        "type": "object",

        "properties": {"command": {"type": "string"}},

        "required": ["command"],

    },

}]

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

# -- The core pattern: a while loop that calls tools until the model stops --

def agent_loop(messages: list):

    while True:

        response = client.messages.create(

            model=MODEL, system=SYSTEM, messages=messages,

            tools=TOOLS, max_tokens=8000,

        )

        # Append assistant turn

        messages.append({"role": "assistant", "content": response.content})

        # If the model didn't call a tool, we're done

        if response.stop_reason != "tool_use":

            return

        # Execute each tool call, collect results

        results = []

        for block in response.content:

            if block.type == "tool_use":

                print(f"\033[33m$ {block.input['command']}\033[0m")

                output = run_bash(block.input["command"])

                print(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[36ms01 >> \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()

效果演示

cd learn-claude-code
python3 agents/s01_agent_loop.py

尝试提示词:

Create a file called hello.py that prints "Hello, World!"
List all Python files in this directory
What is the current git branch?
Create a directory called test_output and write 3 files in it

深入探索

Q:为什么仅靠bash就可以实现agent?

A:Bash 能读写文件、运行任意程序、在进程间传递数据、管理文件系统。任何额外的工具(read_file、write_file 等)都只是 bash 已有能力的子集。增加工具并不会解锁新能力,只会增加模型需要理解的接口。模型只需学习一个工具的 schema,实现代码不超过 100 行。这就是最小可行 agent:一个工具,一个循环。

总结

本节,我们通过几十行代码实现了简易的Agent。后续将在此基础上不断引入新的机制。

核心agent的循环非常简单,没有路由器、决策树和工作流引擎,大模型自己决定做什么,代码只是连接模型与工具的管道。

这是一种设计哲学:agent 行为从模型中涌现,而非由框架定义。

s02:工具 – One Handler Per Tool

The loop stays the same; new tools register into the dispatch map

依旧是那个循环,我们将注册新工具到调度系统中。

问题

只有 bash 时, 所有操作都走 shell,导致:

  • cat 内容过长时,截断不可预测
  • sed 遇到特殊字符崩溃
  • 每次 bash 调用都是不受控制的安全攻击面

使用专用工具 (read_file, write_file) 可以在工具层面做路径沙箱。

解决方案

+--------+      +-------+      +------------------+
|  User  | ---> |  LLM  | ---> | Tool Dispatch    |
| prompt |      |       |      | {                |
+--------+      +---+---+      |   bash: run_bash |
                    ^           |   read: run_read |
                    |           |   write: run_wr  |
                    +-----------+   edit: run_edit |
                    tool_result | }                |
                                +------------------+

The dispatch map is a dict: {tool_name: handler_function}.
One lookup replaces any if/elif chain.

工作原理

  1. 为每个工具定义一个处理函数,并定义路径沙箱防止逃逸:
   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:
       text = safe_path(path).read_text()
       lines = text.splitlines()
   &nbsp; &nbsp;&nbsp;if&nbsp;limit&nbsp;and&nbsp;limit < len(lines):
   &nbsp; &nbsp; &nbsp; &nbsp; lines = lines[:limit]
   &nbsp; &nbsp;&nbsp;return&nbsp;"\n".join(lines)[:50000]

   def&nbsp;run_write(path: str, content: str)&nbsp;-> str:

   &nbsp; &nbsp;&nbsp;try:

   &nbsp; &nbsp; &nbsp; &nbsp; fp = safe_path(path)

   &nbsp; &nbsp; &nbsp; &nbsp; fp.parent.mkdir(parents=True, exist_ok=True)

   &nbsp; &nbsp; &nbsp; &nbsp; fp.write_text(content)

   &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;f"Wrote&nbsp;{len(content)}&nbsp;bytes to&nbsp;{path}"

   &nbsp; &nbsp;&nbsp;except&nbsp;Exception&nbsp;as&nbsp;e:

   &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;f"Error:&nbsp;{e}"

   def&nbsp;run_edit(path: str, old_text: str, new_text: str)&nbsp;-> str:

   &nbsp; &nbsp;&nbsp;try:

   &nbsp; &nbsp; &nbsp; &nbsp; fp = safe_path(path)

   &nbsp; &nbsp; &nbsp; &nbsp; content = fp.read_text()

   &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;old_text&nbsp;not&nbsp;in&nbsp;content:

   &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;f"Error: Text not found in&nbsp;{path}"

   &nbsp; &nbsp; &nbsp; &nbsp; fp.write_text(content.replace(old_text, new_text,&nbsp;1))

   &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;f"Edited&nbsp;{path}"

   &nbsp; &nbsp;&nbsp;except&nbsp;Exception&nbsp;as&nbsp;e:

   &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;f"Error:&nbsp;{e}"
  1. 定义dispatch map将工具名映射至函数:
   TOOL_HANDLERS = {
   &nbsp; &nbsp;&nbsp;"bash": &nbsp; &nbsp; &nbsp;&nbsp;lambda&nbsp;**kw: run_bash(kw["command"]),
   &nbsp; &nbsp;&nbsp;"read_file": &nbsp;lambda&nbsp;**kw: run_read(kw["path"], kw.get("limit")),
   &nbsp; &nbsp;&nbsp;"write_file":&nbsp;lambda&nbsp;**kw: run_write(kw["path"], kw["content"]),
   &nbsp; &nbsp;&nbsp;"edit_file": &nbsp;lambda&nbsp;**kw: run_edit(kw["path"], kw["old_text"],
   &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; kw["new_text"]),
   }
  1. 在循环中按照工具名查找处理函数并调用:
   for&nbsp;block&nbsp;in&nbsp;response.content:
   &nbsp; &nbsp;&nbsp;if&nbsp;block.type ==&nbsp;"tool_use":
   &nbsp; &nbsp; &nbsp; &nbsp; handler = TOOL_HANDLERS.get(block.name)
   &nbsp; &nbsp; &nbsp; &nbsp; output = handler(**block.input)&nbsp;if&nbsp;handler \
   &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;else&nbsp;f"Unknown tool:&nbsp;{block.name}"
   &nbsp; &nbsp; &nbsp; &nbsp; results.append({
   &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;"type":&nbsp;"tool_result",
   &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;"tool_use_id": block.id,
   &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;"content": output,
   &nbsp; &nbsp; &nbsp; &nbsp; })

完整代码

#!/usr/bin/env python3

"""

s02_tool_use.py - Tools

The agent loop from s01 didn't change. We just added tools to the array

and a dispatch map to route calls.

&nbsp; &nbsp; +----------+ &nbsp; &nbsp; &nbsp;+-------+ &nbsp; &nbsp; &nbsp;+------------------+

&nbsp; &nbsp; | &nbsp; User &nbsp; | ---> | &nbsp;LLM &nbsp;| ---> | Tool Dispatch &nbsp; &nbsp;|

&nbsp; &nbsp; | &nbsp;prompt &nbsp;| &nbsp; &nbsp; &nbsp;| &nbsp; &nbsp; &nbsp; | &nbsp; &nbsp; &nbsp;| { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;|

&nbsp; &nbsp; +----------+ &nbsp; &nbsp; &nbsp;+---+---+ &nbsp; &nbsp; &nbsp;| &nbsp; bash: run_bash |

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ^ &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;| &nbsp; read: run_read |

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; | &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;| &nbsp; write: run_wr &nbsp;|

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; +----------+ &nbsp; edit: run_edit |

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; tool_result| } &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;|

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;+------------------+

Key insight: "The loop didn't change at all. I just added tools."

"""

import&nbsp;os

import&nbsp;subprocess

from&nbsp;pathlib&nbsp;import&nbsp;Path

from&nbsp;anthropic&nbsp;import&nbsp;Anthropic

from&nbsp;dotenv&nbsp;import&nbsp;load_dotenv

load_dotenv(override=True)

if&nbsp;os.getenv("ANTHROPIC_BASE_URL"):

&nbsp; &nbsp; os.environ.pop("ANTHROPIC_AUTH_TOKEN",&nbsp;None)

WORKDIR = Path.cwd()

client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))

MODEL = os.environ["MODEL_ID"]

SYSTEM =&nbsp;f"You are a coding agent at&nbsp;{WORKDIR}. Use tools to solve tasks. Act, don't explain."

def&nbsp;safe_path(p: str)&nbsp;-> Path:

&nbsp; &nbsp; path = (WORKDIR / p).resolve()

&nbsp; &nbsp;&nbsp;if&nbsp;not&nbsp;path.is_relative_to(WORKDIR):

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;raise&nbsp;ValueError(f"Path escapes workspace:&nbsp;{p}")

&nbsp; &nbsp;&nbsp;return&nbsp;path

def&nbsp;run_bash(command: str)&nbsp;-> str:

&nbsp; &nbsp; dangerous = ["rm -rf /",&nbsp;"sudo",&nbsp;"shutdown",&nbsp;"reboot",&nbsp;"> /dev/"]

&nbsp; &nbsp;&nbsp;if&nbsp;any(d&nbsp;in&nbsp;command&nbsp;for&nbsp;d&nbsp;in&nbsp;dangerous):

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;"Error: Dangerous command blocked"

&nbsp; &nbsp;&nbsp;try:

&nbsp; &nbsp; &nbsp; &nbsp; r = subprocess.run(command, shell=True, cwd=WORKDIR,

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;capture_output=True, text=True, timeout=120)

&nbsp; &nbsp; &nbsp; &nbsp; out = (r.stdout + r.stderr).strip()

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;out[:50000]&nbsp;if&nbsp;out&nbsp;else&nbsp;"(no output)"

&nbsp; &nbsp;&nbsp;except&nbsp;subprocess.TimeoutExpired:

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;"Error: Timeout (120s)"

def&nbsp;run_read(path: str, limit: int = None)&nbsp;-> str:

&nbsp; &nbsp;&nbsp;try:

&nbsp; &nbsp; &nbsp; &nbsp; text = safe_path(path).read_text()

&nbsp; &nbsp; &nbsp; &nbsp; lines = text.splitlines()

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;limit&nbsp;and&nbsp;limit < len(lines):

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; lines = lines[:limit] + [f"... ({len(lines) - limit}&nbsp;more lines)"]

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;"\n".join(lines)[:50000]

&nbsp; &nbsp;&nbsp;except&nbsp;Exception&nbsp;as&nbsp;e:

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;f"Error:&nbsp;{e}"

def&nbsp;run_write(path: str, content: str)&nbsp;-> str:

&nbsp; &nbsp;&nbsp;try:

&nbsp; &nbsp; &nbsp; &nbsp; fp = safe_path(path)

&nbsp; &nbsp; &nbsp; &nbsp; fp.parent.mkdir(parents=True, exist_ok=True)

&nbsp; &nbsp; &nbsp; &nbsp; fp.write_text(content)

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;f"Wrote&nbsp;{len(content)}&nbsp;bytes to&nbsp;{path}"

&nbsp; &nbsp;&nbsp;except&nbsp;Exception&nbsp;as&nbsp;e:

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;f"Error:&nbsp;{e}"

def&nbsp;run_edit(path: str, old_text: str, new_text: str)&nbsp;-> str:

&nbsp; &nbsp;&nbsp;try:

&nbsp; &nbsp; &nbsp; &nbsp; fp = safe_path(path)

&nbsp; &nbsp; &nbsp; &nbsp; content = fp.read_text()

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;old_text&nbsp;not&nbsp;in&nbsp;content:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;f"Error: Text not found in&nbsp;{path}"

&nbsp; &nbsp; &nbsp; &nbsp; fp.write_text(content.replace(old_text, new_text,&nbsp;1))

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;f"Edited&nbsp;{path}"

&nbsp; &nbsp;&nbsp;except&nbsp;Exception&nbsp;as&nbsp;e:

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return&nbsp;f"Error:&nbsp;{e}"

# -- The dispatch map: {tool_name: handler} --

TOOL_HANDLERS = {

&nbsp; &nbsp;&nbsp;"bash": &nbsp; &nbsp; &nbsp;&nbsp;lambda&nbsp;**kw: run_bash(kw["command"]),

&nbsp; &nbsp;&nbsp;"read_file": &nbsp;lambda&nbsp;**kw: run_read(kw["path"], kw.get("limit")),

&nbsp; &nbsp;&nbsp;"write_file":&nbsp;lambda&nbsp;**kw: run_write(kw["path"], kw["content"]),

&nbsp; &nbsp;&nbsp;"edit_file": &nbsp;lambda&nbsp;**kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),

}

TOOLS = [

&nbsp; &nbsp; {"name":&nbsp;"bash",&nbsp;"description":&nbsp;"Run a shell command.",

&nbsp; &nbsp; &nbsp;"input_schema": {"type":&nbsp;"object",&nbsp;"properties": {"command": {"type":&nbsp;"string"}},&nbsp;"required": ["command"]}},

&nbsp; &nbsp; {"name":&nbsp;"read_file",&nbsp;"description":&nbsp;"Read file contents.",

&nbsp; &nbsp; &nbsp;"input_schema": {"type":&nbsp;"object",&nbsp;"properties": {"path": {"type":&nbsp;"string"},&nbsp;"limit": {"type":&nbsp;"integer"}},&nbsp;"required": ["path"]}},

&nbsp; &nbsp; {"name":&nbsp;"write_file",&nbsp;"description":&nbsp;"Write content to file.",

&nbsp; &nbsp; &nbsp;"input_schema": {"type":&nbsp;"object",&nbsp;"properties": {"path": {"type":&nbsp;"string"},&nbsp;"content": {"type":&nbsp;"string"}},&nbsp;"required": ["path",&nbsp;"content"]}},

&nbsp; &nbsp; {"name":&nbsp;"edit_file",&nbsp;"description":&nbsp;"Replace exact text in file.",

&nbsp; &nbsp; &nbsp;"input_schema": {"type":&nbsp;"object",&nbsp;"properties": {"path": {"type":&nbsp;"string"},&nbsp;"old_text": {"type":&nbsp;"string"},&nbsp;"new_text": {"type":&nbsp;"string"}},&nbsp;"required": ["path",&nbsp;"old_text",&nbsp;"new_text"]}},

]

def&nbsp;agent_loop(messages: list):

&nbsp; &nbsp;&nbsp;while&nbsp;True:

&nbsp; &nbsp; &nbsp; &nbsp; response = client.messages.create(

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; model=MODEL, system=SYSTEM, messages=messages,

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; tools=TOOLS, max_tokens=8000,

&nbsp; &nbsp; &nbsp; &nbsp; )

&nbsp; &nbsp; &nbsp; &nbsp; messages.append({"role":&nbsp;"assistant",&nbsp;"content": response.content})

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;response.stop_reason !=&nbsp;"tool_use":

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;return

&nbsp; &nbsp; &nbsp; &nbsp; results = []

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;for&nbsp;block&nbsp;in&nbsp;response.content:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;block.type ==&nbsp;"tool_use":

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; handler = TOOL_HANDLERS.get(block.name)

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; output = handler(**block.input)&nbsp;if&nbsp;handler&nbsp;else&nbsp;f"Unknown tool:&nbsp;{block.name}"

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print(f">&nbsp;{block.name}:&nbsp;{output[:200]}")

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; results.append({"type":&nbsp;"tool_result",&nbsp;"tool_use_id": block.id,&nbsp;"content": output})

&nbsp; &nbsp; &nbsp; &nbsp; messages.append({"role":&nbsp;"user",&nbsp;"content": results})

if&nbsp;__name__ ==&nbsp;"__main__":

&nbsp; &nbsp; history = []

&nbsp; &nbsp;&nbsp;while&nbsp;True:

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;try:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; query = input("\033[36ms02 >> \033[0m")

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;except&nbsp;(EOFError, KeyboardInterrupt):

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;break

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;query.strip().lower()&nbsp;in&nbsp;("q",&nbsp;"exit",&nbsp;""):

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;break

&nbsp; &nbsp; &nbsp; &nbsp; history.append({"role":&nbsp;"user",&nbsp;"content": query})

&nbsp; &nbsp; &nbsp; &nbsp; agent_loop(history)

&nbsp; &nbsp; &nbsp; &nbsp; response_content = history[-1]["content"]

&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;isinstance(response_content, list):

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;for&nbsp;block&nbsp;in&nbsp;response_content:

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;if&nbsp;hasattr(block,&nbsp;"text"):

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print(block.text)

&nbsp; &nbsp; &nbsp; &nbsp; print()

效果演示

cd&nbsp;learn-claude-code
python3 agents/s02_tool_use.py

尝试提示词:

Read the file requirements.txt
Create a file called greet.py with a greet(name)&nbsp;function
Edit greet.py to add a docstring to the&nbsp;function
Read greet.py to verify the edit worked

深入探索

Q:为什么恰好四个工具?

A:四个工具分别是 bash、read_file、write_file 和 edit_file,覆盖了大约 95% 的编程任务。Bash 处理执行和任意命令;read_file 提供带行号的精确文件读取;write_file 创建或覆盖文件;edit_file 做精确的字符串替换。工具越多,模型的认知负担越重——它必须在更多选项中做选择,选错的概率也随之增加。更少的工具也意味着更少的 schema 需要维护、更少的边界情况需要处理。

总结

本节中,我们:

  • 引入了三个新的工具并为其定义handler函数
  • 新增了dispatch map用于工具名到函数的映射

新增工具 = 定义handler + 新增scheme,scheme是工具的使用说明书,规定了调用工具的参数以避免歧义。


免责声明:

本文所载程序、技术方法仅面向合法合规的安全研究与教学场景,旨在提升网络安全防护能力,具有明确的技术研究属性。

任何单位或个人未经授权,将本文内容用于攻击、破坏等非法用途的,由此引发的全部法律责任、民事赔偿及连带责任,均由行为人独立承担,本站不承担任何连带责任。

本站内容均为技术交流与知识分享目的发布,若存在版权侵权或其他异议,请通过邮件联系处理,具体联系方式可点击页面上方的联系我

本文转载自:Real返璞归真 Real返璞归真 Real返璞归真《Agent开发|从0实现Agent(一):50行代码实现Mini Claude Code(工具与执行篇)》

评论:0   参与:  0