Claude API 工具调用实战
从零构建智能代理应用

01系统架构概览：理解工具调用的工作原理

在开始编写代码之前，我们需要深入理解 Claude API 工具调用（Tool Use）的底层机制。这不是简单的函数调用，而是一套精心设计的"人机协作"协议。

核心工作流程

Claude 的工具调用遵循以下消息轮次（turns）模式：

1. 开发者定义工具：向 API 传递工具的名称、描述和参数 Schema（JSON Schema 格式）
2. 用户发起请求：用户的自然语言问题进入对话
3. Claude 决策调用：模型判断是否需要调用工具，若需要则返回 tool_use 类型消息块
4. 宿主执行工具：你的代码实际执行工具函数，获取真实结果
5. 结果反馈模型：将工具结果以 tool_result 格式传回 Claude
6. 模型生成最终答复：Claude 综合工具结果，给出最终回答

三种智能代理架构模式

根据 Anthropic 官方研究，智能代理系统分为两大类：固定工作流（Workflows）和动态代理（Agents）。常用架构包括：

本教程将从最基础的单工具调用出发，逐步演进到多工具协同的完整代理系统。

02环境准备与项目初始化

我们将基于 Python 3.11+ 搭建开发环境。推荐使用虚拟环境隔离依赖。

# 创建项目目录
mkdir claude-agent-demo && cd claude-agent-demo

# 创建并激活虚拟环境（Python 3.11+）
python -m venv .venv
source .venv/bin/activate  # Linux/macOS
# .venv\Scripts\activate   # Windows

# 验证 Python 版本
python --version  # 应输出 Python 3.11.x 或更高

# 安装 Anthropic Python SDK
pip install anthropic

# 安装其他常用依赖
pip install python-dotenv httpx

ANTHROPIC_API_KEY=sk-ant-api03-xxxxxxxxxxxxxxxxxxxxxxxx

.env
.venv/
__pycache__/
*.pyc

import anthropic
from dotenv import load_dotenv

load_dotenv()

client = anthropic.Anthropic()

# 简单连通性测试
message = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=64,
    messages=[{"role": "user", "content": "回复 OK"}]
)

print(message.content[0].text)  # 应输出 "OK"
print("✅ 环境配置成功！")

03定义工具：Schema 设计与参数规范

工具定义是整个系统的基础。一个设计优良的工具描述能让 Claude 准确判断何时调用、传递什么参数。

工具定义的四要素

每个工具必须包含：

• name：工具的唯一标识符，使用 snake_case 命名
• description：详细描述工具的功能、适用场景和返回值格式——这是 Claude 决策的核心依据
• input_schema：JSON Schema 格式的参数定义，明确每个参数的类型、含义和是否必填

# 示例：定义一组天气查询 + 日历工具
TOOLS = [
    {
        "name": "get_weather",
        "description": (
            "获取指定城市的当前天气信息。"
            "返回温度（摄氏度）、天气状况、湿度和风速。"
            "仅支持国内城市，需要传入城市的中文名或拼音。"
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称，例如：北京、上海、guangzhou"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "温度单位，默认 celsius"
                }
            },
            "required": ["city"]
        }
    },
    {
        "name": "get_calendar_events",
        "description": (
            "查询用户日历中的事件安排。"
            "返回指定日期范围内的所有会议、任务和提醒。"
            "日期格式必须是 YYYY-MM-DD。"
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "start_date": {
                    "type": "string",
                    "description": "查询开始日期，格式 YYYY-MM-DD"
                },
                "end_date": {
                    "type": "string",
                    "description": "查询结束日期，格式 YYYY-MM-DD，不填默认同 start_date"
                },
                "category": {
                    "type": "string",
                    "enum": ["meeting", "task", "reminder", "all"],
                    "description": "事件类型过滤，默认 all"
                }
            },
            "required": ["start_date"]
        }
    }
]

04构建第一个工具调用代理

现在来编写核心代理逻辑。关键在于正确处理 Claude 返回的 stop_reason: "tool_use" 信号，并将工具结果以正确格式反馈回去。

