📌 内容摘要

  • Tool Use 让 Claude 从”只会说”变成”能做事”——通过调用外部工具完成搜索、计算、数据库操作等真实任务。
  • 本文从单工具调用到并行多工具、再到复杂工具链,三个层次递进,每层都有完整可运行代码。
  • 重点解决实际开发中最常遇到的问题:工具定义规范、并行调用优化、工具调用失败处理、防止无限循环。
  • 文末提供一个完整的”个人助理 Agent”示例,集成搜索、日历、邮件、计算四类工具,展示生产级设计。

一、Tool Use 的工作机制

理解 Tool Use 的关键是搞清楚它的消息流——Claude 自己不执行工具,它只是”请求”你的代码去执行,然后把执行结果反馈给它继续生成。

Tool Use 消息流:

你 → Claude:用户消息 + 工具定义列表
Claude → 你:stop_reason="tool_use",包含工具名和参数
你:执行工具,得到结果
你 → Claude:把工具结果作为 user 消息发回
Claude → 你:stop_reason="end_turn",生成最终回答

关键点:工具实际上是你的代码执行的,Claude 只负责决定调什么、传什么参数。

二、基础:单工具调用

import anthropic
import json

client = anthropic.Anthropic()

# ── Step 1:定义工具 ──────────────────────────────

def get_stock_price(symbol: str) -> dict:
    """模拟获取股票价格(实际项目替换为真实 API)"""
    mock_prices = {
        "AAPL": {"price": 189.5, "change": +1.2, "volume": 52_000_000},
        "GOOGL":{"price": 175.3, "change": -0.8, "volume": 21_000_000},
        "TSLA": {"price": 248.9, "change": +3.5, "volume": 98_000_000},
    }
    if symbol.upper() in mock_prices:
        return mock_prices[symbol.upper()]
    return {"error": f"未找到股票代码:{symbol}"}


STOCK_TOOL = {
    "name": "get_stock_price",
    "description": "获取指定股票代码的当前价格、涨跌幅和成交量",
    "input_schema": {
        "type": "object",
        "properties": {
            "symbol": {
                "type": "string",
                "description": "股票代码,如 'AAPL'(苹果)、'GOOGL'(谷歌)"
            }
        },
        "required": ["symbol"]
    }
}


# ── Step 2:单轮工具调用 ──────────────────────────

def run_with_tool(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    # 第一次调用:Claude 决定是否使用工具
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=[STOCK_TOOL],
        messages=messages,
    )

    # 如果不需要工具,直接返回
    if response.stop_reason == "end_turn":
        return response.content[0].text

    # 处理工具调用
    tool_results = []
    for block in response.content:
        if block.type == "tool_use":
            print(f"  → 调用工具:{block.name}({json.dumps(block.input)})")

            # 执行工具
            result = get_stock_price(**block.input)

            tool_results.append({
                "type":        "tool_result",
                "tool_use_id": block.id,
                "content":     json.dumps(result, ensure_ascii=False),
            })

    # 把工具结果发回给 Claude
    messages.append({"role": "assistant", "content": response.content})
    messages.append({"role": "user",      "content": tool_results})

    # 第二次调用:Claude 生成最终答案
    final_response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=[STOCK_TOOL],
        messages=messages,
    )
    return final_response.content[0].text


# 测试
print(run_with_tool("苹果公司的股票现在多少钱?"))
print(run_with_tool("今天天气怎么样?"))  # 不需要工具,Claude 直接回答

三、进阶:多工具定义与选择

import datetime
import math

# ── 工具函数库 ────────────────────────────────────

def search_web(query: str, max_results: int = 3) -> list[dict]:
    """搜索网页(实际项目用 Serper/Brave API)"""
    return [
        {"title": f"搜索结果:{query}", "snippet": "相关内容摘要...", "url": "https://example.com"},
    ]

