用 Claude API 的工具调用做一个命令行记账助手

用 Claude API 的工具调用做一个命令行记账助手：你说“午饭花了 32 块”，Claude 调用 add_expense 写入本地 SQLite；你问“这个月交通花了多少”，它调用 sum_expenses 查出真实数字再回答，不会自己编。

先看清成品

成品是 ledger_agent.py。本文已用模拟客户端实际运行 3 个测试并全部通过：工具调用结果按正确格式（tool_result + tool_use_id）传回；金额为负、缺少参数时作为错误回传给模型；按日期范围汇总正确。另外用真实 SDK（anthropic 1.17.0）和无效密钥发了一次带工具定义的请求，返回 401 invalid x-api-key，说明请求格式被接受。本文环境没有有效的 API 密钥，没有测试真实对话效果。

注意：Claude 只在 Anthropic 官方“支持的国家和地区”列表中提供，请在支持地区合规使用，不要用代理或他人账号绕过限制，也不要共享账号。遵守 Anthropic 的使用政策。价格和用量以官方价格页为准。

做完之后你会得到什么？

需要准备哪些环境？

第一步：工具调用是怎么工作的？

真正读写数据库的是你的代码，模型只决定“调用哪个工具、传什么参数”。

第二步：代码怎么写？

1. 项目结构

ledger/
├── ledger_agent.py
├── test_ledger.py
└── ledger.db        # 运行后生成

2. ledger_agent.py

"""用 Claude 工具调用做一个记账助手：说人话记账、查账，数据存在本地 SQLite。
用法：python ledger_agent.py      然后输入“午饭花了 32 块”“这个月餐饮花了多少”
"""
import json
import os
import sqlite3
from datetime import date

MODEL = os.environ.get("CLAUDE_MODEL", "claude-sonnet-5-5")
DB = os.environ.get("LEDGER_DB", "ledger.db")
CATEGORIES = ["餐饮", "交通", "购物", "居住", "娱乐", "其他"]

TOOLS = [
    {
        "name": "add_expense",
        "description": "记录一笔支出。用户说花了多少钱、买了什么时调用。",
        "input_schema": {
            "type": "object",
            "properties": {
                "amount": {"type": "number", "description": "金额，单位元，必须大于 0"},
                "category": {"type": "string", "enum": CATEGORIES},
                "note": {"type": "string", "description": "简短备注，例如“午饭”"},
                "day": {"type": "string", "description": "日期 YYYY-MM-DD，用户没说就用今天"},
            },
            "required": ["amount", "category", "note", "day"],
        },
    },
    {
        "name": "sum_expenses",
        "description": "按日期范围和可选分类汇总支出，返回总额和笔数。",
        "input_schema": {
            "type": "object",
            "properties": {
                "start": {"type": "string", "description": "开始日期 YYYY-MM-DD（含）"},
                "end": {"type": "string", "description": "结束日期 YYYY-MM-DD（含）"},
                "category": {"type": "string", "enum": CATEGORIES},
            },
            "required": ["start", "end"],
        },
    },
]


def conn():
    c = sqlite3.connect(DB)
    c.execute("CREATE TABLE IF NOT EXISTS exp (id INTEGER PRIMARY KEY, day TEXT, amount REAL, category TEXT, note TEXT)")
    return c


def add_expense(amount, category, note, day):
    if amount <= 0:
        return {"error": "金额必须大于 0"}
    if category not in CATEGORIES:
        return {"error": f"分类必须是 {CATEGORIES} 之一"}
    with conn() as c:
        cur = c.execute("INSERT INTO exp (day, amount, category, note) VALUES (?, ?, ?, ?)", (day, amount, category, note))
    return {"ok": True, "id": cur.lastrowid}


def sum_expenses(start, end, category=None):
    sql, args = "SELECT COALESCE(SUM(amount), 0), COUNT(*) FROM exp WHERE day BETWEEN ? AND ?", [start, end]
    if category:
        sql += " AND category = ?"
        args.append(category)
    total, n = conn().execute(sql, args).fetchone()
    return {"total": round(total, 2), "count": n}


HANDLERS = {"add_expense": add_expense, "sum_expenses": sum_expenses}


def system_prompt():
    return f"你是记账助手，今天是 {date.today().isoformat()}。用工具记账和查账，不要自己编造数字。回答简短，用中文。"


def chat(client, messages, max_rounds=5):
    """执行一轮对话：模型要求调用工具就执行，直到它给出最终回答。"""
    for _ in range(max_rounds):
        resp = client.messages.create(model=MODEL, max_tokens=1024, system=system_prompt(), tools=TOOLS, messages=messages)
        messages.append({"role": "assistant", "content": resp.content})
        if resp.stop_reason != "tool_use":
            return "".join(b.text for b in resp.content if b.type == "text")
        results = []
        for b in resp.content:
            if b.type == "tool_use":
                try:
                    out = HANDLERS[b.name](**b.input)
                except Exception as e:  # 参数不对时把错误告诉模型，让它修正
                    out = {"error": str(e)}
                print(f"  [工具] {b.name}({json.dumps(b.input, ensure_ascii=False)}) -> {json.dumps(out, ensure_ascii=False)}")
                results.append({"type": "tool_result", "tool_use_id": b.id, "content": json.dumps(out, ensure_ascii=False),
                                "is_error": "error" in out})
        messages.append({"role": "user", "content": results})
    return "（工具调用次数过多，已停止）"


