【Agent工坊】从零构建AI Agent工具箱：7个核心工具让你的Agent真正能干活的完整指南

def run_bash(command: str) -> str:

    """Run a bash command and return its output."""

    result = subprocess.run(

        command, shell=True, text=True, capture_output=True

    )

    output = result.stdout

    if result.stderr:

        output += f"\nSTDERR:\n{result.stderr}"

    return output or "(no output)"

def read_file(path: str, offset: int = 1, limit: int = 200) -> str:

    """Read lines from a file, with optional offset and limit."""

    p = Path(path)

    if not p.exists():

        return f"Error: file not found: {path}"

    lines = p.read_text(errors="replace").splitlines()

    selected = lines[offset - 1 : offset - 1 + limit]

    return "\n".join(

        f"{offset + i}: {line}"

        for i, line in enumerate(selected)

    )

def glob_files(pattern: str, path: str = ".") -> str:

    """Find files matching a glob pattern."""

    matches = glob_module.glob(f"{path}/**/{pattern}", recursive=True)

    matches += glob_module.glob(f"{path}/{pattern}")

    unique = sorted(set(matches))

    return "\n".join(unique) if unique else "(no matches)"

def grep(pattern: str, path: str = ".", include: str = "*") -> str:

    """Search file contents for a regex pattern."""

    results = []

    for filepath in glob_module.glob(

        f"{path}/**/{include}", recursive=True

    ):

        fp = Path(filepath)

        if not fp.is_file():

            continue

        try:

            for i, line in enumerate(

                fp.read_text(errors="replace").splitlines(), 1

            ):

                if re.search(pattern, line):

                    results.append(f"{filepath}:{i}: {line}")

        except OSError:

            pass

    return "\n".join(results) if results else "(no matches)"

def write_file(path: str, content: str) -> str:

    """Write content to a file, creating parent dirs if needed."""

    p = Path(path)

    p.parent.mkdir(parents=True, exist_ok=True)

    p.write_text(content)

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

def edit_file(path: str, old_string: str, new_string: str) -> str:

    """Replace first occurrence of old_string with new_string."""

    p = Path(path)

    if not p.exists():

        return f"Error: file not found: {path}"

    original = p.read_text()

    if old_string not in original:

        return f"Error: string not found in {path}"

    p.write_text(original.replace(old_string, new_string, 1))

    return f"Edited {path}"

def webfetch(url: str) -> str:

    """Fetch a URL and return plain-text content (max 2MB)."""

    parsed = urlparse(url)

    if parsed.scheme not in ("http", "https"):

        return f"Error: unsupported scheme '{parsed.scheme}'."

    req = urllib.request.Request(

        url, headers={"User-Agent": "agent/1.0"}

    )

    with urllib.request.urlopen(req, timeout=15) as resp:

        raw = b"".join(...).decode(charset, errors="replace")

        soup = BeautifulSoup(raw, "html.parser")

        text = soup.get_text(separator="\n", strip=True)

        return re.sub(r"\n{3,}", "\n\n", text).strip()

def get_tool_schemas():

    return [

        {

            "type": "function",

            "function": {

                "name": "read_file",

                "description": "Read lines from a file, returns lines with line numbers.",

                "parameters": {

                    "type": "object",

                    "properties": {

                        "path": {

                            "type": "string",

                            "description": "Absolute or relative path to the file."

                        },

                        "offset": {

                            "type": "integer",

                            "description": "First line to read (1-indexed). Default 1."

                        },

                        "limit": {

                            "type": "integer",

                            "description": "Max lines to return. Default 200."

                        }

                    },

                    "required": ["path"]

                }

            }

        },

        # ... 其余6个工具的Schema定义

    ]

TOOL_REGISTRY = get_tool_registry() # {"run_bash": run_bash, ...}

TOOL_SCHEMAS = get_tool_schemas()

def handle_tool_calls(tool_calls, messages):

    """Execute each tool and append results to messages."""

    for tool_call in tool_calls:

        name = tool_call.function.name

        args = json.loads(tool_call.function.arguments)

        if name not in TOOL_REGISTRY:

            result = f"Error: unknown tool '{name}'."

        else:

            result = TOOL_REGISTRY[name](**args)

        messages.append({

            "role": "tool",

            "tool_call_id": tool_call.id,

            "content": result

        })

def agent_loop(client):

    messages = [{"role": "system", "content": SYSTEM_PROMPT}]

    while True:

        user_input = input("You: ")

        if user_input.lower() == "\\exit":

            break

        messages.append({"role": "user", "content": user_input})

        while True:

            response = client.chat.completions.create(

                model="gemma4",

                messages=messages,

                tools=TOOL_SCHEMAS,

                temperature=0.7

            )

            message = response.choices[0].message

            messages.append(message)

            if message.tool_calls:

                handle_tool_calls(message.tool_calls, messages)

            else:

                print(f"Assistant: {message.content}")

                break

$ python agent.py

You: 帮我读取 ruxu.dev 首页，列出所有文章并保存到 ruxu.md

[tool] webfetch({'url': 'example.com'})

[tool result] Blog | Roger Oriol ...

[tool] write_file({

    'path': 'ruxu.md',

    'content': '# Articles on ruxu.dev\n\n- Build a Basic AI Agent...'

})

[tool result] Wrote 375 bytes to ruxu.md

Assistant: 完成！ruxu.md 已创建，包含了首页的所有文章列表。

MAX_RESULT = 8000

if len(result) > MAX_RESULT:

    result = result[:MAX_RESULT] + f"\n... (truncated {len(result)} chars total)"

MAX_TOOL_ROUNDS = 10

tool_rounds = 0

while message.tool_calls and tool_rounds < MAX_TOOL_ROUNDS:

    handle_tool_calls(message.tool_calls, messages)

    tool_rounds += 1

    # ... 再次调用LLM

if tool_rounds >= MAX_TOOL_ROUNDS:

    messages.append({"role": "system", "content": "请基于已有信息给出回答，不要再调用工具。"})