def get_current_time(timezone: str = "Asia/Shanghai") -> str:
    """获取当前时间"""
    now = datetime.datetime.now()
    return now.strftime(f"%Y-%m-%d %H:%M:%S ({timezone})")

def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        safe_env = {"__builtins__": {}, "math": math,
                    "sqrt": math.sqrt, "pi": math.pi, "e": math.e}
        result = eval(expression, safe_env)
        return str(round(result, 6))
    except Exception as err:
        return f"计算错误:{err}"

def get_exchange_rate(from_currency: str, to_currency: str) -> dict:
    """获取汇率(模拟)"""
    rates = {"USD_CNY": 7.24, "EUR_CNY": 7.85, "USD_EUR": 0.92}
    key = f"{from_currency.upper()}_{to_currency.upper()}"
    rate = rates.get(key) or (1 / rates.get(f"{to_currency.upper()}_{from_currency.upper()}", 0) or None)
    if rate:
        return {"from": from_currency, "to": to_currency, "rate": rate}
    return {"error": f"不支持的货币对:{from_currency}/{to_currency}"}

def query_database(table: str, condition: str = "", limit: int = 10) -> list[dict]:
    """查询数据库(模拟)"""
    mock_data = {
        "users": [
            {"id": 1, "name": "张三", "email": "zhang@example.com", "vip": True},
            {"id": 2, "name": "李四", "email": "li@example.com",    "vip": False},
        ],
        "orders": [
            {"id": 101, "user_id": 1, "amount": 299, "status": "completed"},
            {"id": 102, "user_id": 2, "amount": 599, "status": "pending"},
        ]
    }
    data = mock_data.get(table, [])
    return data[:limit]


# ── 工具注册表(统一管理)────────────────────────

TOOL_REGISTRY = {
    "search_web":      search_web,
    "get_current_time":get_current_time,
    "calculate":       calculate,
    "get_exchange_rate":get_exchange_rate,
    "query_database":  query_database,
    "get_stock_price": get_stock_price,
}

TOOLS = [
    {
        "name": "search_web",
        "description": "搜索互联网获取实时信息,适合查询新闻、事实或不确定的信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "query":       {"type": "string",  "description": "搜索关键词"},
                "max_results": {"type": "integer", "description": "返回结果数量,默认3", "default": 3}
            },
            "required": ["query"]
        }
    },
    {
        "name": "get_current_time",
        "description": "获取当前日期和时间",
        "input_schema": {
            "type": "object",
            "properties": {
                "timezone": {"type": "string", "description": "时区,默认 Asia/Shanghai", "default": "Asia/Shanghai"}
            }
        }
    },
    {
        "name": "calculate",
        "description": "计算数学表达式,支持加减乘除、幂运算、三角函数等",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {"type": "string", "description": "数学表达式,如 '(100 + 200) * 1.13'"}
            },
            "required": ["expression"]
        }
    },
    {
        "name": "get_exchange_rate",
        "description": "获取两种货币之间的实时汇率",
        "input_schema": {
            "type": "object",
            "properties": {
                "from_currency": {"type": "string", "description": "源货币代码,如 'USD'"},
                "to_currency":   {"type": "string", "description": "目标货币代码,如 'CNY'"}
            },
            "required": ["from_currency", "to_currency"]
        }
    },
    {
        "name": "query_database",
        "description": "查询业务数据库,获取用户或订单信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "table":     {"type": "string",  "description": "表名:users 或 orders"},
                "condition": {"type": "string",  "description": "查询条件(可选)"},
                "limit":     {"type": "integer", "description": "返回记录数,默认10"}
            },
            "required": ["table"]
        }
    },
    STOCK_TOOL,
]


def execute_tool(name: str, inputs: dict) -> str:
    """统一工具执行入口,含错误处理"""
    if name not in TOOL_REGISTRY:
        return json.dumps({"error": f"未知工具:{name}"})
    try:
        result = TOOL_REGISTRY[name](**inputs)
        return json.dumps(result, ensure_ascii=False, default=str)
    except TypeError as e:
        return json.dumps({"error": f"参数错误:{e}"})
    except Exception as e:
        return json.dumps({"error": f"执行失败:{e}"})

