使用Claude API进行程序化工具调用（PTC）

本教程面向有编程基础的开发者，深入讲解Claude API的程序化工具调用（PTC）技术。通过一个团队费用管理模拟API的实战案例，对比传统工具调用与PTC的性能差异，展示PTC如何通过让模型编写代码在代码执行环境中处理数据，大幅降低令牌消耗和端到端延迟。教程涵盖环境搭建、工具定义、代码实现、性能对比及适用场景，帮助开发者掌握PTC的核心原理与工程实践。

技术背景

在AI代理应用中，工具调用是模型与外部系统交互的关键机制。传统工具调用中，每次工具调用都需要模型参与，导致多次往返延迟，且大量原始数据（如冗长的JSON响应）会占用模型上下文窗口，增加令牌消耗。PTC允许模型编写代码在代码执行环境中调用工具，从而减少往返次数，并通过代码处理数据，避免不相关上下文进入模型。该技术特别适用于处理大型数据集、存在顺序依赖或需要重复调用的场景，能显著提升效率并降低成本。

核心原理

PTC的核心原理是让模型生成Python代码，在沙箱化的代码执行环境中运行，通过代码调用工具API。模型通过特殊的code_execution工具类型获得代码执行能力，工具定义中通过allowed_callers字段指定哪些工具可由代码执行环境调用。当模型决定使用代码执行时，API返回包含代码的tool_use块，系统在容器中执行代码，代码内部可调用其他工具（如get_expenses），并将结果直接处理（如解析、过滤、聚合），最终将处理后的结果返回给模型。这样，原始数据不会进入模型上下文，模型只接收精简后的结果，从而大幅减少令牌消耗。

架构分析

系统架构包含三个核心组件：Claude模型、代码执行容器和外部工具API。模型负责生成代码和决策，代码执行容器提供隔离的Python环境，外部工具API提供数据访问。工作流程为：模型收到用户请求后，生成代码并请求代码执行；容器执行代码，代码内部调用工具API获取数据，进行本地处理；处理结果返回给模型，模型基于结果继续推理或生成最终答案。关键设计决策包括：通过allowed_callers控制工具调用权限，确保安全；使用container_id维持有状态执行，支持多轮交互；代码执行环境与模型上下文分离，实现数据过滤。

完整实战教程

使用Claude API进行程序化工具调用（PTC）

程序化工具调用（PTC）允许Claude在代码执行环境中编写代码以编程方式调用工具，而不是每次工具调用都需要通过模型进行往返。这大大减少了多次工具调用的端到端延迟，并通过让模型编写代码在上下文窗口之前移除不相关的上下文（例如，在大型嘈杂文件中grep关键信息），从而显著降低令牌消耗。

当面对无法直接修改的第三方API和工具时，PTC可以通过允许Claude编写可在代码执行环境中调用的代码来帮助减少上下文的使用。

在本教程中，我们将使用一个用于团队费用管理的模拟API。该API设计为需要多次调用，并返回大量结果，这有助于说明程序化工具调用的优势。

学完本教程后，你将能够：

理解常规工具调用与程序化工具调用（PTC）之间的区别
编写利用 PTC 的智能体

前置条件

在遵循本指南之前，请确保你具备：

所需知识

Python 基础知识——熟悉 async/await、函数和基本数据结构
对代理模式和工具调用的基本理解

所需工具

Python 3.11 或更高版本
Anthropic API 密钥
Anthropic Python SDK >= 0.72

环境搭建

首先，安装所需的依赖：

# %pip install -r requirements.txt

注意：确保你的 .env 文件包含：

ANTHROPIC_API_KEY=your_key_here

加载你的环境变量并配置客户端。我们还加载了一个辅助工具来可视化 Claude 消息响应。

from dotenv import load_dotenv
from utils.visualize import visualize

load_dotenv()

MODEL = "claude-sonnet-4-6"

viz = visualize(auto_show=True)

理解第三方 API

