工具调用（Tool Use）深度集成

Tool Use API 架构

Computer Use 与 Tool Use 的关系

Anthropic 的 Tool Use（工具调用）是一项通用能力，允许 Claude 调用开发者定义的任意函数。Computer Use 工具（computer、text_editor、bash）实际上是 Anthropic 预定义的特殊工具。理解这一点，就能把两者有机结合：

Tool Use 消息流架构

┌─────────────────────────────────────────────┐
│  用户消息："帮我查一下北京的天气，然后截图桌面"     │
└──────────────────────┬──────────────────────┘
                       │
                       ▼
             ┌─────────────────┐
             │   Claude 推理    │
             └────────┬────────┘
                      │  决定调用工具
         ┌────────────┼────────────┐
         ▼            ▼            ▼
   get_weather   screenshot   (等待结果)
   (自定义工具)  (内置工具)
         │            │
         ▼            ▼
    返回天气数据   返回截图 base64
         │            │
         └──────┬─────┘
                ▼
         Claude 综合信息
         生成最终回复

自定义工具定义

JSON Schema 工具定义格式

# 自定义工具的完整定义格式
from typing import Any

# 工具定义列表（传给 API 的 tools 参数）
CUSTOM_TOOLS = [
    {
        "name": "get_weather",
        "description": """获取指定城市的实时天气信息。
        返回温度、湿度、风速和天气状况描述。
        当用户询问天气时使用此工具，比截图天气网站更高效。""",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称，如 '北京'、'Shanghai'、'New York'"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "温度单位，默认 celsius"
                }
            },
            "required": ["city"]
        }
    },
    {
        "name": "query_database",
        "description": """执行数据库查询并返回结果。
        只支持 SELECT 查询，不允许 INSERT/UPDATE/DELETE。
        当需要查询业务数据时使用，避免通过截图 Web 界面获取数据。""",
        "input_schema": {
            "type": "object",
            "properties": {
                "sql": {
                    "type": "string",
                    "description": "要执行的 SQL SELECT 语句"
                },
                "limit": {
                    "type": "integer",
                    "description": "返回记录数上限，默认 100",
                    "default": 100
                }
            },
            "required": ["sql"]
        }
    },
    {
        "name": "send_notification",
        "description": """发送通知到指定渠道（Slack/邮件/钉钉）。
        在任务完成或需要人工确认时调用此工具。""",
        "input_schema": {
            "type": "object",
            "properties": {
                "channel": {
                    "type": "string",
                    "enum": ["slack", "email", "dingtalk"]
                },
                "message": {
                    "type": "string",
                    "description": "通知内容"
                },
                "urgency": {
                    "type": "string",
                    "enum": ["low", "normal", "high"],
                    "default": "normal"
                }
            },
            "required": ["channel", "message"]
        }
    }
]

工具执行循环

完整的工具执行引擎

import anthropic
import json
from typing import Callable, Optional


class ToolExecutor:
    """工具执行引擎：管理工具注册和执行"""

    def __init__(self):
        self._handlers: dict[str, Callable] = {}
        self._tool_defs: list[dict] = []

    def register(self, tool_def: dict, handler: Callable):
        """注册工具定义和对应的处理函数"""
        self._tool_defs.append(tool_def)
        self._handlers[tool_def["name"]] = handler

    async def execute(self, tool_name: str, tool_input: dict) -> dict:
        """执行指定工具并返回结果"""
        handler = self._handlers.get(tool_name)
        if not handler:
            return {"error": f"Unknown tool: {tool_name}"}
        try:
            result = await handler(**tool_input)
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}

    @property
    def tool_definitions(self) -> list[dict]:
        return self._tool_defs


