#s02：工具使用

{ "name": "read_file", "input": { "path": "main.py" } }

dispatch = {
    "bash":       run_bash,
    "read_file":  run_read,
    "write_file": run_write,
    "edit_file":  run_edit,
}
handler = dispatch[tool_name]   # 查表
result  = handler(tool_input)   # 执行

TOOLS = [{"name": "bash", ...}]
 
def run_bash(command): ...

TOOLS = [
    {"name": "bash",       "description": "Run a shell command.", ...},
    {"name": "read_file",  "description": "Read file contents.",  ...},
    {"name": "write_file", "description": "Write content to file.", ...},
    {"name": "edit_file",  "description": "Replace text in file once.", ...},
    {"name": "glob",       "description": "Find files by pattern.", ...},
]

def run_read(path, limit=None):
    lines = safe_path(path).read_text().splitlines()
    if limit:
        lines = lines[:limit]          # LLM 可以指定只看前 N 行，避免读超大文件
    return "\n".join(lines)
 
def run_write(path, content):
    safe_path(path).write_text(content)
    return f"Wrote {len(content)} bytes to {path}"
 
def run_edit(path, old_text, new_text):
    text = safe_path(path).read_text()
    if old_text not in text:
        return "Error: text not found"  # 找不到就报错，而不是静默失败
    safe_path(path).write_text(text.replace(old_text, new_text, 1))  # 只替换第一处
    return f"Edited {path}"
 
def run_glob(pattern):
    import glob as g
    # root_dir 把搜索范围锁在工作目录，LLM 看不到工作目录之外的文件
    return "\n".join(g.glob(pattern, root_dir=WORKDIR))

# dispatch map：工具名 → 处理函数
# 加新工具只需在这里加一行，下面的循环不用动
TOOL_HANDLERS = {
    "bash":       run_bash,
    "read_file":  run_read,
    "write_file": run_write,
    "edit_file":  run_edit,
    "glob":       run_glob,
}
 
# 循环里只改了一行——从硬编码 run_bash 变成查表：
for block in response.content:
    if block.type == "tool_use":
        handler = TOOL_HANDLERS[block.name]    # 按名字找到对应函数
        output = handler(**block.input)        # block.input 是 LLM 传来的参数字典
        results.append(...)

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()
    if limit and limit < len(lines):
        lines = lines[:limit]
    return "\n".join(lines)[:50000]

# 用 lambda 把 block.input 字典拆包成各函数期望的具名参数
# kw.get("limit") 而非 kw["limit"]：limit 是可选参数，缺省时返回 None
TOOL_HANDLERS = {
    "bash":       lambda **kw: run_bash(kw["command"]),
    "read_file":  lambda **kw: run_read(kw["path"], kw.get("limit")),
    "write_file": lambda **kw: run_write(kw["path"], kw["content"]),
    "edit_file":  lambda **kw: run_edit(kw["path"], kw["old_text"],
                                        kw["new_text"]),
}

for block in response.content:
    if block.type == "tool_use":
        handler = TOOL_HANDLERS.get(block.name)   # 未知工具返回 None，不抛异常
        output = handler(**block.input) if handler \
            else f"Unknown tool: {block.name}"     # 降级为报错字符串，让 LLM 感知
        results.append({
            "type": "tool_result",
            "tool_use_id": block.id,   # 必须与请求的 block.id 对应，API 用它匹配结果
            "content": output,
        })

{
  "name": "read_file",
  "description": "读取指定路径的文件内容",
  "input_schema": {
    "type": "object",
    "properties": {
      "path": { "type": "string", "description": "文件路径" },
      "limit": { "type": "integer", "description": "最多读多少行" }
    },
    "required": ["path"]
  }
}

def normalize_messages(messages: list) -> list:
    """发送给 API 前调用，确保 messages 符合 Anthropic 协议格式。"""
    normalized = []
 
    # ── Pass 1：剥离内部字段 ──────────────────────────────────────────────────
    # agent 内部会给消息附加调试字段（_internal / _source / _timestamp 等）。
    # API 只认协议定义的字段，多余字段会导致 400 报错，必须在发送前过滤掉。
    for msg in messages:
        clean = {"role": msg["role"]}
        if isinstance(msg.get("content"), str):
            clean["content"] = msg["content"]
        elif isinstance(msg.get("content"), list):
            clean["content"] = [
                {k: v for k, v in block.items()
                 if k not in ("_internal", "_source", "_timestamp")}
                for block in msg["content"]
            ]
        normalized.append(clean)
 
    # ── Pass 2：补齐孤立的 tool_use ──────────────────────────────────────────
    # API 要求每个 tool_use 必须有配对的 tool_result（通过 tool_use_id 关联）。
    # 工具超时、用户取消、压缩截断都可能导致 tool_use 没有配对。
    # 先收集所有已有的 tool_use_id → set，再扫 assistant 消息找孤立的 tool_use，
    # 在末尾追加一条 "(cancelled)" 占位 result，让 API 能看到合法的配对。
    existing_results = set()
    for msg in normalized:
        if isinstance(msg.get("content"), list):
            for block in msg["content"]:
                if block.get("type") == "tool_result":
                    existing_results.add(block.get("tool_use_id"))
 
    for msg in normalized:
        if msg["role"] == "assistant" and isinstance(msg.get("content"), list):
            for block in msg["content"]:
                if (block.get("type") == "tool_use"
                        and block.get("id") not in existing_results):
                    normalized.append({"role": "user", "content": [{
                        "type": "tool_result",
                        "tool_use_id": block["id"],
                        "content": "(cancelled)",   # 统一占位符，不区分超时/取消原因
                    }]})
 
    # ── Pass 3：合并连续同角色消息 ────────────────────────────────────────────
    # API 要求 user / assistant 严格交替，不能连续两条同角色。
    # Pass 2 追加的 tool_result 放在 user 消息里，可能紧跟在另一条 user 消息后面，
    # 同样会触发这里的合并。合并方式：把后一条的 content 拼到前一条的 content 列表里。
    merged = [normalized[0]] if normalized else []
    for msg in normalized[1:]:
        if msg["role"] == merged[-1]["role"]:
            prev = merged[-1]
            prev_content = prev["content"] if isinstance(prev["content"], list) \
                else [{"type": "text", "text": prev["content"]}]
            curr_content = msg["content"] if isinstance(msg["content"], list) \
                else [{"type": "text", "text": msg["content"]}]
            prev["content"] = prev_content + curr_content
        else:
            merged.append(msg)
 
    return merged

response = client.messages.create(
    model=MODEL, system=system,
    messages=normalize_messages(messages),  # 规范化后再发送
    tools=TOOLS, max_tokens=8000,
)

cd s02-tool-use
python3 -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -r requirements.txt
cp .env.example .env
# 编辑 .env，填入 API_KEY / BASE_URL / MODEL_ID
python code.py