在 utils/team_expense_api.py 中定义了三个函数：get_team_members、get_expenses 和 get_custom_budget。get_team_members 函数允许我们检索给定部门的所有员工，包括他们的角色、级别和联系信息。get_expenses 函数返回员工在特定季度的所有费用明细项——每个员工可能有数百条记录，每条记录包含大量元数据，包括收据 URL、审批链、商户详情等。get_custom_budget 函数检查特定员工是否有自定义差旅预算例外（否则他们使用标准的每季度 5000 美元限额）。

在此场景中，我们需要分析团队费用并确定哪些员工超出了预算。传统上，我们可能会手动提取每个人的费用报告，按类别汇总他们的费用，与预算限额进行比较（检查自定义预算例外），并编制报告。相反，我们将要求 Claude 为我们执行此分析，使用可用工具检索团队数据，获取可能包含丰富元数据的数百条费用明细项，并确定谁超出了预算。

这里的关键挑战是，每个员工可能有 100 多条费用明细项需要获取、解析和汇总——而 get_custom_budget 工具只能在分析费用以查看是否有人超出标准预算后调用。这创建了一个顺序依赖链，使其成为展示编程式工具调用优势的理想用例。

我们将把工具定义传递给 messages API，并要求 Claude 执行分析。如果您不熟悉 Claude API 中工具使用的工作原理，请阅读实现工具使用的文档。

import json

import anthropic
from utils.team_expense_api import get_custom_budget, get_expenses, get_team_members

client = anthropic.Anthropic()

# Tool definitions for the team expense API
tools = [
    {
        "name": "get_team_members",
        "description": 'Returns a list of team members for a given department. Each team member includes their ID, name, role, level (junior, mid, senior, staff, principal), and contact information. Use this to get a list of people whose expenses you want to analyze. Available departments are: engineering, sales, and marketing.\n\nRETURN FORMAT: Returns a JSON string containing an ARRAY of team member objects (not wrapped in an outer object). Parse with json.loads() to get a list. Example: [{"id": "ENG001", "name": "Alice", ...}, {"id": "ENG002", ...}]',
        "input_schema": {
            "type": "object",
            "properties": {
                "department": {
                    "type": "string",
                    "description": "The department name. Case-insensitive.",
                }
            },
            "required": ["department"],
        },
        "input_examples": [
            {"department": "engineering"},
            {"department": "sales"},
            {"department": "marketing"},
        ],
    },
    {
        "name": "get_expenses",
        "description": "Returns all expense line items for a given employee in a specific quarter. Each expense includes extensive metadata: date, category, description, amount (in USD), currency, status (approved, pending, rejected), receipt URL, approval chain, merchant name and location, payment method, and project codes. An employee may have 20-50+ expense line items per quarter, and each line item contains substantial metadata for audit and compliance purposes. Categories include: 'travel' (flights, trains, rental cars, taxis, parking), 'lodging' (hotels, airbnb), 'meals', 'software', 'equipment', 'conference', 'office', and 'internet'. IMPORTANT: Only expenses with status='approved' should be counted toward budget limits.\n\nRETURN FORMAT: Returns a JSON string containing an ARRAY of expense objects (not wrapped in an outer object with an 'expenses' key). Parse with json.loads() to get a list directly. Example: [{\"expense_id\": \"ENG001_Q3_001\", \"amount\": 1250.50, \"category\": \"travel\", ...}, {...}]",
        "input_schema": {
            "type": "object",
            "properties": {
                "employee_id": {
                    "type": "string",
                    "description": "The unique employee identifier",
                },
                "quarter": {
                    "type": "string",
                    "description": "Quarter identifier: 'Q1', 'Q2', 'Q3', or 'Q4'",
                },
            },
            "required": ["employee_id", "quarter"],
        },
        "input_examples": [
            {"employee_id": "ENG001", "quarter": "Q3"},
            {"employee_id": "SAL002", "quarter": "Q1"},
            {"employee_id": "MKT001", "quarter": "Q4"},
        ],
    },
    {
        "name": "get_custom_budget",
        "description": 'Get the custom quarterly travel budget for a specific employee. Most employees have a standard $5,000 quarterly travel budget. However, some employees have custom budget exceptions based on their role requirements. This function checks if a specific employee has a custom budget assigned.\n\nRETURN FORMAT: Returns a JSON string containing a SINGLE OBJECT (not an array). Parse with json.loads() to get a dict. Example: {"user_id": "ENG001", "has_custom_budget": false, "travel_budget": 5000, "reason": "Standard", "currency": "USD"}',
        "input_schema": {
            "type": "object",
            "properties": {
                "user_id": {
                    "type": "string",
                    "description": "The unique employee identifier",
                }
            },
            "required": ["user_id"],
        },
        "input_examples": [
            {"user_id": "ENG001"},
            {"user_id": "SAL002"},
            {"user_id": "MKT001"},
        ],
    },
]