import anthropic
import json
from dotenv import load_dotenv

load_dotenv()

client = anthropic.Anthropic()

# ── 模拟工具实现 ──────────────────────────────────────
def get_weather(city: str, unit: str = "celsius") -> dict:
    """模拟天气查询（实际项目中对接真实 API）"""
    mock_data = {
        "北京": {"temp": 8, "condition": "晴转多云", "humidity": 35, "wind": "北风 3级"},
        "上海": {"temp": 14, "condition": "阴天", "humidity": 72, "wind": "东南风 2级"},
    }
    city_data = mock_data.get(city, {"temp": 20, "condition": "未知", "humidity": 50, "wind": "无风"})
    temp = city_data["temp"]
    if unit == "fahrenheit":
        temp = temp * 9 / 5 + 32
    return {
        "city": city,
        "temperature": f"{temp}°{'F' if unit == 'fahrenheit' else 'C'}",
        "condition": city_data["condition"],
        "humidity": f"{city_data['humidity']}%",
        "wind": city_data["wind"]
    }

def get_calendar_events(start_date: str, end_date: str = None, category: str = "all") -> list:
    """模拟日历查询"""
    if end_date is None:
        end_date = start_date
    return [
        {"time": "09:30", "title": "产品评审会议", "type": "meeting", "date": start_date},
        {"time": "14:00", "title": "季度 OKR 对齐", "type": "meeting", "date": start_date},
        {"time": "17:00", "title": "提交周报", "type": "task", "date": start_date},
    ]

# ── 工具分发器 ────────────────────────────────────────
def execute_tool(tool_name: str, tool_input: dict) -> str:
    """根据工具名称路由到对应实现"""
    if tool_name == "get_weather":
        result = get_weather(**tool_input)
        return json.dumps(result, ensure_ascii=False)
    elif tool_name == "get_calendar_events":
        result = get_calendar_events(**tool_input)
        return json.dumps(result, ensure_ascii=False)
    else:
        return json.dumps({"error": f"未知工具: {tool_name}"})

# ── 核心代理循环 ──────────────────────────────────────
def run_agent(user_message: str, tools: list) -> str:
    """
    单轮工具调用代理。
    自动处理 tool_use 循环，直到 Claude 给出最终文本答复。
    """
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )

        print(f"[DEBUG] stop_reason={response.stop_reason}, "
              f"blocks={[b.type for b in response.content]}")

        # 情况 1：模型直接回答，无需工具
        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text

        # 情况 2：模型请求调用工具
        elif response.stop_reason == "tool_use":
            # 将 assistant 消息加入历史
            messages.append({"role": "assistant", "content": response.content})

            # 收集所有工具调用结果
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    print(f"[TOOL] 调用 {block.name}，参数: {block.input}")
                    result = execute_tool(block.name, block.input)
                    print(f"[TOOL] 结果: {result}")

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

            # 将工具结果以 user 角色传回
            messages.append({"role": "user", "content": tool_results})

        else:
            # 其他 stop_reason（max_tokens 等）
            break

    return "代理执行异常终止"


# ── 测试运行 ──────────────────────────────────────────
if __name__ == "__main__":
    from tools_definition import TOOLS

    query = "今天北京天气怎么样？另外我明天（2026-03-01）有什么安排？"
    print(f"\n用户问题：{query}\n")
    answer = run_agent(query, TOOLS)
    print(f"\n🤖 Claude 答复：\n{answer}")

05多工具协同与并行调用

Claude 支持在单次响应中同时调用多个工具（并行工具调用）。上面的代码已经正确处理了这种情况：tool_results 是一个列表，可以包含多个结果。

让我们用一个更复杂的示例来演示并行调用——构建一个"市场情报助手"，能同时查询多个数据源：

import anthropic
import json
import asyncio
from concurrent.futures import ThreadPoolExecutor
from dotenv import load_dotenv

load_dotenv()
client = anthropic.Anthropic()