四、核心:并行工具调用

Claude 可以在一次响应中同时请求多个工具调用——并行执行,大幅节省时间。这是 Tool Use 最重要的性能优化点:

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def process_tool_calls_parallel(tool_use_blocks: list) -> list[dict]:
    """
    并行执行所有工具调用
    同一批次的工具调用互相独立时,并行执行速度快得多

    对比:
    - 串行 3 个工具(各需1秒):约 3 秒
    - 并行 3 个工具(各需1秒):约 1 秒
    """
    results = [None] * len(tool_use_blocks)

    def run_one(idx: int, block):
        start = time.time()
        result = execute_tool(block.name, block.input)
        elapsed = time.time() - start
        print(f"  ✓ {block.name}({elapsed:.2f}s)")
        return idx, block.id, result

    with ThreadPoolExecutor(max_workers=len(tool_use_blocks)) as executor:
        futures = {
            executor.submit(run_one, i, block): i
            for i, block in enumerate(tool_use_blocks)
        }
        for future in as_completed(futures):
            idx, tool_use_id, result = future.result()
            results[idx] = {
                "type":        "tool_result",
                "tool_use_id": tool_use_id,
                "content":     result,
            }

    return results


# ── 完整的多工具 Agent 循环 ───────────────────────

class MultiToolAgent:
    """
    支持并行工具调用的完整 Agent
    能处理多轮工具调用(Claude 调完一批工具,可能再调另一批)
    """

    def __init__(
        self,
        tools:     list[dict],
        model:     str = "claude-sonnet-4-6",
        max_turns: int = 6,
        verbose:   bool = True,
    ):
        self.tools     = tools
        self.model     = model
        self.max_turns = max_turns
        self.verbose   = verbose

    def run(self, user_message: str, system: str = "") -> str:
        messages = [{"role": "user", "content": user_message}]
        turn = 0

        while turn < self.max_turns:
            turn += 1

            # 调用 Claude
            kwargs = dict(
                model=self.model,
                max_tokens=4096,
                tools=self.tools,
                messages=messages,
            )
            if system:
                kwargs["system"] = system

            response = client.messages.create(**kwargs)

            if self.verbose:
                print(f"\n[Turn {turn}] stop_reason={response.stop_reason}")

            # 任务完成
            if response.stop_reason == "end_turn":
                texts = [b.text for b in response.content if hasattr(b, "text")]
                return "\n".join(texts)

            # 没有工具调用(异常情况)
            if response.stop_reason != "tool_use":
                texts = [b.text for b in response.content if hasattr(b, "text")]
                return "\n".join(texts) if texts else "(无输出)"

            # 收集本轮所有工具调用
            tool_blocks = [b for b in response.content if b.type == "tool_use"]

            if self.verbose:
                print(f"  本轮工具调用数:{len(tool_blocks)}")
                for b in tool_blocks:
                    print(f"  → {b.name}({json.dumps(b.input, ensure_ascii=False)[:80]})")

            # 并行执行所有工具
            tool_results = process_tool_calls_parallel(tool_blocks)

            # 更新消息历史
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user",      "content": tool_results})

        return "超过最大对话轮数,任务未完成"


# ── 测试并行调用 ──────────────────────────────────

agent = MultiToolAgent(tools=TOOLS, verbose=True)

# 这个问题会触发 Claude 同时调用多个工具
answer = agent.run(
    "帮我查一下:① 苹果股票当前价格 ② 美元对人民币汇率 ③ 现在北京时间"
)
print(f"\n最终答案:{answer}")

五、工具调用控制策略

tool_choice 参数详解

# tool_choice 三种模式

# 1. auto(默认):Claude 自行决定是否使用工具
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=TOOLS,
    tool_choice={"type": "auto"},    # 默认,Claude 自由决定
    messages=[{"role": "user", "content": "今天天气怎么样?"}]
)