tool_functions = {
    "get_team_members": get_team_members,
    "get_expenses": get_expenses,
    "get_custom_budget": get_custom_budget,
}

传统工具调用（基线）

在第一个示例中，我们将使用传统工具调用来建立基线。

我们将使用初始查询调用 messages.create API。当模型以 tool_use 原因停止时，我们将按请求执行工具，然后将工具的输出添加到消息中，并再次调用模型。

import time

from anthropic.types import TextBlock, ToolUseBlock
from anthropic.types.beta import (
    BetaMessageParam as MessageParam,
)
from anthropic.types.beta import (
    BetaTextBlock,
    BetaToolUseBlock,
)

messages: list[MessageParam] = []


def run_agent_without_ptc(user_message):
    """Run agent using traditional tool calling"""
    messages.append({"role": "user", "content": user_message})
    total_tokens = 0
    start_time = time.time()
    api_counter = 0

    while True:
        response = client.beta.messages.create(
            model=MODEL,
            max_tokens=4000,
            tools=tools,
            messages=messages,
            betas=["advanced-tool-use-2025-11-20"],
        )

        api_counter += 1

        # Track token usage
        total_tokens += response.usage.input_tokens + response.usage.output_tokens
        viz.capture(response)
        if response.stop_reason == "end_turn":
            # Extract the first text block from the response
            final_response = next(
                (
                    block.text
                    for block in response.content
                    if isinstance(block, (BetaTextBlock, TextBlock))
                ),
                None,
            )
            elapsed_time = time.time() - start_time
            return final_response, messages, total_tokens, elapsed_time, api_counter

        # Process tool calls
        if response.stop_reason == "tool_use":
            # First, add the assistant's response to messages
            messages.append({"role": "assistant", "content": response.content})

            # Collect all tool results
            tool_results = []

            for block in response.content:
                if isinstance(block, (BetaToolUseBlock, ToolUseBlock)):
                    tool_name = block.name
                    tool_input = block.input
                    tool_use_id = block.id

                    result = tool_functions[tool_name](**tool_input)

                    content = str(result)

                    tool_result = {
                        "type": "tool_result",
                        "tool_use_id": tool_use_id,
                        "content": content,
                    }
                    tool_results.append(tool_result)

            # Append all tool results at once after collecting them
            messages.append({"role": "user", "content": tool_results})

        else:
            print(f"\nUnexpected stop reason: {response.stop_reason}")
            elapsed_time = time.time() - start_time

            final_response = next(
                (
                    block.text
                    for block in response.content
                    if isinstance(block, (BetaTextBlock, TextBlock))
                ),
                f"Stopped with reason: {response.stop_reason}",
            )
            return final_response, messages, total_tokens, elapsed_time, api_counter

我们向模型发出的初始查询提供了一些指令，以帮助引导模型。为简洁起见，我们要求模型每个工具只调用一次。对于更深入的调查，模型可能希望查看多个系统或时间跨度。

query = "Which engineering team members exceeded their Q3 travel budget? Standard quarterly travel budget is $5,000. However, some employees have custom budget limits. For anyone who exceeded the $5,000 standard budget, check if they have a custom budget exception. If they do, use that custom limit instead to determine if they truly exceeded their budget."