# 扩展工具集
MARKET_TOOLS = [
    {
        "name": "get_stock_price",
        "description": "获取指定股票的实时价格和涨跌幅。支持 A 股（沪深）和港股代码。",
        "input_schema": {
            "type": "object",
            "properties": {
                "symbol": {"type": "string", "description": "股票代码，如 000001、600036、00700"},
                "market": {"type": "string", "enum": ["A股", "港股", "美股"]}
            },
            "required": ["symbol", "market"]
        }
    },
    {
        "name": "get_industry_news",
        "description": "搜索指定行业或公司的最新新闻，返回最近 10 条新闻摘要。",
        "input_schema": {
            "type": "object",
            "properties": {
                "keyword": {"type": "string", "description": "搜索关键词，如公司名、行业名"},
                "days": {"type": "integer", "description": "搜索最近 N 天，默认 7 天", "default": 7}
            },
            "required": ["keyword"]
        }
    },
    {
        "name": "get_financial_metrics",
        "description": "获取公司的核心财务指标，包括 PE、PB、ROE、营收增速等。",
        "input_schema": {
            "type": "object",
            "properties": {
                "company": {"type": "string", "description": "公司名称或股票代码"},
                "quarter": {"type": "string", "description": "财报季度，如 2025Q3，默认最新"}
            },
            "required": ["company"]
        }
    }
]

def execute_market_tool(name: str, input_data: dict) -> str:
    """市场工具分发（模拟实现）"""
    if name == "get_stock_price":
        # 实际应对接行情 API
        return json.dumps({"symbol": input_data["symbol"], "price": 42.50,
                           "change": "+2.3%", "volume": "1.2亿手"})
    elif name == "get_industry_news":
        return json.dumps([
            {"title": f"{input_data['keyword']}行业最新动态", "date": "2026-02-28",
             "summary": "行业整体呈现上升趋势..."},
            {"title": f"{input_data['keyword']}政策利好", "date": "2026-02-27",
             "summary": "监管层发布新规..."}
        ], ensure_ascii=False)
    elif name == "get_financial_metrics":
        return json.dumps({"company": input_data["company"], "PE": 18.5,
                           "PB": 2.1, "ROE": "15.3%", "revenue_growth": "+22.4%"})
    return json.dumps({"error": "未知工具"})


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

    while True:
        resp = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            tools=MARKET_TOOLS,
            messages=messages
        )

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

        if resp.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": resp.content})

            # ⚡ 并行执行多个工具调用
            tool_blocks = [b for b in resp.content if b.type == "tool_use"]
            print(f"[PARALLEL] 并行执行 {len(tool_blocks)} 个工具调用")

            with ThreadPoolExecutor(max_workers=len(tool_blocks)) as executor:
                futures = {
                    block.id: executor.submit(execute_market_tool, block.name, block.input)
                    for block in tool_blocks
                }

            tool_results = [
                {"type": "tool_result", "tool_use_id": tid, "content": f.result()}
                for tid, f in futures.items()
            ]
            messages.append({"role": "user", "content": tool_results})

    return "执行异常"


if __name__ == "__main__":
    answer = run_market_agent(
        "帮我分析一下腾讯（00700）的投资价值，包括最新股价、近期新闻和财务指标"
    )
    print(answer)

06对话管理与上下文维护

真实的代理应用需要维持多轮对话状态。以下是一个带有对话历史管理的会话管理器实现：

import anthropic
import json
from typing import Optional
from dotenv import load_dotenv

load_dotenv()

