Claude API – tool use loop bằng Python SDK

uv add anthropic
export ANTHROPIC_API_KEY='***'

import ipaddress
import json
import subprocess
 
import anthropic
 
client = anthropic.Anthropic()  # đọc ANTHROPIC_API_KEY từ env
MODEL = "claude-sonnet-5-5"
MAX_TURNS = 10
 
TOOLS = [
    {
        "name": "ping",
        "description": "Ping một địa chỉ IP từ máy quản trị. Dùng để kiểm tra reachability. Chỉ nhận IP, không nhận hostname.",
        "input_schema": {
            "type": "object",
            "properties": {
                "target": {"type": "string", "description": "IPv4/IPv6, vd 198.51.100.11"},
                "count": {"type": "integer", "minimum": 1, "maximum": 10, "default": 3},
            },
            "required": ["target"],
        },
    },
    {
        "name": "get_device",
        "description": "Tra inventory theo tên thiết bị, trả về IP quản trị, role, OS.",
        "input_schema": {
            "type": "object",
            "properties": {"name": {"type": "string", "description": "vd pe1-hni"}},
            "required": ["name"],
        },
    },
]
 
INVENTORY = {"pe1-hni": {"mgmt": "198.51.100.11", "role": "pe", "os": "iosxr"}}
 
 
def ping(target: str, count: int = 3) -> str:
    ip = ipaddress.ip_address(target)
    out = subprocess.run(["ping", "-c", str(count), "-W", "1", str(ip)],
                         capture_output=True, text=True, timeout=30)
    return out.stdout[-1500:] or out.stderr
 
 
def get_device(name: str) -> str:
    return json.dumps(INVENTORY[name])
 
 
HANDLERS = {"ping": ping, "get_device": get_device}
 
 
def run(question: str) -> str:
    messages = [{"role": "user", "content": question}]
    for _ in range(MAX_TURNS):
        resp = client.messages.create(
            model=MODEL,
            max_tokens=2048,
            system="Bạn là trợ lý NOC. Chỉ dùng tool để đọc trạng thái, không đoán.",
            tools=TOOLS,
            messages=messages,
        )
        # luôn giữ nguyên content của assistant (gồm cả block tool_use)
        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 block in resp.content:
            if block.type != "tool_use":
                continue
            try:
                output = HANDLERS[block.name](**block.input)
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})
            except Exception as e:  # trả lỗi cho model tự sửa, không crash loop
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": f"{type(e).__name__}: {e}", "is_error": True})
        # mọi tool_result của một lượt nằm chung trong MỘT message user
        messages.append({"role": "user", "content": results})
    raise RuntimeError("Vượt MAX_TURNS, có thể agent đang lặp")
 
 
if __name__ == "__main__":
    print(run("pe1-hni có ping được không?"))