# Run the agent
result, conversation, total_tokens, elapsed_time, api_count_without_ptc = run_agent_without_ptc(
    query
)

print(f"Result: {result}")
print(f"API calls made: {api_count_without_ptc}")
print(f"Total tokens used: {total_tokens:,}")
print(f"Total time taken: {elapsed_time:.2f}s")

很好！我们可以看到 Claude 能够成功使用可用工具来识别哪些团队成员超出了他们的旅行预算。然而，我们也可以看到我们使用了大量的令牌来完成这项任务。Claude 必须通过其上下文窗口摄取所有费用明细项——每位员工可能有 100 多条记录，每条记录都包含大量元数据，包括收据 URL、审批链、商户信息等——以便解析它们、按类别汇总并与预算限制进行比较。

此外，传统的工具调用方法需要多次顺序往返：首先获取团队成员，然后获取每个人的费用，然后检查超出标准限额的人员的自定义预算。每次往返都会增加延迟，并且费用记录中的所有丰富元数据都会流经模型的上下文。

让我们看看是否可以使用 PTC 来改进性能，让 Claude 编写代码在代码执行环境中处理这些大型数据集。

要在工具上启用 PTC，我们必须首先向任何应通过代码执行调用的工具添加 allowed_callers 字段。

需要考虑的要点

没有 allowed_callers 的工具默认为仅模型调用
通过包含多个调用者，工具可以被模型和代码执行同时调用：["direct", "code_execution_20250825"]
仅选择对程序化/重复执行安全的工具。

import copy

ptc_tools = copy.deepcopy(tools)
for tool in ptc_tools:
    tool["allowed_callers"] = ["code_execution_20250825"]  # type: ignore


# Add the code execution tool
ptc_tools.append(
    {
        "type": "code_execution_20250825",  # type: ignore
        "name": "code_execution",
    }
)

现在我们已经更新了工具定义以允许程序化工具调用，我们可以使用PTC运行我们的代理。为此，我们对函数进行了一些更改。我们必须使用beta消息API。

我们在betas中添加了"advanced-tool-use-2025-11-20"。
如果定义了container_id，我们会在请求中传递它。这仅对于像我们这样的有状态工作流是必要的。在单轮工作流中，这不是必需的。
我们可以检查tool_use块中的caller字段，以确定此工具调用是来自直接模型调用还是程序化调用。

请注意，在任一情况下，我们都通过Claude API发送工具结果，但只有direct调用会被模型“看到”。code_execution_20250825类型只会被代码执行容器看到。

messages = []