def main():
    import anthropic
    client = anthropic.Anthropic()  # 读取环境变量 ANTHROPIC_API_KEY
    messages = []
    print("记账助手已启动，输入 q 退出")
    while (q := input("你：").strip()) not in ("q", "quit"):
        if q:
            messages.append({"role": "user", "content": q})
            print("助手：" + chat(client, messages))


if __name__ == "__main__":
    main()

第三步：怎样测试和运行？

1. 离线测试

from types import SimpleNamespace as NS

import pytest

import ledger_agent as la


@pytest.fixture(autouse=True)
def tmpdb(tmp_path, monkeypatch):
    monkeypatch.setattr(la, "DB", str(tmp_path / "l.db"))


def text(t): return NS(type="text", text=t)
def tool(i, name, args): return NS(type="tool_use", id=i, name=name, input=args)


class FakeClient:
    """按顺序返回预设的模型回复，并记录每次请求"""
    def __init__(self, replies):
        self.replies, self.calls = list(replies), []
        self.messages = NS(create=self._create)

    def _create(self, **kw):
        self.calls.append(kw)
        return self.replies.pop(0)


def test_record_then_answer():
    c = FakeClient([
        NS(stop_reason="tool_use", content=[text("好的"), tool("t1", "add_expense", {"amount": 32, "category": "餐饮", "note": "午饭", "day": "2026-10-21"})]),
        NS(stop_reason="end_turn", content=[text("已记录午饭 32 元。")]),
    ])
    msgs = [{"role": "user", "content": "午饭 32"}]
    assert la.chat(c, msgs) == "已记录午饭 32 元。"
    assert c.calls[0]["tools"][0]["name"] == "add_expense"
    tr = msgs[2]["content"][0]
    assert tr["type"] == "tool_result" and tr["tool_use_id"] == "t1" and tr["is_error"] is False
    assert la.sum_expenses("2026-10-01", "2026-10-31", "餐饮") == {"total": 32, "count": 1}


def test_bad_args_reported_as_error():
    c = FakeClient([
        NS(stop_reason="tool_use", content=[tool("t1", "add_expense", {"amount": -5, "category": "餐饮", "note": "x", "day": "2026-10-21"})]),
        NS(stop_reason="tool_use", content=[tool("t2", "sum_expenses", {"start": "2026-10-01"})]),
        NS(stop_reason="end_turn", content=[text("金额有误")]),
    ])
    msgs = [{"role": "user", "content": "x"}]
    la.chat(c, msgs)
    assert msgs[2]["content"][0]["is_error"] is True
    assert "end" in msgs[4]["content"][0]["content"]  # 缺参数的 TypeError 被传回给模型


def test_sum_range():
    la.add_expense(10, "交通", "地铁", "2026-09-30")
    la.add_expense(20, "交通", "打车", "2026-10-02")
    assert la.sum_expenses("2026-10-01", "2026-10-31") == {"total": 20, "count": 1}

python -m pytest -q

本文实际输出 3 passed。

2. 真实运行

export ANTHROPIC_API_KEY="你的密钥"
python ledger_agent.py

期望的对话（具体措辞会不同）：

你：午饭花了 32，打车 18
  [工具] add_expense({"amount": 32, "category": "餐饮", "note": "午饭", "day": "2026-10-21"}) -> {"ok": true, "id": 1}
  [工具] add_expense({"amount": 18, "category": "交通", "note": "打车", "day": "2026-10-21"}) -> {"ok": true, "id": 2}
助手：已记录午饭 32 元、打车 18 元。
你：这个月一共花了多少？
  [工具] sum_expenses({"start": "2026-10-01", "end": "2026-10-31"}) -> {"total": 50, "count": 2}
助手：本月共 2 笔，合计 50 元。

3. 用数据库核对

sqlite3 ledger.db "SELECT * FROM exp"

助手说的数字应该和数据库一致。

出问题时该怎么调？

1. 模型没调用工具，直接回答

工具描述写清“什么时候调用”；系统提示里要求“不要自己编造数字”。

2. 报 400，提示 tool_use_id 不匹配

每个 tool_use 都必须有对应的 tool_result，并且放在紧接着的 user 消息里。

3. 日期算错

检查系统提示里的日期；时区不同时，date.today() 用的是本机时区。

常见问题怎么处理？

项目资源

建议先跑离线测试理解循环，再用真实密钥记三五笔账，用 sqlite3 核对每个数字。

确认可靠后再增加工具，例如按分类列出明细、设置月度预算提醒。