# 2. any:强制 Claude 必须调用至少一个工具
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=TOOLS,
    tool_choice={"type": "any"},     # 强制调用,适合数据采集场景
    messages=[{"role": "user", "content": "分析一下当前市场状况"}]
)

# 3. tool:强制调用指定工具
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=TOOLS,
    tool_choice={
        "type": "tool",
        "name": "calculate"          # 只能调用这个工具
    },
    messages=[{"role": "user", "content": "计算 (15 + 25) * 3.14"}]
)

# 适用场景:
# auto:通用对话 Agent,Claude 自行判断是否需要工具
# any:数据采集管道,确保每次都调用工具获取新数据
# tool:结构化提取,强制用特定工具返回特定格式数据

利用 tool_choice 做结构化信息提取

def extract_structured(text: str, schema: dict) -> dict:
    """
    用 tool_choice="tool" 强制 Claude 返回结构化 JSON
    比让 Claude 自由生成 JSON 更稳定
    """
    extraction_tool = {
        "name": "extract_info",
        "description": "提取文本中的结构化信息",
        "input_schema": schema
    }

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=[extraction_tool],
        tool_choice={"type": "tool", "name": "extract_info"},
        messages=[{
            "role": "user",
            "content": f"从以下文本中提取信息:\n\n{text}"
        }]
    )

    for block in response.content:
        if block.type == "tool_use" and block.name == "extract_info":
            return block.input  # 直接返回结构化数据(已经是 dict)

    return {}


# 使用示例:从简历提取信息
resume_schema = {
    "type": "object",
    "properties": {
        "name":       {"type": "string",       "description": "姓名"},
        "email":      {"type": "string",       "description": "邮箱"},
        "experience": {"type": "integer",      "description": "工作年限"},
        "skills":     {"type": "array",
                       "items": {"type": "string"},
                       "description": "技能列表"},
        "education":  {"type": "string",       "description": "最高学历"}
    },
    "required": ["name", "skills"]
}

resume_text = """
张伟,邮箱 zhangwei@example.com
8年后端开发经验,熟悉 Python/Go/Java
精通 MySQL、Redis、Kafka
本科:北京大学计算机科学 2016年毕业
"""

result = extract_structured(resume_text, resume_schema)
print(json.dumps(result, ensure_ascii=False, indent=2))
# 输出:{"name": "张伟", "email": "zhangwei@example.com",
#         "experience": 8, "skills": ["Python", "Go", "Java", ...], ...}

六、工具链编排:工具调用工具

真实业务中,常见一种模式:一个工具的输出作为另一个工具的输入。Claude 能自动处理这种依赖链:

def create_pipeline_tools():
    """
    创建一组支持工具链的工具:
    1. 搜索股票代码  →  2. 获取股价  →  3. 换算成人民币
    """

    def search_stock_symbol(company_name: str) -> dict:
        """根据公司名称查找股票代码"""
        mapping = {
            "苹果": "AAPL", "谷歌": "GOOGL", "特斯拉": "TSLA",
            "微软": "MSFT", "亚马逊": "AMZN",
        }
        symbol = mapping.get(company_name)
        if symbol:
            return {"symbol": symbol, "company": company_name}
        return {"error": f"未找到公司:{company_name}"}

    def convert_to_rmb(usd_amount: float) -> dict:
        """将美元金额转换为人民币"""
        rate = 7.24
        rmb = usd_amount * rate
        return {"usd": usd_amount, "rmb": round(rmb, 2), "rate": rate}

    pipeline_tools = [
        {
            "name": "search_stock_symbol",
            "description": "根据公司中文名称查找股票代码,在获取股价前先用此工具",
            "input_schema": {
                "type": "object",
                "properties": {
                    "company_name": {"type": "string", "description": "公司中文名称"}
                },
                "required": ["company_name"]
            }
        },
        STOCK_TOOL,
        {
            "name": "convert_to_rmb",
            "description": "将美元金额转换为人民币,用于汇报股价时同时显示人民币价格",
            "input_schema": {
                "type": "object",
                "properties": {
                    "usd_amount": {"type": "number", "description": "美元金额"}
                },
                "required": ["usd_amount"]
            }
        }
    ]

    TOOL_REGISTRY["search_stock_symbol"] = search_stock_symbol
    TOOL_REGISTRY["convert_to_rmb"]      = convert_to_rmb

    return pipeline_tools