def run_agent_with_ptc(user_message):
    """Run agent using PTC"""
    messages.append({"role": "user", "content": user_message})
    total_tokens = 0
    start_time = time.time()
    container_id = None
    api_counter = 0

    while True:
        # Build request with PTC beta headers
        request_params = {
            "model": MODEL,
            "max_tokens": 4000,
            "tools": ptc_tools,
            "messages": messages,
        }

        response = client.beta.messages.create(
            **request_params,
            betas=[
                "advanced-tool-use-2025-11-20",
            ],
            extra_body={"container": container_id} if container_id else None,
        )
        viz.capture(response)
        api_counter += 1

        # Track container for stateful execution
        if hasattr(response, "container") and response.container:
            container_id = response.container.id
            print(f"\n[Container] ID: {container_id}")
            if hasattr(response.container, "expires_at"):
                # If the container has expired, we would need to restart our workflow. In our case, it completes before expiration.
                print(f"[Container] Expires at: {response.container.expires_at}")

        # Track token usage
        total_tokens += response.usage.input_tokens + response.usage.output_tokens

        if response.stop_reason == "end_turn":
            # Extract the first text block from the response
            final_response = next(
                (block.text for block in response.content if isinstance(block, BetaTextBlock)),
                None,
            )
            elapsed_time = time.time() - start_time
            return final_response, messages, total_tokens, elapsed_time, api_counter

        # As before, we process tool calls
        if response.stop_reason == "tool_use":
            # First, add the assistant's response to messages
            messages.append({"role": "assistant", "content": response.content})

            # Collect all tool results
            tool_results = []

            for block in response.content:
                if isinstance(block, BetaToolUseBlock):
                    tool_name = block.name
                    tool_input = block.input
                    tool_use_id = block.id

                    # We can use caller type to understand how the tool was invoked
                    caller_type = block.caller["type"]  # type: ignore

                    if caller_type == "code_execution_20250825":
                        print(f"[PTC] Tool called from code execution environment: {tool_name}")

                    elif caller_type == "direct":
                        print(f"[Direct] Tool called by model: {tool_name}")

                    result = tool_functions[tool_name](**tool_input)

                    # Format result as proper content for the API
                    if isinstance(result, list) and result and isinstance(result[0], str):
                        content = "\n".join(result)
                    elif isinstance(result, (dict, list)):
                        content = json.dumps(result)
                    else:
                        content = str(result)

                    tool_results.append(
                        {
                            "type": "tool_result",
                            "tool_use_id": tool_use_id,
                            "content": content,
                        }
                    )

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

        else:
            print(f"\nUnexpected stop reason: {response.stop_reason}")
            elapsed_time = time.time() - start_time

            final_response = next(
                (block.text for block in response.content if isinstance(block, BetaTextBlock)),
                f"Stopped with reason: {response.stop_reason}",
            )
            return final_response, messages, total_tokens, elapsed_time, api_counter

# Run the PTC agent
result_ptc, conversation_ptc, total_tokens_ptc, elapsed_time_ptc, api_count_with_ptc = (
    run_agent_with_ptc(query)
)

print(f"\n{'=' * 60}")
print(f"Result: {result_ptc}")
print(f"\n{'=' * 60}")
print("Performance Metrics:")
print(
    f"  Total API calls to Claude: {len([m for m in conversation_ptc if m['role'] == 'assistant'])}"
)
print(f"  Total tokens used: {total_tokens_ptc:,}")
print(f"  Total time taken: {elapsed_time_ptc:.2f}s")

性能对比

让我们比较传统工具调用与 PTC 之间的性能：

关于 API 调用次数的说明： 您可能会注意到，在此示例中 PTC 需要更多的 API 调用。这是因为 PTC 编写了更结构化、顺序化的代码，遵循最佳实践——例如，将费用获取步骤与预算检查步骤分开。传统工具调用有时可以在单轮中批量处理操作，但代价是将所有原始数据通过模型的上下文发送。PTC 带来的令牌效率提升远远超过往返次数的少量增加，尤其是在处理大型、元数据丰富的数据集时。

import pandas as pd

# Create comparison dataframe
comparison_data = {
    "Metric": [
        "API Calls",
        "Total Tokens",
        "Elapsed Time (s)",
        "Token Reduction",
        "Time Reduction",
    ],
    "Traditional": [
        api_count_without_ptc,
        f"{total_tokens:,}",
        f"{elapsed_time:.2f}",
        "-",
        "-",
    ],
    "PTC": [
        api_count_with_ptc,
        f"{total_tokens_ptc:,}",
        f"{elapsed_time_ptc:.2f}",
        f"{((total_tokens - total_tokens_ptc) / total_tokens * 100):.1f}%",
        f"{((elapsed_time - elapsed_time_ptc) / elapsed_time * 100):.1f}%",
    ],
}

df = pd.DataFrame(comparison_data)
print(df.to_string(index=False))

关键要点

在本例中，PTC 通过三项核心能力展示了显著的性能提升：

1. 通过大数据解析保持上下文

这是我们工作流程中展示的主要优势。Claude 编写代码，在代码执行环境中获取并处理数百个费用明细项。通过以编程方式处理这些数据，Claude 解析了 JSON、按状态筛选、按类别汇总金额，并与预算限额进行比较——所有这些都无需将原始费用数据和元数据发送到模型的上下文窗口中。这导致了token 使用量的大幅减少。

