第 3 章：工具调用
出处
来源：tool_use/ · 状态：已读
一句话
工具把模型从「会说」变成「能做」；真正的约束是 schema、choice、谁在调用、结果进不进上下文。 计算器是教具，代码里调工具（Programmatic Tool Calling，PTC）和工具搜索才是规模问题。
核心概念
最小工具：模型不算，工具算
tool_use/calculator_tool.ipynb 是入门环：定义 schema → 模型发 tool_use → 你执行 → tool_result 回去。评测是演示用的。真正要带走的是：算术、IO、副作用都不该发生在 token 里。
tool_choice 是控制面，不是开关
tool_use/tool_choice.ipynb：auto（模型决定）、tool（强制某一工具）、any（必须用工具但不能指定哪把）。强制抽取走 tool；开放对话走 auto；「这轮必须动手」走 any。和 Book 1 04-工具权限与中断 是同一类判断：权限不在散文里，在这一轮允许什么。
并行、Pydantic、客户端完整环
parallel_tools.ipynb：一次多工具降延迟。3.7 一代曾用batch_tool绕过限制；现在看官方并行语义，不要把 workaround 当永恒 API。tool_use_with_pydantic.ipynb：schema 从类型来。工具的输入输出都应该能被校验，散文描述只是给人看的。customer_service_agent.ipynb/threat_intel_enrichment_agent.ipynb：完整客户端环——多源工具、结构化报告、循环直到停。威胁情报那份强调：报告形状第 1 章就该钉死，工具只负责填。vision_with_tools.ipynb：看图和动手叠在一起。图是输入模态，工具是行动权；不要把「能看见」当成「能改」。
PTC：让代码调工具，不让每一跳都过模型
programmatic_tool_calling_ptc.ipynb 用模拟费用 API：get_team_members → 每人上百条带元数据的 get_expenses → 超标再查 get_custom_budget。传统 tool use 把全部行项目灌进上下文，再串行 round-trip。
PTC：模型在 code execution 里写代码调工具；大结果在沙箱里 grep / 聚合，只有摘要回模型。要点：
- 工具要加
allowed_callers；默认只许模型直接调。 - 可同时
["direct", "code_execution_20250825"]。 - 只对 安全、可重复执行 的工具开放 programmatic。
caller字段区分 direct / code；只有 direct 的结果会被模型看见。- 走 beta：
advanced-tool-use-2025-11-20，有状态工作流要带container_id。
示例里 PTC 的 API 调用次数可能更多（代码写得更结构化、步骤拆开），省的是 token 和延迟，不是请求数。第 10 章 agentic search 复现把 PTC 当成深搜的第一杠杆：50+ 次 fetch 若每页都进对话，窗口会先炸。
工具上千时，定义本身就是检索问题
tool_search_with_embeddings.ipynb：>20 把专用工具就开始吃窗口；~100 把以上前置全部定义不现实。做法：先只给 tool_search 这把元工具，用本地 embedding（示例 all-MiniLM-L6-v2）按自然语言搜，把命中的 schema 当 tool result 返回，模型立刻能调。声明可砍 90%+ 上下文。生产要缓存 embedding、换更大模型提准。tool_search_alternate_approaches.ipynb 未进 registry，对照搜索策略时去仓库看。
评测工具本身见 10-评测与成本 的 tool_evaluation/。
对照 recipe（默认折叠；机制片段摘自官方 notebook，全文在 GitHub）：
tool_use/calculator_tool.ipynb — 最小工具：模型不算，工具算
python
def process_tool_call(tool_name, tool_input):
if tool_name == "calculator":
return calculate(tool_input["expression"])
def chat_with_claude(user_message):
print(f"\n{'=' * 50}\nUser Message: {user_message}\n{'=' * 50}")
message = client.messages.create(
model=MODEL_NAME,
max_tokens=4096,
messages=[{"role": "user", "content": user_message}],
tools=tools,
)
print("\nInitial Response:")
print(f"Stop Reason: {message.stop_reason}")
print(f"Content: {message.content}")
if message.stop_reason == "tool_use":
tool_use = next(block for block in message.content if block.type == "tool_use")
tool_name = tool_use.name
tool_input = tool_use.input
print(f"\nTool Used: {tool_name}")
print(f"Tool Input: {tool_input}")
tool_result = process_tool_call(tool_name, tool_input)
print(f"Tool Result: {tool_result}")
response = client.messages.create(
model=MODEL_NAME,
max_tokens=4096,
messages=[
{"role": "user", "content": user_message},
{"role": "assistant", "content": message.content},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": tool_result,
}
],
},
],
tools=tools,
)
else:
response = message
final_response = next(
(block.text for block in response.content if hasattr(block, "text")),
None,
)
print(response.content)
print(f"\nFinal Response: {final_response}")
return final_responsetool_use/tool_choice.ipynb — 强制 / 自动 / 禁用不只是开关
python
def analyze_tweet_sentiment(query):
response = client.messages.create(
model=MODEL_NAME,
max_tokens=4096,
tools=tools,
tool_choice={"type": "auto"},
messages=[{"role": "user", "content": query}],
)
print(response)
def analyze_tweet_sentiment(query):
response = client.messages.create(
model=MODEL_NAME,
max_tokens=4096,
tools=tools,
tool_choice={"type": "tool", "name": "print_sentiment_scores"},
messages=[{"role": "user", "content": query}],
)
print(response)
def analyze_tweet_sentiment(query):
prompt = f"""
Analyze the sentiment in the following tweet:
<tweet>{query}</tweet>
"""
response = client.messages.create(
model=MODEL_NAME,
max_tokens=4096,
tools=tools,
tool_choice={"type": "auto"},
messages=[{"role": "user", "content": prompt}],
)
print(response)tool_use/parallel_tools.ipynb — 一次多工具，延迟和一致性
python
def make_query_and_print_result(messages, tools=None):
response = client.messages.create(
model=MODEL_NAME,
messages=messages,
max_tokens=1000,
tool_choice={"type": "auto"},
tools=tools or [weather_tool, time_tool],
)
for block in response.content:
match block.type:
case "text":
print(block.text)
case "tool_use":
print(f"Tool: {block.name}({block.input})")
case _:
raise ValueError(f"Unexpected block type: {block.type}")
return response
MESSAGES = [{"role": "user", "content": "What's the weather and time in San Francisco?"}]
response = make_query_and_print_result(MESSAGES)
import json
batch_tool = {
"name": "batch_tool",
"description": "Invoke multiple other tool calls simultaneously",
"input_schema": {
"type": "object",
"properties": {
"invocations": {
"type": "array",
"description": "The tool calls to invoke",
"items": {
"types": "object",
"properties": {
"name": {
"types": "string",
"description": "The name of the tool to invoke",
},
"arguments": {
"types": "string",
"description": "The arguments to the tool",
},
},
"required": ["name", "arguments"],
},
}
},
"required": ["invocations"],
},
}
def process_tool_with_maybe_batch(tool_name, tool_input):
if tool_name == "batch_tool":
results = []
for invocation in tool_input["invocations"]:
results.append(
process_tool_call(invocation["name"], json.loads(invocation["arguments"]))
)
return "\n".join(results)
else:
return process_tool_call(tool_name, tool_input)tool_use/tool_use_with_pydantic.ipynb — schema 从类型来，不从散文来
python
def chatbot_interaction(user_message):
print(f"\n{'=' * 50}\nUser Message: {user_message}\n{'=' * 50}")
messages = [{"role": "user", "content": user_message}]
message = client.messages.create(
model=MODEL_NAME, max_tokens=4096, tools=tools, messages=messages
)
print("\nInitial Response:")
print(f"Stop Reason: {message.stop_reason}")
print(f"Content: {message.content}")
if message.stop_reason == "tool_use":
tool_use = next(block for block in message.content if block.type == "tool_use")
tool_name = tool_use.name
tool_input = tool_use.input
print(f"\nTool Used: {tool_name}")
print(f"Tool Input: {tool_input}")
save_note_response = process_tool_call(tool_name, tool_input)
print(f"Tool Result: {save_note_response}")
response = client.messages.create(
model=MODEL_NAME,
max_tokens=4096,
messages=[
{"role": "user", "content": user_message},
{"role": "assistant", "content": message.content},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": str(save_note_response),
}
],
},
],
tools=tools,
)
else:
response = message
final_response = next(
(block.text for block in response.content if hasattr(block, "text")),
None,
)
print(response.content)
print(f"\nFinal Response: {final_response}")
return final_responsetool_use/programmatic_tool_calling_ptc.ipynb — 让模型写代码去调工具，省 round-trip 和上下文
python
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_countertool_use/tool_search_with_embeddings.ipynb — 工具上千时不能全塞进 prompt
python
def run_tool_search_conversation(user_message: str, max_turns: int = 5) -> None:
"""
Run a conversation with Claude using the tool search pattern.
Args:
user_message: The initial user message
max_turns: Maximum number of conversation turns
"""
print(f"\n{'=' * 80}")
print(f"USER: {user_message}")
print(f"{'=' * 80}\n")
# Initialize conversation with only tool_search available
messages = [{"role": "user", "content": user_message}]
for turn in range(max_turns):
print(f"\n--- Turn {turn + 1} ---")
# Call Claude with current message history
response = claude_client.messages.create(
model=MODEL,
max_tokens=1024,
tools=TOOL_LIBRARY + [TOOL_SEARCH_DEFINITION],
messages=messages,
# IMPORTANT: This beta header enables tool definitions in tool results
extra_headers={"anthropic-beta": "advanced-tool-use-2025-11-20"},
)
# Add assistant's response to messages
messages.append({"role": "assistant", "content": response.content})
# Check if we're done
if response.stop_reason == "end_turn":
print("\n✓ Conversation complete\n")
# Print final response
for block in response.content:
if block.type == "text":
print(f"ASSISTANT: {block.text}")
break
# Handle tool uses
if response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type == "text":
print(f"\nASSISTANT: {block.text}")
elif block.type == "tool_use":
tool_name = block.name
tool_input = block.input
tool_use_id = block.id
print(f"\n🔧 Tool invocation: {tool_name}")
print(f" Input: {json.dumps(tool_input, indent=2)}")
if tool_name == "tool_search":
# Handle tool search
query = tool_input["query"]
top_k = tool_input.get("top_k", 5)
# Get tool references
tool_references = handle_tool_search(query, top_k)
# Create tool result with tool_reference content blocks
tool_results.append(
{
"type": "tool_result",
"tool_use_id": tool_use_id,
"content": tool_references,
}
)
else:
# Execute the discovered tool with mock data
mock_result = mock_tool_execution(tool_name, tool_input)
# Print a preview of the result
if len(mock_result) > 150:
print(f" ✅ Mock result: {mock_result[:150]}...")
else:
print(f" ✅ Mock result: {mock_result}")
tool_results.append(
{
"type": "tool_result",
"tool_use_id": tool_use_id,
"content": mock_result,
}
)
# Add tool results to messages
if tool_results:
messages.append({"role": "user", "content": tool_results})
else:
print(f"\nUnexpected stop reason: {response.stop_reason}")
break
print(f"\n{'=' * 80}\n")
print("✓ Conversation loop implemented")tool_use/tool_search_alternate_approaches.ipynb — 未登记；搜索策略对照
python
def run_conversation(user_message: str, max_turns: int = 10):
"""Run a conversation with dynamic tool loading."""
print(f"\n{'=' * 60}")
print(f"USER: {user_message}")
print(f"{'=' * 60}\n")
messages = [{"role": "user", "content": user_message}]
# Start with ONLY describe_tool - no other tools in the request
active_tools = [DESCRIBE_TOOL]
loaded_tools = set() # Track which tools we've added
for turn in range(max_turns):
print(f"--- Turn {turn + 1} (tools in request: {len(active_tools)}) ---")
response = client.messages.create(
model=MODEL,
max_tokens=1024,
system=SYSTEM_PROMPT,
tools=active_tools,
messages=messages,
extra_headers={"anthropic-beta": "advanced-tool-use-2025-11-20"},
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason == "end_turn":
for block in response.content:
if block.type == "text":
print(f"\nASSISTANT: {block.text}")
break
# Process tool calls
tool_results = []
for block in response.content:
if block.type == "text" and block.text:
print(f"ASSISTANT: {block.text}")
elif block.type == "tool_use":
tool_name = block.name
tool_input = block.input
if tool_name == "describe_tool":
requested_tool = tool_input["tool_name"]
print(f"🔍 describe_tool({requested_tool})")
if requested_tool in TOOL_LIBRARY:
# Add tool to active_tools with defer_loading=True
# This is critical for prompt caching!
if requested_tool not in loaded_tools:
tool_def = {**TOOL_LIBRARY[requested_tool], "defer_loading": True}
active_tools.append(tool_def)
loaded_tools.add(requested_tool)
print(f" ✓ Added {requested_tool} to tools (defer_loading=True)")
# Return tool_reference so Claude can use it
tool_results.append(
{
"type": "tool_result",
"tool_use_id": block.id,
"content": [
{"type": "tool_reference", "tool_name": requested_tool}
],
}
)
else:
tool_results.append(
{
"type": "tool_result",
"tool_use_id": block.id,
"content": f"Tool '{requested_tool}' not found.",
}
)
else:
# Execute discovered tool
print(f"🔧 {tool_name}({json.dumps(tool_input)})")
result = execute_tool(tool_name, tool_input)
print(f" → {result}")
tool_results.append(
{
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
}
)
if tool_results:
messages.append({"role": "user", "content": tool_results})
print(f"\n{'=' * 60}\n")
print("✓ Conversation loop ready")tool_use/customer_service_agent.ipynb — 客户端工具的完整环
python
import json
def chatbot_interaction(user_message):
print(f"\n{'=' * 50}\nUser Message: {user_message}\n{'=' * 50}")
messages = [{"role": "user", "content": user_message}]
response = client.messages.create(
model=MODEL_NAME, max_tokens=4096, tools=tools, messages=messages
)
print("\nInitial Response:")
print(f"Stop Reason: {response.stop_reason}")
print(f"Content: {response.content}")
while response.stop_reason == "tool_use":
tool_use = next(block for block in response.content if block.type == "tool_use")
tool_name = tool_use.name
tool_input = tool_use.input
print(f"\nTool Used: {tool_name}")
print("Tool Input:")
print(json.dumps(tool_input, indent=2))
tool_result = process_tool_call(tool_name, tool_input)
print("\nTool Result:")
print(json.dumps(tool_result, indent=2))
messages = [
{"role": "user", "content": user_message},
{"role": "assistant", "content": response.content},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": str(tool_result),
}
],
},
]
response = client.messages.create(
model=MODEL_NAME, max_tokens=4096, tools=tools, messages=messages
)
print("\nResponse:")
print(f"Stop Reason: {response.stop_reason}")
print(f"Content: {response.content}")
final_response = next(
(block.text for block in response.content if hasattr(block, "text")),
None,
)
print(f"\nFinal Response: {final_response}")
return final_responsetool_use/threat_intel_enrichment_agent.ipynb — 多源工具 + 结构化报告
python
SYSTEM_PROMPT = """You are a senior threat intelligence analyst. When given an Indicator of Compromise (IOC), you systematically investigate it by:
1. Identifying the IOC type and querying the most relevant intelligence source first
2. Analyzing initial results to determine what follow-up queries would be valuable
3. Cross-referencing findings across multiple sources when related indicators are found
4. Mapping observed behaviors and malware families to MITRE ATT&CK techniques
5. Synthesizing all findings into a clear, evidence-based assessment
Always query multiple sources when possible. If an IP lookup reveals associated malware, look up the MITRE techniques for that malware. If a hash lookup reveals contacted domains or IPs, investigate those too. Build the fullest picture you can before forming your assessment.
State your confidence level (low/medium/high) and explain what evidence supports it."""
MAX_TURNS = 10 # cap the agent loop to prevent runaway costs
def process_tool_call(tool_name: str, tool_input: dict) -> str:
"""Route tool calls to the appropriate backend function."""
handlers = {
"lookup_ip_reputation": lambda inp: lookup_ip_reputation(inp["ip_address"]),
"lookup_file_hash": lambda inp: lookup_file_hash(inp["file_hash"], inp["hash_type"]),
"lookup_domain": lambda inp: lookup_domain(inp["domain"]),
"get_mitre_techniques": lambda inp: get_mitre_techniques(inp["query"]),
}
handler = handlers.get(tool_name)
if handler is None:
return json.dumps({"error": f"Unknown tool: {tool_name}"})
return json.dumps(handler(tool_input), indent=2)
def run_threat_intel_agent(ioc: str, ioc_type: str) -> tuple[str, list[dict]]:
"""
Run the threat intel enrichment agent on a single IOC.
Returns (analysis_text, list_of_tool_calls_made).
"""
user_message = (
f"Investigate this indicator of compromise and provide a threat assessment:\n"
f" IOC: {ioc}\n"
f" Type: {ioc_type}\n\n"
f"Query all relevant intelligence sources, cross-reference findings, "
f"map to MITRE ATT&CK where applicable, and provide your assessment with "
f"severity rating, confidence score, and recommended response actions."
)
messages = [{"role": "user", "content": user_message}]
tool_calls_made = []
for _turn in range(MAX_TURNS):
response = client.messages.create(
model=MODEL_NAME,
max_tokens=4096,
system=SYSTEM_PROMPT,
tools=tools,
messages=messages,
)
# If Claude is done reasoning, extract final text
if response.stop_reason == "end_turn":
final_text = next(
(block.text for block in response.content if hasattr(block, "text")),
"No analysis generated.",
)
return final_text, tool_calls_made
# If Claude wants to use tools, process all tool calls in this turn
if response.stop_reason == "tool_use":
# Add assistant's response (contains both text and tool_use blocks)
messages.append({"role": "assistant", "content": response.content})
# Process each tool call and collect results
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(f" -> Calling {block.name}({json.dumps(block.input)})")
tool_calls_made.append({"tool": block.name, "input": block.input})
result = process_tool_call(block.name, block.input)
tool_results.append(
{
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
}
)
# Send tool results back to Claude
messages.append({"role": "user", "content": tool_results})
else:
# Unexpected stop reason
return f"Agent stopped unexpectedly: {response.stop_reason}", tool_calls_made
# Hit turn limit without end_turn
return (
f"Agent reached max_turns limit ({MAX_TURNS}) without completing analysis. "
f"Consider raising MAX_TURNS or simplifying the investigation scope.",
tool_calls_made,
)
print("Agent loop ready.")tool_use/vision_with_tools.ipynb — 看图和动手叠在一起
python
def get_base64_encoded_image(image_path):
with open(image_path, "rb") as image_file:
binary_data = image_file.read()
base_64_encoded_data = base64.b64encode(binary_data)
base64_string = base_64_encoded_data.decode("utf-8")
return base64_string
message_list = [
{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": get_base64_encoded_image("../images/tool_use/nutrition_label.png"),
},
},
{
"type": "text",
"text": "Please print the nutrition information from this nutrition label image.",
},
],
}
]
response = client.messages.create(
model=MODEL_NAME, max_tokens=4096, messages=message_list, tools=[nutrition_tool]
)
if response.stop_reason == "tool_use":
last_content_block = response.content[-1]
if last_content_block.type == "tool_use":
tool_name = last_content_block.name
tool_inputs = last_content_block.input
print(f"=======Claude Wants To Call The {tool_name} Tool=======")
print(tool_inputs)
else:
print("No tool was called. This shouldn't happen!")关键洞察
工具层有三道闸：(1) 能不能看见这把工具（search / 前置列表），(2) 这一轮许不许用（tool_choice / allowed_callers），(3) 结果进不进模型上下文（direct vs PTC excluded）。只做 (1) 的系统，会在长任务里被自己的 tool result 撑死。
PTC 把「工具」和「上下文」拆开：执行发生在沙箱，模型只吃过滤后的信号。这是第 4 章 compaction 的亲戚——都在问「哪些 token 还值得占注意力」。
个人思考
第 1 章用 tool schema 钉输出；本章用同一机制钉行动。区别是副作用。开放 allowed_callers 给 code execution，等于把重复执行权交给沙箱——费用查询可以，转账不行。下一章处理工具环必然带来的窗口膨胀。