# Claude 会自动按依赖顺序调用:search → get_price → convert
pipeline_tools = create_pipeline_tools()
agent = MultiToolAgent(tools=pipeline_tools, verbose=True)

answer = agent.run("苹果公司的股价是多少?同时告诉我对应的人民币价格")
print(f"\n最终答案:{answer}")

七、完整示例:个人助理 Agent

import re

# ── 个人助理工具集 ────────────────────────────────

_calendar: list[dict] = []   # 模拟内存日历
_emails:   list[dict] = []   # 模拟发件箱

def add_calendar_event(title: str, date: str, time: str, duration_minutes: int = 60) -> dict:
    """添加日历事件"""
    event = {"id": len(_calendar) + 1, "title": title,
             "date": date, "time": time, "duration": duration_minutes}
    _calendar.append(event)
    return {"status": "已添加", "event": event}

def list_calendar_events(date: str = "") -> list[dict]:
    """查看日历事件"""
    if date:
        return [e for e in _calendar if e["date"] == date]
    return _calendar

def send_email(to: str, subject: str, body: str) -> dict:
    """发送邮件"""
    email = {"to": to, "subject": subject, "body": body,
             "sent_at": datetime.datetime.now().isoformat()}
    _emails.append(email)
    return {"status": "邮件已发送", "to": to, "subject": subject}

def create_reminder(message: str, remind_at: str) -> dict:
    """创建提醒"""
    return {"status": "提醒已创建", "message": message, "remind_at": remind_at}

def take_note(title: str, content: str, tags: list[str] = None) -> dict:
    """记录笔记"""
    note = {"title": title, "content": content, "tags": tags or [],
            "created_at": datetime.datetime.now().isoformat()}
    return {"status": "笔记已保存", "note": note}


# 注册到工具注册表
TOOL_REGISTRY.update({
    "add_calendar_event": add_calendar_event,
    "list_calendar_events": list_calendar_events,
    "send_email":  send_email,
    "create_reminder": create_reminder,
    "take_note":   take_note,
})

ASSISTANT_TOOLS = TOOLS + [
    {
        "name": "add_calendar_event",
        "description": "在日历中添加会议或事件",
        "input_schema": {
            "type": "object",
            "properties": {
                "title":            {"type": "string",  "description": "事件标题"},
                "date":             {"type": "string",  "description": "日期,格式 YYYY-MM-DD"},
                "time":             {"type": "string",  "description": "时间,格式 HH:MM"},
                "duration_minutes": {"type": "integer", "description": "持续时间(分钟),默认60"}
            },
            "required": ["title", "date", "time"]
        }
    },
    {
        "name": "list_calendar_events",
        "description": "查看日历事件,可按日期筛选",
        "input_schema": {
            "type": "object",
            "properties": {
                "date": {"type": "string", "description": "日期,格式 YYYY-MM-DD,不填则返回所有"}
            }
        }
    },
    {
        "name": "send_email",
        "description": "发送电子邮件",
        "input_schema": {
            "type": "object",
            "properties": {
                "to":      {"type": "string", "description": "收件人邮箱"},
                "subject": {"type": "string", "description": "邮件主题"},
                "body":    {"type": "string", "description": "邮件正文"}
            },
            "required": ["to", "subject", "body"]
        }
    },
    {
        "name": "create_reminder",
        "description": "创建一个定时提醒",
        "input_schema": {
            "type": "object",
            "properties": {
                "message":   {"type": "string", "description": "提醒内容"},
                "remind_at": {"type": "string", "description": "提醒时间,格式 YYYY-MM-DD HH:MM"}
            },
            "required": ["message", "remind_at"]
        }
    },
    {
        "name": "take_note",
        "description": "保存一条笔记",
        "input_schema": {
            "type": "object",
            "properties": {
                "title":   {"type": "string", "description": "笔记标题"},
                "content": {"type": "string", "description": "笔记内容"},
                "tags":    {"type": "array",  "items": {"type": "string"}, "description": "标签列表"}
            },
            "required": ["title", "content"]
        }
    },
]