2. 顺序依赖优化

API 存在顺序依赖：get_custom_budget(user_id) 只能在分析费用以确定谁超出了标准 5,000 美元预算后才能调用。在传统的工具调用中，这需要多次往返——获取团队成员、获取每个人的费用、识别超出预算的人员，然后逐一检查他们的自定义预算。使用 PTC，Claude 编写的代码在代码执行环境中编排了整个工作流程，在循环中进行程序化工具调用，并在调用之间保持状态。这将许多顺序 API 往返转换为更少的调用，并实现了更智能的编排。

3. 代码执行中的计算逻辑

Claude 没有让模型在脑海中跟踪和汇总数十个带有复杂元数据的费用，而是将算术和聚合逻辑委托给 Python 代码。这减少了模型的认知负担，确保了精确的计算，并将不相关的元数据（如收据 URL 和商家位置）完全排除在模型上下文之外。

何时使用 PTC

PTC 在以下情况下最为有益：

处理大型、元数据丰富的数据集，需要筛选、解析或聚合（如我们的费用分析，包含收据 URL、审批链、商家详细信息等）
存在顺序依赖，其中一个工具调用依赖于先前调用的结果（如仅检查超出标准限额的员工的预算）
需要按顺序或循环对相似实体进行多次工具调用（如检查每个团队成员的费用和预算）
计算逻辑可以减少需要通过模型上下文流动的内容
工具对于程序化/重复执行是安全的，无需人工监督

结论

我们的团队费用分析展示了 PTC 的优势：在处理大型、元数据丰富的数据集时大幅减少上下文消耗，以及优化具有顺序依赖的工作流程。通过让 Claude 编写代码来编排工具调用并以编程方式处理结果，我们在保持准确性和洞察质量的同时，实现了可观的 token 节省。

PTC 对于涉及批量数据处理（具有丰富元数据）、重复工具调用（具有依赖关系）或原始工具输出会污染模型上下文的工作流程尤其有价值。

后续步骤

尝试将此模式应用于您自己的用例：

具有顺序查找的财务数据分析和报告
依赖于初始扫描结果的多实体健康检查
具有元数据的大型文件处理（CSV、JSON、XML 解析）
数据库查询结果聚合及后续查询
基于初始结果进行条件逻辑的批量 API 操作

常见问题

常见错误包括：1. 未在工具定义中添加allowed_callers字段，导致代码执行环境无法调用工具；2. 忘记添加code_execution工具类型，模型无法生成代码；3. 未使用beta API或未包含advanced-tool-use-2025-11-20 beta标志；4. 容器过期导致状态丢失，需重新初始化；5. 工具返回格式不符合预期，导致代码解析失败。排查方法：检查API请求参数，确保包含正确的beta标志和工具定义；查看响应中的container信息，确认容器状态；使用可视化工具（如viz.capture）跟踪消息流；逐步调试代码执行环境中的异常。生产环境优化：使用有状态容器减少重复初始化；对工具调用进行缓存；设置合理的max_tokens限制；监控令牌使用和延迟，动态调整策略。

企业应用扩展

PTC可扩展应用于多种场景：1. 财务数据分析与报告，如批量处理发票、费用明细；2. 多实体健康检查，基于初始扫描结果进行后续查询；3. 大型文件处理（CSV、JSON、XML），在代码中解析和过滤；4. 数据库查询结果聚合及后续查询；5. 基于初始结果进行条件逻辑的批量API操作。开发者可参考本教程的模式，将PTC集成到自己的代理工作流中，实现高效的数据处理和工具编排。

总结

内容来源与验证

内容等级：PREMIUM_A
质量等级：S_CANDIDATE
难度：中级
验证等级：V1
原文作者：PedramNavid
原文日期：2025-11-24
来源许可：MIT
来源版本：c47db6904f05023ecf0c51d3fcf3be27af782d2f
查看原文

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员