async def run_agent_with_tools(
    client: anthropic.Anthropic,
    executor: ToolExecutor,
    user_message: str,
    computer_use_tools: list[dict],  # Computer Use 内置工具
    max_iterations: int = 30
) -> str:
    """
    混合工具 Agent 主循环：
    同时支持 Computer Use 内置工具和自定义工具。
    """
    # 合并内置工具和自定义工具
    all_tools = computer_use_tools + executor.tool_definitions

    messages = [{"role": "user", "content": user_message}]

    for iteration in range(max_iterations):
        response = client.beta.messages.create(
            model="claude-opus-4-5",
            max_tokens=4096,
            tools=all_tools,
            messages=messages,
            betas=["computer-use-2024-10-22"]
        )

        # 收集工具调用
        tool_use_blocks = [
            block for block in response.content
            if block.type == "tool_use"
        ]

        # 没有工具调用，任务完成
        if not tool_use_blocks or response.stop_reason == "end_turn":
            # 提取最终文本回复
            text_blocks = [b.text for b in response.content if b.type == "text"]
            return "\n".join(text_blocks)

        # 将 Claude 的回复加入 messages
        messages.append({"role": "assistant", "content": response.content})

        # 执行所有工具调用（支持并行）
        tool_results = []
        import asyncio
        tasks = [
            executor.execute(block.name, block.input)
            for block in tool_use_blocks
        ]
        results = await asyncio.gather(*tasks)

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

        # 将工具结果作为 user 消息返回
        messages.append({"role": "user", "content": tool_results})

    return "[超过最大迭代次数]"

并行工具调用

让 Claude 同时调用多个工具

当任务中有多个相互独立的操作时，Claude 会在单次回复中输出多个 tool_use 块。正确处理这种情况可以显著提升效率：

import asyncio
from dataclasses import dataclass


@dataclass
class ToolCallResult:
    tool_use_id: str
    tool_name: str
    success: bool
    result: Any
    error: Optional[str] = None
    duration_ms: float = 0.0


async def execute_tools_in_parallel(
    tool_use_blocks: list,
    executor: ToolExecutor,
    computer_controller  # Computer Use 控制器
) -> list[ToolCallResult]:
    """
    并行执行所有工具调用。
    Computer Use 工具（截图等）和自定义工具同步并行执行。
    """
    async def execute_one(block) -> ToolCallResult:
        start = asyncio.get_event_loop().time()
        try:
            # 判断是 Computer Use 内置工具还是自定义工具
            if block.name in ("computer", "str_replace_editor", "bash"):
                result = await computer_controller.execute_action(
                    block.input.get("action"),
                    **{k: v for k, v in block.input.items() if k != "action"}
                )
            else:
                result = await executor.execute(block.name, block.input)

            duration = (asyncio.get_event_loop().time() - start) * 1000
            return ToolCallResult(
                tool_use_id=block.id,
                tool_name=block.name,
                success=True,
                result=result,
                duration_ms=duration
            )
        except Exception as e:
            duration = (asyncio.get_event_loop().time() - start) * 1000
            return ToolCallResult(
                tool_use_id=block.id,
                tool_name=block.name,
                success=False,
                result=None,
                error=str(e),
                duration_ms=duration
            )

    tasks = [execute_one(block) for block in tool_use_blocks]
    return await asyncio.gather(*tasks)


def build_tool_results_message(call_results: list[ToolCallResult]) -> dict:
    """将工具执行结果构建为 API 所需的 message 格式"""
    content = []
    for r in call_results:
        if r.success:
            # Computer Use 截图需要特殊格式
            if (r.tool_name == "computer" and
                    isinstance(r.result, dict) and
                    r.result.get("type") == "image"):
                content.append({
                    "type": "tool_result",
                    "tool_use_id": r.tool_use_id,
                    "content": [r.result]  # 截图需要包装成列表
                })
            else:
                content.append({
                    "type": "tool_result",
                    "tool_use_id": r.tool_use_id,
                    "content": json.dumps(r.result, ensure_ascii=False)
                })
        else:
            content.append({
                "type": "tool_result",
                "tool_use_id": r.tool_use_id,
                "is_error": True,
                "content": f"Error: {r.error}"
            })
    return {"role": "user", "content": content}

工具结果错误处理

优雅地处理工具失败

from enum import Enum


class ToolErrorStrategy(Enum):
    # 将错误返回给 Claude，让它决定如何处理
    REPORT_TO_CLAUDE = "report"
    # 重试最多 N 次
    RETRY = "retry"
    # 使用备用工具
    FALLBACK = "fallback"
    # 直接终止任务
    ABORT = "abort"