ASSISTANT_SYSTEM = """你是一个高效的个人助理,能够帮助用户管理日程、发送邮件、查询信息和记录笔记。

工作原则:
- 对于复合任务,同时调用多个独立的工具(并行执行更高效)
- 对于有顺序依赖的任务,按正确顺序执行
- 完成操作后,给出简洁的确认摘要
- 如果信息不足(如日期不明确),先推断合理值再执行,并在回答中说明

当前时间:""" + datetime.datetime.now().strftime("%Y-%m-%d %H:%M")


# ── 测试个人助理 ──────────────────────────────────

assistant = MultiToolAgent(
    tools=ASSISTANT_TOOLS,
    model="claude-sonnet-4-6",
    max_turns=8,
    verbose=True,
)
assistant.system = ASSISTANT_SYSTEM

# 覆盖 run 方法以传入 system prompt
def assistant_run(msg: str) -> str:
    return assistant.run(msg, system=ASSISTANT_SYSTEM)

# 复合任务:同时做多件事
print("=" * 60)
print("任务:复合日程安排")
print("=" * 60)
result = assistant_run(
    "帮我:1)明天下午3点安排一个产品评审会议,持续90分钟 "
    "2)给 pm@company.com 发邮件通知这个会议 "
    "3)同时记录一条笔记:会议议题包括Q2路线图和用户反馈"
)
print(f"\n✅ 结果:{result}")

# 信息查询 + 计算组合
print("\n" + "=" * 60)
print("任务:跨工具信息整合")
print("=" * 60)
result = assistant_run(
    "查一下苹果和特斯拉的股价,告诉我如果各买100股需要多少人民币"
)
print(f"\n✅ 结果:{result}")

八、生产级优化:工具调用的健壮性

import time
from collections import defaultdict

