30行Python实现AI Agent工具调用：从Claude API到完整生命周期

在实际 AI 应用开发中Agent 的核心能力往往被包装在复杂的框架和 SDK 中导致很多开发者只知其名、不明其理。真正理解 Agent 如何调用工具、处理结果、维持会话比单纯学会使用某个框架更有价值。本文将围绕 Claude Platform 文档中提到的tool_use、tool_result和messages机制用约 30 行 Python 代码实现一个最小可运行的 Agent 核心逻辑帮助读者掌握工具调用的完整生命周期。这个简化版 Agent 将展示如何接收用户请求、解析出需要调用的工具、执行具体函数、返回工具结果并继续后续对话。虽然生产级 Agent 需要考虑上下文管理、错误处理、异步调用等复杂问题但这个最小实现能让你看清最本质的交互流程。1. 先理解 Agent 工具调用的三个关键环节Agent 的核心是能够根据当前对话上下文决定是否需要调用外部工具如计算器、搜索引擎、数据库查询等并将工具执行结果整合回对话流。这个流程涉及三个关键环节1.1 tool_useAgent 决定调用哪个工具当 Agent 判断需要工具协助时会在回复中插入tool_use块明确指定要调用的工具名称和传入参数。这相当于 Agent 的决策输出。{ type: tool_use, id: toolu_01, name: calculate, input: {expression: 2 3 * 4} }1.2 tool_result执行工具并返回结果应用程序收到tool_use后需要找到对应的工具函数执行然后将执行结果包装成tool_result格式返回给 Agent。这相当于工具执行反馈。{ type: tool_result, tool_use_id: toolu_01, content: 14 }1.3 messages维持完整的对话上下文所有的用户输入、Agent 回复、工具调用和工具结果都以消息形式保存在会话中。Agent 基于完整的消息历史来决定下一步行动这是维持连贯对话的关键。2. 准备最小化开发环境这个实现只需要 Python 3.7 和 requests 库不需要安装任何 AI 框架或 SDK。2.1 环境要求组件要求说明Python3.7需要支持 f-string 和类型提示requests2.25HTTP 客户端库API 密钥Claude API需要有效的 Anthropic API 密钥2.2 安装依赖pip install requests2.3 设置 API 密钥将你的 Claude API 密钥设置为环境变量export ANTHROPIC_API_KEYyour-api-key-here或者在代码中直接设置import os os.environ[ANTHROPIC_API_KEY] your-api-key-here3. 实现 30 行核心 Agent 逻辑下面是最小化 Agent 核心实现完整代码约 30 行import json import requests from typing import Dict, Any, List class MiniAgent: def __init__(self, api_key: str): self.api_key api_key self.base_url https://api.anthropic.com/v1/messages self.messages [] def add_message(self, role: str, content: str): self.messages.append({role: role, content: content}) def call_tool(self, tool_name: str, input_data: Dict) - str: tools { calculate: lambda x: str(eval(x[expression])), get_weather: lambda x: fSunny, {x.get(temperature, 25°C)}, search_web: lambda x: fResults for: {x.get(query, )} } return tools.get(tool_name, lambda x: Tool not found)(input_data) def send_request(self, tools: List[Dict]) - Dict: headers { x-api-key: self.api_key, anthropic-version: 2023-06-01, content-type: application/json } data { model: claude-3-haiku-20240307, max_tokens: 1000, messages: self.messages, tools: tools } response requests.post(self.base_url, headersheaders, jsondata) return response.json() def process(self, user_input: str, tools: List[Dict]) - str: self.add_message(user, user_input) response self.send_request(tools) if content in response: for block in response[content]: if block[type] text: return block[text] elif block[type] tool_use: tool_result self.call_tool(block[name], block[input]) self.add_message(assistant, json.dumps([block])) self.add_message(user, json.dumps([{ type: tool_result, tool_use_id: block[id], content: tool_result }])) return self.process(Continue, tools) return No response4. 关键代码详解理解每个组件的作用4.1 消息管理维持对话上下文messages列表保存完整的对话历史每条消息包含角色user/assistant和内容。这是 Agent 理解上下文的基础def add_message(self, role: str, content: str): self.messages.append({role: role, content: content})在工具调用场景中需要特殊处理消息格式Assistant 的消息可能包含tool_use块User 的消息可能包含tool_result块4.2 工具注册与执行扩展 Agent 能力call_tool方法维护一个工具字典将工具名映射到具体的执行函数def call_tool(self, tool_name: str, input_data: Dict) - str: tools { calculate: lambda x: str(eval(x[expression])), # 简单计算器 get_weather: lambda x: fSunny, {x.get(temperature, 25°C)}, search_web: lambda x: fResults for: {x.get(query, )} } return tools.get(tool_name, lambda x: Tool not found)(input_data)实际项目中这些工具函数可以替换为真实的 API 调用、数据库查询或复杂计算逻辑。4.3 API 请求封装与 Claude 模型交互send_request方法封装了与 Claude API 的通信细节def send_request(self, tools: List[Dict]) - Dict: headers { x-api-key: self.api_key, anthropic-version: 2023-06-01, # 指定 API 版本 content-type: application/json } data { model: claude-3-haiku-20240307, # 使用轻量级模型 max_tokens: 1000, # 限制响应长度 messages: self.messages, # 传入完整对话历史 tools: tools # 定义可用的工具列表 } response requests.post(self.base_url, headersheaders, jsondata) return response.json()关键参数说明model: 选择适合的 Claude 模型版本max_tokens: 控制响应长度避免过长回复tools: 定义 Agent 可以使用的工具列表4.4 核心处理逻辑工具调用循环process方法实现了完整的工具调用生命周期def process(self, user_input: str, tools: List[Dict]) - str: # 1. 添加用户消息到历史 self.add_message(user, user_input) # 2. 发送请求到 Claude API response self.send_request(tools) # 3. 处理响应内容 if content in response: for block in response[content]: if block[type] text: # 如果是普通文本回复直接返回 return block[text] elif block[type] tool_use: # 如果是工具调用请求 # 4. 执行工具函数 tool_result self.call_tool(block[name], block[input]) # 5. 将工具调用和结果添加到消息历史 self.add_message(assistant, json.dumps([block])) self.add_message(user, json.dumps([{ type: tool_result, tool_use_id: block[id], content: tool_result }])) # 6. 递归调用继续处理工具结果 return self.process(Continue, tools) return No response这个循环确保了Agent 决定调用工具 → 执行工具 → 返回结果 → Agent 基于结果继续思考的完整流程。5. 运行完整示例从数学计算到天气查询5.1 定义可用工具列表首先定义 Agent 可以使用的工具tools [ { name: calculate, description: Evaluate a mathematical expression, input_schema: { type: object, properties: { expression: {type: string, description: Math expression to evaluate} }, required: [expression] } }, { name: get_weather, description: Get current weather information, input_schema: { type: object, properties: { location: {type: string, description: City name}, temperature: {type: string, description: Temperature unit} }, required: [location] } } ]5.2 初始化并运行 Agent# 初始化 AgentAPI 密钥需要替换为实际值 agent MiniAgent(api_keyos.getenv(ANTHROPIC_API_KEY)) # 测试数学计算 result1 agent.process(Calculate 15 * 8 20, tools) print(fResult 1: {result1}) # 测试天气查询会调用工具 result2 agent.process(Whats the weather in Beijing?, tools) print(fResult 2: {result2}) # 查看完整的消息历史 print(Message history:) for msg in agent.messages: print(f{msg[role]}: {msg[content][:100]}...)5.3 预期输出示例Result 1: 15 * 8 20 140 Result 2: The weather in Beijing is sunny with a temperature of 25°C. Message history: user: Calculate 15 * 8 20 assistant: [{type: tool_use, id: toolu_01, name: calculate, input: {expression: 15 * 8 20}}] user: [{type: tool_result, tool_use_id: toolu_01, content: 140}] assistant: The result of 15 * 8 20 is 140. user: Whats the weather in Beijing? assistant: [{type: tool_use, id: toolu_02, name: get_weather, input: {location: Beijing}}] user: [{type: tool_result, tool_use_id: toolu_02, content: Sunny, 25°C}] assistant: The weather in Beijing is sunny with a temperature of 25°C.6. 常见问题排查与解决方案在实际运行过程中可能会遇到以下典型问题6.1 API 认证失败现象收到 401 错误或认证失败提示。排查步骤检查ANTHROPIC_API_KEY环境变量是否设置正确确认 API 密钥是否有调用 Messages API 的权限验证 API 密钥是否过期或被撤销解决方案# 验证 API 密钥格式以 sk-ant- 开头 api_key os.getenv(ANTHROPIC_API_KEY) if not api_key or not api_key.startswith(sk-ant-): print(Invalid API key format)6.2 工具调用格式错误现象Agent 返回了tool_use块但执行时报参数错误。排查步骤检查工具定义中的input_schema与实际调用参数是否匹配验证工具函数是否能处理传入的参数类型查看 Claude 响应中的tool_use块内容解决方案# 添加参数验证 def call_tool(self, tool_name: str, input_data: Dict) - str: try: tools { calculate: lambda x: str(eval(x.get(expression, 0))), # 其他工具... } return tools[tool_name](input_data) except KeyError: return fTool {tool_name} not found except Exception as e: return fTool error: {str(e)}6.3 消息历史过长导致 API 限制现象随着对话进行响应变慢或出现 token 超限错误。排查步骤监控messages列表的长度和内容大小检查每次 API 调用的 token 使用量确认模型上下文窗口限制解决方案# 实现消息历史截断策略 def truncate_messages(self, max_messages20): if len(self.messages) max_messages: # 保留最新的消息但确保工具调用相关的消息完整 self.messages self.messages[-max_messages:]6.4 工具执行结果格式错误现象工具返回结果后Agent 无法正确理解或处理。排查步骤检查tool_result的格式是否符合 Claude API 要求验证返回内容是否为字符串类型确认tool_use_id与对应的tool_use块匹配解决方案# 确保工具结果格式正确 tool_result_block { type: tool_result, tool_use_id: block[id], # 必须与 tool_use 的 id 匹配 content: str(tool_result) # 确保内容为字符串 }7. 生产环境最佳实践虽然这个 30 行实现展示了核心逻辑但生产环境需要考虑更多因素7.1 工具管理规范化不要使用简单的字典映射而是建立工具注册机制class ToolRegistry: def __init__(self): self._tools {} def register(self, name: str, description: str, func: callable, schema: Dict): self._tools[name] { function: func, description: description, schema: schema } def get_tool_definitions(self): return [{ name: name, description: info[description], input_schema: info[schema] } for name, info in self._tools.items()]7.2 错误处理与重试机制添加完善的异常处理def safe_call_tool(self, tool_name: str, input_data: Dict) - str: try: result self.call_tool(tool_name, input_data) return result except Exception as e: logger.error(fTool {tool_name} failed: {e}) return fError executing tool: {str(e)} def send_request_with_retry(self, tools: List[Dict], max_retries3): for attempt in range(max_retries): try: return self.send_request(tools) except requests.exceptions.RequestException as e: if attempt max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避7.3 会话状态管理实现会话持久化和恢复class SessionManager: def __init__(self, storage_path: str): self.storage_path storage_path def save_session(self, session_id: str, messages: List[Dict]): with open(f{self.storage_path}/{session_id}.json, w) as f: json.dump({messages: messages}, f) def load_session(self, session_id: str) - List[Dict]: try: with open(f{self.storage_path}/{session_id}.json, r) as f: return json.load(f)[messages] except FileNotFoundError: return []7.4 性能监控与限流添加基本的监控指标class MonitoringAgent(MiniAgent): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.metrics { api_calls: 0, tool_calls: 0, total_tokens: 0 } def send_request(self, tools: List[Dict]) - Dict: self.metrics[api_calls] 1 response super().send_request(tools) if usage in response: self.metrics[total_tokens] response[usage][total_tokens] return response8. 扩展方向与深入学习建议掌握这个核心逻辑后可以从以下几个方向深入8.1 工具生态扩展数据库工具添加 SQL 查询、数据统计等工具API 集成连接外部服务如日历、邮件、支付等文件操作实现文档处理、图片分析等能力自定义计算添加领域特定的计算和推理工具8.2 高级 Agent 模式多步推理让 Agent 能够规划复杂的多步任务工具组合支持多个工具的顺序或并行执行动态工具加载根据上下文动态注册和卸载工具工具学习基于使用反馈优化工具选择策略8.3 工程化改进异步处理使用 async/await 提高并发性能缓存机制对常用工具结果进行缓存流量控制实现 API 调用速率限制和配额管理可观测性添加完整的日志、指标和追踪这个 30 行实现虽然简单但包含了 Agent 工具调用的核心范式。理解这个基础后再学习 LangChain、AutoGPT 等框架时会更容易把握其设计理念和实现原理。实际项目中建议从这种最小可行实现开始逐步添加所需的功能和优化。