class ResilientToolExecutor(ToolExecutor):
    """带弹性策略的工具执行器"""

    def __init__(self):
        super().__init__()
        self._retry_config: dict[str, int] = {}   # tool_name -> max_retries
        self._fallbacks: dict[str, str] = {}        # tool_name -> fallback_tool

    def set_retry(self, tool_name: str, max_retries: int = 3):
        """为指定工具配置重试次数"""
        self._retry_config[tool_name] = max_retries

    def set_fallback(self, tool_name: str, fallback_tool: str):
        """为指定工具配置备用工具"""
        self._fallbacks[tool_name] = fallback_tool

    async def execute(self, tool_name: str, tool_input: dict) -> dict:
        max_retries = self._retry_config.get(tool_name, 1)

        for attempt in range(max_retries):
            result = await super().execute(tool_name, tool_input)
            if result.get("success"):
                return result

            # 最后一次重试失败，尝试备用工具
            if attempt == max_retries - 1:
                fallback = self._fallbacks.get(tool_name)
                if fallback:
                    print(f"[工具失败] {tool_name} 切换到备用工具 {fallback}")
                    return await super().execute(fallback, tool_input)

            # 等待后重试
            await asyncio.sleep(0.5 * (attempt + 1))

        return result  # 返回最后一次失败的结果给 Claude

工具链：串联工具输出

将一个工具的输出作为另一个的输入

工具链（Tool Chain）是指通过提示词引导 Claude 在多轮对话中依次调用工具，每一步的输出成为下一步的输入：

"""
工具链示例：数据提取 → 处理 → 入库 → 通知

任务：从网页截图中提取订单信息，写入数据库，发送确认通知
工具链：
  1. computer(screenshot) → 获取页面截图
  2. extract_order_data(image) → 解析订单字段
  3. insert_order(order_data) → 写入数据库
  4. send_notification(order_id) → 通知相关人员
"""

async def handle_order_workflow(page_url: str):
    # 定义工具
    tools = [
        # Computer Use 内置工具
        {"type": "computer_20241022", "name": "computer", "display_width_px": 1280, "display_height_px": 800},
        # 自定义工具
        {
            "name": "extract_order_data",
            "description": "从 HTML 或文本内容中解析订单数据，返回结构化 JSON",
            "input_schema": {
                "type": "object",
                "properties": {
                    "raw_content": {"type": "string", "description": "原始页面内容"}
                },
                "required": ["raw_content"]
            }
        },
        {
            "name": "insert_order",
            "description": "将订单数据插入数据库，返回新订单的 ID",
            "input_schema": {
                "type": "object",
                "properties": {
                    "order_data": {"type": "object", "description": "订单 JSON 数据"}
                },
                "required": ["order_data"]
            }
        },
        {
            "name": "send_notification",
            "description": "发送订单确认通知",
            "input_schema": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "channel": {"type": "string"}
                },
                "required": ["order_id", "channel"]
            }
        }
    ]

    task = f"""
    请按照以下步骤处理订单：
    1. 打开浏览器导航到 {page_url}，截图查看页面
    2. 使用 extract_order_data 工具从页面内容中提取订单信息
    3. 使用 insert_order 工具将订单写入数据库
    4. 使用 send_notification 工具通过 slack 发送确认通知
    5. 报告整个流程的执行结果

    注意：每一步都要确认上一步成功后再进行下一步。
    """

    # 运行 Agent
    result = await run_agent_with_tools(client, executor, task, tools)
    return result

Computer Use + 自定义工具混合使用

最优工具选择策略

合理的系统提示词可以引导 Claude 在正确的场景下选择正确的工具，避免过度依赖截图操作：

SYSTEM_PROMPT = """你是一个智能自动化助手，拥有以下工具：

【Computer Use 工具】用于屏幕操作：
- computer: 截图、鼠标点击、键盘输入（当需要操作 GUI 时使用）

【自定义工具】用于直接获取数据（优先使用这些工具）：
- query_database: 当需要查询业务数据时直接调用，不要截图 Web 界面
- send_notification: 发送通知，不要手动打开 Slack/邮件客户端
- get_file_content: 读取服务器文件，不要截图文件内容

工具选择原则：
1. 如果有专用工具可以完成任务，优先使用专用工具
2. 只有当必须通过 GUI 操作（如遗留系统没有 API）时才使用 computer 工具
3. 需要查看操作结果时，再使用截图确认
4. 不要重复截图，每次截图都有 Token 成本"""

# 使用示例：查询订单后截图确认
task_example = """
查询最近 10 笔订单，然后打开内部系统的订单页面截图确认数据一致性。

期望行为：
- 先调用 query_database 获取数据库中的订单（高效）
- 再用 computer 工具打开浏览器，截图系统页面
- 对比两者数据是否一致
"""

工具版本管理

处理 API 工具版本变更

from enum import Enum