class AgentSession:
    """带状态管理的多轮对话代理"""

    def __init__(self, tools: list, system_prompt: str = None,
                 model: str = "claude-opus-4-6", max_history_turns: int = 20):
        self.client = anthropic.Anthropic()
        self.tools = tools
        self.model = model
        self.max_history_turns = max_history_turns
        self.messages: list = []
        self.system_prompt = system_prompt or (
            "你是一位专业的智能助手。当需要工具来完成任务时，"
            "主动调用相应工具获取信息。回答要简洁、准确、有帮助。"
        )

    def _trim_history(self):
        """限制历史长度，避免超出上下文窗口"""
        # 保留偶数条消息（user/assistant 成对）
        if len(self.messages) > self.max_history_turns * 2:
            # 保留最新的 N 轮
            keep = self.max_history_turns * 2
            self.messages = self.messages[-keep:]
            print(f"[SESSION] 历史裁剪：保留最近 {self.max_history_turns} 轮对话")

    def chat(self, user_input: str) -> str:
        """发送消息并处理工具调用，返回最终文本回复"""
        self.messages.append({"role": "user", "content": user_input})
        self._trim_history()

        while True:
            kwargs = {
                "model": self.model,
                "max_tokens": 4096,
                "tools": self.tools,
                "messages": self.messages,
            }
            if self.system_prompt:
                kwargs["system"] = self.system_prompt

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

            if response.stop_reason == "end_turn":
                text = next((b.text for b in response.content if hasattr(b, "text")), "")
                self.messages.append({"role": "assistant", "content": response.content})
                return text

            elif response.stop_reason == "tool_use":
                self.messages.append({"role": "assistant", "content": response.content})

                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        result = execute_tool(block.name, block.input)  # 复用上面的分发器
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result
                        })

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

            else:
                return f"意外的 stop_reason: {response.stop_reason}"

    def clear_history(self):
        """清空对话历史（开启新话题时调用）"""
        self.messages = []

    def get_history_summary(self) -> str:
        """返回对话历史摘要"""
        turns = len([m for m in self.messages if m["role"] == "user"])
        return f"当前会话：{turns} 轮对话，{len(self.messages)} 条消息"


# 交互式演示
if __name__ == "__main__":
    from tools_definition import TOOLS

    session = AgentSession(tools=TOOLS)
    print("智能助手已启动，输入 exit 退出，输入 clear 清空历史\n")

    while True:
        user_input = input("你：").strip()
        if user_input.lower() == "exit":
            break
        elif user_input.lower() == "clear":
            session.clear_history()
            print("✅ 对话历史已清空\n")
            continue
        elif not user_input:
            continue

        response = session.chat(user_input)
        print(f"\n助手：{response}\n")
        print(f"[{session.get_history_summary()}]\n")

07生产环境最佳实践与常见问题

将工具调用代理推向生产环境前，需要重点关注以下几个维度：

错误处理策略

import anthropic
from anthropic import APIConnectionError, RateLimitError, APIStatusError
import time

def robust_agent_call(messages: list, tools: list, max_retries: int = 3) -> str:
    client = anthropic.Anthropic()
    retries = 0

    while retries < max_retries:
        try:
            response = client.messages.create(
                model="claude-opus-4-6",
                max_tokens=4096,
                tools=tools,
                messages=messages
            )
            return response

        except RateLimitError as e:
            # 速率限制：指数退避重试
            wait_time = 2 ** retries
            print(f"[RATE_LIMIT] 触发限流，{wait_time}s 后重试...")
            time.sleep(wait_time)
            retries += 1

        except APIConnectionError as e:
            # 网络连接错误
            print(f"[CONN_ERROR] 连接失败: {e}")
            if retries < max_retries - 1:
                time.sleep(1)
                retries += 1
            else:
                raise

        except APIStatusError as e:
            # API 返回错误状态码
            if e.status_code == 529:  # Overloaded
                time.sleep(5)
                retries += 1
            else:
                print(f"[API_ERROR] {e.status_code}: {e.message}")
                raise

    raise Exception(f"重试 {max_retries} 次后仍然失败")

常见问题速查表

性能与成本优化建议

• 选择合适的模型：复杂多步推理用 claude-opus-4-6，简单任务用 claude-haiku-4-5-20251001，成本相差 5-10 倍
• 精简工具集：每次 API 调用只传入当前场景需要的工具，减少 token 消耗
• 缓存工具结果：对于短期内相同参数的查询，优先返回缓存值
• 流式输出（Streaming）：使用 client.messages.stream() 改善用户体验，减少等待感
• 设置合理的 max_tokens：根据任务预期输出长度设置，避免无谓的 token 浪费