实战工具：数据库与文件

上一章的 HTTP 工具解决"查外面"，这一章做本地持久化类工具：sqlite3（Python 标准库自带）存结构化数据、文件系统存笔记。重点同时解决 Agent 工具最常见的两个安全问题：SQL 注入与路径穿越。所有工具仍用第 5 章的 @tool 装饰器登记，随后演示查询-回答轨迹。

sqlite3：待办工具的建表与增删查

sqlite3 无需安装。两个习惯必须养成：参数一律用 ? 占位符传值（防 SQL 注入，绝不 f-string 拼 SQL）；连接用完即关，写成短连接助手，后面第 15 章多线程执行时也不会踩"跨线程共用连接"的坑：

# agent_demo/db_tools.py —— 第 10 章：待办工具
import os
import sqlite3
import time

from tools import tool        # 沿用第 5 章：@tool 装饰器 + call_tool 分发

DB_FILE = os.path.join(os.path.dirname(__file__), "todo.db")

def _exec(sql, args=()):
    """打开短连接执行并提交，返回 (影响行数, 自增 id)。"""
    conn = sqlite3.connect(DB_FILE)
    try:
        cur = conn.execute(sql, args)
        conn.commit()
        return cur.rowcount, cur.lastrowid
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

def init_db():
    _exec("""CREATE TABLE IF NOT EXISTS todo(
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        task TEXT NOT NULL,
        done INTEGER NOT NULL DEFAULT 0,
        created_at TEXT NOT NULL)""")

@tool(description="新增一条待办事项", params={"task": "待办内容，如 买牛奶"})
def todo_add(task: str) -> str:
    """新增一条待办，返回给模型看的一句话。"""
    task = task.strip()
    if not task:
        return "待办内容不能为空"
    _, rowid = _exec("INSERT INTO todo(task, created_at) VALUES(?, ?)",
                     (task, time.strftime("%Y-%m-%d %H:%M")))
    return f"已添加待办 #{rowid}：{task}"

@tool(description="查看待办列表，status 取 all / open / done",
      params={"status": "过滤状态：all 全部 / open 未完成 / done 已完成"})
def todo_list(status: str = "all") -> str:
    """列出待办。"""
    if status not in ("all", "open", "done"):
        return "status 只支持 all / open / done"
    conn = sqlite3.connect(DB_FILE)
    try:
        sql, args = "SELECT id, task, done FROM todo", ()
        if status != "all":
            sql += " WHERE done = ?"
            args = (1 if status == "done" else 0,)
        rows = conn.execute(sql + " ORDER BY id", args).fetchall()
    finally:
        conn.close()
    if not rows:
        return "待办列表是空的"
    return "\n".join(f"{'[x]' if d else '[ ]'} #{i} {t}" for i, t, d in rows)

@tool(description="把某条待办标记为已完成",
      params={"todo_id": "待办编号（正整数）"})
def todo_done(todo_id: int) -> str:
    rows, _ = _exec("UPDATE todo SET done = 1 WHERE id = ?", (int(todo_id),))
    return f"待办 #{todo_id} 已标记完成" if rows else f"找不到待办 #{todo_id}"

文件工具：路径校验 + 大小限制

让模型读写文件等于把文件系统交出去，两条防线缺一不可：路径穿越校验（只允许读写固定沙箱目录内的纯文件名）+ 大小限制（防止把超大文件读进上下文）：

# agent_demo/file_tools.py —— 第 10 章：笔记读写
import os

from tools import tool

NOTES_DIR = os.path.join(os.path.dirname(__file__), "notes")   # 沙箱根目录
MAX_NOTE = 64 * 1024                                            # 单文件上限 64KB
os.makedirs(NOTES_DIR, exist_ok=True)

def _safe_path(name: str):
    """返回安全的绝对路径；含路径分隔符/.. 等一律拒绝。"""
    if not name or name != os.path.basename(name):     # 拦下 a/b、..、空名
        return None
    real = os.path.realpath(os.path.join(NOTES_DIR, name))
    root = os.path.realpath(NOTES_DIR) + os.sep
    return real if real.startswith(root) else None     # realpath 后再兜底一次

@tool(description="写入或覆盖一篇笔记",
      params={"name": "笔记文件名（纯文件名，如 提醒.txt）",
              "content": "笔记正文"})