class ComputerUseVersion(Enum):
    # 2024年10月发布的版本（当前稳定版）
    V_20241022 = "computer_20241022"


def build_computer_use_tools(
    version: ComputerUseVersion = ComputerUseVersion.V_20241022,
    display_width: int = 1280,
    display_height: int = 800
) -> list[dict]:
    """构建 Computer Use 工具列表，支持版本切换"""
    computer_type = version.value
    # text_editor 和 bash 的版本号与 computer 对应
    version_suffix = computer_type.replace("computer_", "")

    return [
        {
            "type": computer_type,
            "name": "computer",
            "display_width_px": display_width,
            "display_height_px": display_height,
            "display_number": 1
        },
        {
            "type": f"text_editor_{version_suffix}",
            "name": "str_replace_editor"
        },
        {
            "type": f"bash_{version_suffix}",
            "name": "bash"
        }
    ]


# 自定义工具版本管理：通过工具名称后缀区分版本
def get_tool_registry_v2() -> list[dict]:
    """v2 版本工具集：新增 batch_query 支持"""
    return [
        {"name": "query_database_v2", "description": "支持批量查询的新版数据库工具", "input_schema": {
            "type": "object",
            "properties": {
                "queries": {"type": "array", "items": {"type": "string"}},
                "parallel": {"type": "boolean", "default": True}
            },
            "required": ["queries"]
        }}
    ]

工具调用的内部机制

Claude 如何决定是否调用工具

了解 Claude 在工具调用背后的决策逻辑，有助于你设计更有效的工具和系统提示词：

tool_choice 参数详解

import anthropic

client = anthropic.Anthropic()

# tool_choice 的三种模式

# 1. auto（默认）：Claude 自主决定是否调用工具
response = client.beta.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    tools=tools,
    # tool_choice 不设置，等同于 tool_choice={"type": "auto"}
    messages=[{"role": "user", "content": "北京天气怎么样？"}]
)

# 2. any：强制 Claude 必须调用至少一个工具（但可以选择哪个）
response = client.beta.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "any"},  # 必须调用工具
    messages=[{"role": "user", "content": "我需要你帮我做一些事"}]
)

# 3. tool：强制调用指定的工具
response = client.beta.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    tools=tools,
    tool_choice={
        "type": "tool",
        "name": "get_weather"  # 强制调用 get_weather，无论用户说什么
    },
    messages=[{"role": "user", "content": "查询北京天气"}]
)

# 4. Computer Use 场景的 tool_choice
# Computer Use 工具（computer/bash/text_editor）不能与 tool_choice="tool" 一起使用
# 它们必须依赖 Claude 的自主判断（auto 模式）

自定义工具的最佳设计模式

# 优秀的工具描述示例

GOOD_TOOL_EXAMPLE = {
    "name": "search_products",
    "description": """在产品数据库中搜索产品。

适用场景：
- 用户需要查找特定产品时
- 需要获取产品 ID 用于后续操作时
- 不适合用于修改产品信息（请用 update_product）

返回值：最多 10 条产品记录的列表，每条包含 id、名称、价格、库存。

示例调用：
- 搜索名称含"手机"的产品：{"keyword": "手机"}
- 按分类搜索：{"category": "electronics", "max_price": 5000}
- 搜索低库存产品：{"min_stock": 0, "max_stock": 10}""",
    "input_schema": {
        "type": "object",
        "properties": {
            "keyword": {
                "type": "string",
                "description": "产品名称关键词，支持模糊匹配（如 '手机' 会匹配 '苹果手机'）"
            },
            "category": {
                "type": "string",
                "enum": ["electronics", "clothing", "food", "other"],
                "description": "产品分类，精确匹配"
            },
            "max_price": {
                "type": "number",
                "description": "最高价格（元），仅返回价格不超过此值的产品"
            },
            "max_stock": {
                "type": "integer",
                "description": "最大库存量，用于查找库存不足的产品"
            },
            "limit": {
                "type": "integer",
                "description": "返回结果数量上限，默认 10，最大 50",
                "default": 10
            }
        },
        "required": []  # 所有参数可选，至少传入一个
    }
}

# 差的工具描述（反例）
BAD_TOOL_EXAMPLE = {
    "name": "search",
    "description": "搜索",  # 太模糊，Claude 不知道搜索什么
    "input_schema": {
        "type": "object",
        "properties": {
            "q": {"type": "string"}  # 参数名和描述都太简单
        }
    }
}