class RobustMultiToolAgent(MultiToolAgent):
    """
    生产级 Agent,增加以下健壮性机制:
    1. 工具调用重试(网络抖动等临时失败)
    2. 循环检测(防止 Claude 反复调用同一工具)
    3. 调用频率限制(保护下游 API)
    4. 详细调用日志
    """

    def __init__(self, *args, max_retries: int = 2, rate_limit_per_tool: int = 5, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_retries          = max_retries
        self.rate_limit_per_tool  = rate_limit_per_tool
        self._call_counts: dict   = defaultdict(int)
        self._call_history: list  = []

    def _check_loop(self, tool_name: str, tool_input: dict) -> bool:
        """检测是否在重复调用同一个工具(同参数)"""
        call_sig = f"{tool_name}:{json.dumps(tool_input, sort_keys=True)}"
        recent   = self._call_history[-6:]   # 看最近6次调用
        repeat_count = sum(1 for h in recent if h == call_sig)
        return repeat_count >= 2             # 同样的调用出现2次以上,判定为循环

    def _execute_with_retry(self, name: str, inputs: dict) -> str:
        """带重试的工具执行"""
        # 频率限制检查
        if self._call_counts[name] >= self.rate_limit_per_tool:
            return json.dumps({"error": f"工具 {name} 调用次数超限({self.rate_limit_per_tool}次)"})

        # 循环检测
        if self._check_loop(name, inputs):
            return json.dumps({"error": f"检测到循环调用 {name},已停止重复执行"})

        call_sig = f"{name}:{json.dumps(inputs, sort_keys=True)}"
        self._call_history.append(call_sig)
        self._call_counts[name] += 1

        # 带重试执行
        last_error = None
        for attempt in range(self.max_retries + 1):
            try:
                result = execute_tool(name, inputs)
                parsed = json.loads(result)
                # 检查工具是否返回了错误
                if isinstance(parsed, dict) and "error" in parsed:
                    last_error = parsed["error"]
                    if attempt < self.max_retries:
                        time.sleep(0.5 * (attempt + 1))  # 指数退避
                        continue
                return result
            except Exception as e:
                last_error = str(e)
                if attempt < self.max_retries:
                    time.sleep(0.5 * (attempt + 1))

        return json.dumps({"error": f"重试{self.max_retries}次后仍失败:{last_error}"})

    def run(self, user_message: str, system: str = "") -> str:
        """重写 run,使用健壮版工具执行"""
        self._call_counts  = defaultdict(int)
        self._call_history = []
        messages = [{"role": "user", "content": user_message}]
        turn     = 0

        while turn < self.max_turns:
            turn += 1
            kwargs = dict(model=self.model, max_tokens=4096, tools=self.tools, messages=messages)
            if system:
                kwargs["system"] = system

            response = client.messages.create(**kwargs)

            if response.stop_reason == "end_turn":
                return " ".join(b.text for b in response.content if hasattr(b, "text"))

            if response.stop_reason != "tool_use":
                return " ".join(b.text for b in response.content if hasattr(b, "text"))

            tool_blocks  = [b for b in response.content if b.type == "tool_use"]
            tool_results = []

            for block in tool_blocks:
                result = self._execute_with_retry(block.name, block.input)
                tool_results.append({
                    "type":        "tool_result",
                    "tool_use_id": block.id,
                    "content":     result,
                })

            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user",      "content": tool_results})

        return "超过最大对话轮数"


# 使用健壮版 Agent
robust_agent = RobustMultiToolAgent(
    tools=ASSISTANT_TOOLS,
    max_turns=8,
    max_retries=2,
    rate_limit_per_tool=5,
    verbose=True,
)
result = robust_agent.run(
    "帮我查询数据库里所有用户,然后计算他们的总订单金额",
    system=ASSISTANT_SYSTEM,
)
print(result)

常见问题

Q:工具定义的 description 写得好不好,对调用准确率影响有多大?
影响非常大——description 是 Claude 决定"该不该调这个工具、传什么参数"的主要依据。好的 description 应该说清楚三件事:这个工具做什么(功能),什么时候用它(适用场景),以及重要的限制(不能做什么)。参数的 description 同样重要,要说明格式要求(如"日期格式 YYYY-MM-DD")和含义。

Q:Claude 调用了不该调的工具怎么办?
两个方向优化:一是改进工具 description,让不同工具的适用边界更清晰;二是在 System Prompt 里加约束,比如"只有在用户明确提到需要查询数据库时才调用 query_database"。如果某类任务从不需要某个工具,在该任务的请求里不传这个工具的定义——Claude 只能从你传给它的工具列表里选。

Q:并行工具调用时,工具之间有数据依赖怎么处理?
有依赖的工具不会被 Claude 并行调用——Claude 足够聪明,知道"先查股票代码再用代码查价格"这两个操作有顺序依赖,会分两批执行。你不需要在代码里特别处理,只要工具定义和 description 写清楚了,Claude 会自动判断哪些可以并行、哪些必须串行。

总结

Tool Use 的核心是三件事:写清楚工具的 description(影响调用准确率)、用并行调用优化性能(同批次工具并发执行)、做好错误处理和循环防护(生产环境稳定性)。工具定义的质量比代码实现更重要——Claude 靠 description 决策,description 模糊就会调错工具或传错参数。利用 tool_choice="tool" 做结构化信息提取,是比让 Claude 生成 JSON 更稳定的方案,适合需要严格格式保证的场景。