def note_write(name: str, content: str) -> str:
    """写入/覆盖一篇笔记（校验路径 + 限制大小）。"""
    path = _safe_path(name)
    if path is None:
        return "非法文件名：只允许纯文件名，且不能含 .. 或路径分隔符"
    if len(content.encode("utf-8")) > MAX_NOTE:
        return f"内容超过 {MAX_NOTE // 1024}KB 上限，已拒绝写入"
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"笔记 {name} 已保存（{len(content)} 字符）"

@tool(description="读取一篇笔记",
      params={"name": "笔记文件名", "max_len": "最多读取的字符数，默认 2000"})
def note_read(name: str, max_len: int = 2000) -> str:
    """读取笔记（超限内容截断，避免撑爆上下文）。"""
    path = _safe_path(name)
    if path is None or not os.path.exists(path):
        return f"笔记 {name} 不存在或文件名非法"
    if os.path.getsize(path) > MAX_NOTE:
        return f"笔记 {name} 超过 {MAX_NOTE // 1024}KB，拒绝读取"
    with open(path, encoding="utf-8") as f:
        return f.read(max_len)

手测：无需模型，结果确定

装饰器在 import 时完成注册，函数本身不受影响：既能直接调用，也能被 call_tool 按名字分发。把两个工具模块 import 进来后写个手测脚本：

# demo_tools.py —— 手测 5 个工具（不联网，输出确定）
import db_tools          # import 即完成 @tool 注册
import file_tools
from tools import call_tool

if __name__ == "__main__":
    db_tools.init_db()
    print(call_tool("todo_add", {"task": "买牛奶"}))
    # 输出：已添加待办 #1：买牛奶
    print(call_tool("todo_add", {"task": "周末交房租"}))
    # 输出：已添加待办 #2：周末交房租
    print(call_tool("todo_done", {"todo_id": 1}))
    # 输出：待办 #1 已标记完成
    print(call_tool("todo_list", {"status": "open"}))
    # 输出：[ ] #2 周末交房租
    print(call_tool("note_write",
                    {"name": "提醒.txt", "content": "周日记得还书"}))
    # 输出：笔记 提醒.txt 已保存（6 字符）
    print(call_tool("note_read", {"name": "../secret.txt"}))   # 穿越攻击被拦
    # 输出：笔记 ../secret.txt 不存在或文件名非法

python demo_tools.py

两条安全防线都验证到了：../secret.txt 因含路径分隔符被 _safe_path 拦下；SQL 全程 ? 占位符，模型输出再"野"也注入不进去。重复运行本脚本时自增 id 会继续累加，输出编号以实际为准。

Agent 查询-回答轨迹

让模型自主完成"记待办→查待办→写文件"整串动作：先 import 两个工具模块完成注册（装饰器在 import 时执行），再用第 8 章 agent.py 的 run_agent 跑——它自动携带 TOOL_SCHEMAS、用 call_tool 分发并打印每一步，还带步骤上限等保险：

# demo_agent.py —— 需先配好 .env（LLM_API_KEY/LLM_BASE_URL，见第 2 章）
import db_tools          # 仅 import 即完成 @tool 注册
import file_tools
from agent import run_agent            # 沿用第 8 章 agent.py 的运行器
from tools import TOOL_SCHEMAS

messages = [{"role": "user",
             "content": "帮我记下：周六交房租。然后列出我没完成的事，"
                        "写进笔记 待办汇报.txt"}]
messages = run_agent("你是待办助手：需要工具就调用，最后用简体中文如实汇报。",
                     TOOL_SCHEMAS, messages)
print("\n最终回答：", messages[-1]["content"])

一次运行的对话轨迹形如（内容随模型回复变化）：

# 输出（示例轨迹，实际以模型回复为准）：
===== 第 1/10 步 =====
模型：（发起工具调用）
  <- todo_add 返回：已添加待办 #3：周六交房租
  <- todo_list 返回：[ ] #2 周末交房租
                    [ ] #3 周六交房租
  <- note_write 返回：笔记 待办汇报.txt 已保存（20 字符）
===== 第 2/10 步 =====
模型：已记下"周六交房租"；未完成事项已写入笔记 待办汇报.txt。

要点回顾：工具返回值是给模型看的一句人话（如"已添加待办 #3：周六交房租"），模型照抄即可组织自然回答；路径校验与大小限制让"模型控制文件读写"变得安全可控。

小结：数据库与文件是智能体"记住并落地"的左右手——sqlite3 短连接 + ? 占位符防注入，文件读写用沙箱目录 + realpath 校验防穿越并限制大小；用第 5 章 @tool 装饰器登记后，交给第 8 章 run_agent 即可让模型自主完成"记录-查询-落盘"闭环。