Claude Agent SDK Python实战指南:从基础集成到自定义工具开发
一、核心价值:为什么选择Claude Agent SDK?
在AI应用开发中,你是否曾面临这些挑战:如何将大语言模型无缝集成到Python应用?如何安全地扩展AI的能力边界?如何构建可复用的AI工具组件?Claude Agent SDK Python正是为解决这些问题而生。
这个SDK就像为你的Python应用安装了一个"AI能力扩展槽",让你能够轻松接入Claude的强大能力,并通过标准化接口扩展自定义功能。它的核心价值体现在三个方面:
- 无缝集成:提供简洁API,几行代码即可实现与Claude的交互
- 安全可控:完善的权限控制机制,确保AI工具使用安全
- 灵活扩展:MCP服务器架构,让你轻松开发和集成自定义工具
无论你是想为应用添加智能问答功能,还是构建复杂的AI辅助开发系统,Claude Agent SDK都能提供坚实的技术基础。
开发小贴士
评估项目需求时,先明确AI功能在整体架构中的定位,避免过度设计。从核心场景入手,逐步扩展功能。
二、5分钟环境搭建:从零基础到可运行
系统准备清单
在开始前,请确保你的开发环境满足以下要求:
- Python 3.10或更高版本
- Node.js环境(用于运行Claude Code)
- Claude Code 2.0.0或更高版本
快速安装步骤
首先,通过pip安装SDK:
pip install claude-agent-sdk
然后安装Claude Code:
npm install -g @anthropic-ai/claude-code
⚠️ 注意:如果安装过程中遇到权限问题,在Linux/Mac系统上可尝试添加sudo前缀,Windows系统建议使用管理员模式运行命令提示符。
验证安装是否成功:
claude-code --version
如果一切顺利,你将看到Claude Code的版本信息。
开发小贴士
建议使用虚拟环境(如venv或conda)隔离项目依赖,避免不同项目间的包冲突。生产环境中,可考虑使用Docker容器化部署。
三、场景化入门:构建你的第一个AI交互应用
基础问答交互
让我们从一个简单的问答程序开始,体验Claude Agent SDK的基本用法:
import anyio
from claude_agent_sdk import query
async def main():
# 发送查询并处理响应流
async for message in query(prompt="解释一下什么是异步编程"):
# 实时打印AI的响应内容
print(message, end="")
# 运行异步函数
anyio.run(main)
用途说明:这个示例展示了最基本的AI交互流程,通过query函数发送问题并异步接收响应。
参数解读:
- prompt:要发送给Claude的问题或指令
- query函数返回一个异步迭代器,允许你实时处理AI的响应流
定制化交互体验
你可以通过ClaudeAgentOptions类来自定义AI的行为:
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
# 创建配置选项
options = ClaudeAgentOptions(
system_prompt="你是一位专业的Python开发者,回答应简洁且包含代码示例",
max_turns=3, # 限制对话轮次
temperature=0.7 # 控制回答的创造性,0-1之间
)
# 使用自定义选项进行查询
async for message in query(
prompt="如何使用Python实现单例模式?",
options=options
):
print(message, end="")
anyio.run(main)
用途说明:通过配置选项,你可以定制AI的角色、行为模式和交互限制,使其更符合特定场景需求。
开发小贴士
系统提示(system_prompt)是塑造AI行为的关键。清晰、具体的系统提示能显著提高AI响应的质量和相关性。建议在提示中明确AI的角色、专业领域和回答风格。
四、模块化进阶:核心功能深度解析
问题:如何实现持续对话?
解决方案:使用ClaudeSDKClient类
ClaudeSDKClient提供了比基础query函数更强大的功能,支持持续对话和状态管理:
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
async def chatbot():
# 创建客户端实例
options = ClaudeAgentOptions(
system_prompt="你是一个友好的聊天机器人,擅长解释技术概念"
)
# 使用上下文管理器确保资源正确释放
async with ClaudeSDKClient(options=options) as client:
while True:
# 获取用户输入
user_input = input("你: ")
if user_input.lower() in ["exit", "quit"]:
print("聊天结束,再见!")
break
# 发送查询
await client.query(user_input)
# 接收并打印响应
print("Claude: ", end="")
async for message in client.receive_response():
print(message, end="")
print("\n")
anyio.run(chatbot)
用途说明:这个示例实现了一个简单的聊天机器人,能够维持对话状态,进行多轮交互。
问题:如何安全地使用工具能力?
解决方案:工具权限控制机制
💡 图解说明:工具权限控制流程
用户请求 → SDK → 权限检查 → 工具执行 → 结果返回
↑ ↓
响应 权限决策
以下是一个实现工具权限控制的示例:
from claude_agent_sdk import ClaudeAgentOptions, HookMatcher, query
async def tool_permission_check(input_data, tool_use_id, context):
"""检查工具使用权限的钩子函数"""
tool_name = input_data["tool_name"]
tool_input = input_data["tool_input"]
# 拒绝危险的Bash命令
if tool_name == "Bash":
command = tool_input.get("command", "")
dangerous_commands = ["rm", "mv", "sudo", "chmod"]
for cmd in dangerous_commands:
if cmd in command:
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": f"禁止执行危险命令: {cmd}"
}
}
# 允许其他工具使用
return {"hookSpecificOutput": {"permissionDecision": "allow"}}
async def main():
options = ClaudeAgentOptions(
allowed_tools=["Bash", "Read"], # 允许使用的工具列表
permission_mode="manual", # 手动权限模式
hooks={
"PreToolUse": [
HookMatcher(matcher="*", hooks=[tool_permission_check]),
],
}
)
async for message in query(
prompt="帮我删除当前目录下的所有文件",
options=options
):
print(message)
anyio.run(main)
用途说明:这个示例实现了一个安全检查机制,防止AI执行危险的系统命令,保护系统安全。
开发小贴士
在生产环境中,始终对AI的工具使用设置严格的权限控制。采用最小权限原则,只授予必要的工具访问权限,并实现详细的审计日志。
五、实战案例:构建自定义计算器工具
需求分析
假设你需要为AI添加数学计算能力,让它能够执行复杂的数学运算。我们需要开发一个计算器工具集,包含基本的加减乘除运算以及一些高级数学函数。
设计思路
我们将创建一个MCP服务器(可以把它想象成工具插件的调度中心),注册多个计算工具,并通过Claude SDK将其集成到AI能力中。
MCP服务器架构的优势在于:
- 模块化设计,便于维护和扩展
- 进程内运行,性能高效
- 标准化接口,易于集成
实现步骤
- 定义工具函数
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient
from typing import Any
# 加法工具
@tool(
name="add",
description="Add two numbers",
parameters={"a": {"type": "number", "description": "First number"},
"b": {"type": "number", "description": "Second number"}}
)
async def add_numbers(args: dict[str, Any]) -> dict[str, Any]:
"""计算两个数的和"""
result = args["a"] + args["b"]
return {
"content": [{"type": "text", "text": f"计算结果: {args['a']} + {args['b']} = {result}"}]
}
# 减法工具
@tool(
name="subtract",
description="Subtract two numbers",
parameters={"a": {"type": "number", "description": "Minuend"},
"b": {"type": "number", "description": "Subtrahend"}}
)
async def subtract_numbers(args: dict[str, Any]) -> dict[str, Any]:
"""计算两个数的差"""
result = args["a"] - args["b"]
return {
"content": [{"type": "text", "text": f"计算结果: {args['a']} - {args['b']} = {result}"}]
}
# 乘法工具
@tool(
name="multiply",
description="Multiply two numbers",
parameters={"a": {"type": "number", "description": "First factor"},
"b": {"type": "number", "description": "Second factor"}}
)
async def multiply_numbers(args: dict[str, Any]) -> dict[str, Any]:
"""计算两个数的积"""
result = args["a"] * args["b"]
return {
"content": [{"type": "text", "text": f"计算结果: {args['a']} × {args['b']} = {result}"}]
}
# 除法工具
@tool(
name="divide",
description="Divide two numbers",
parameters={"a": {"type": "number", "description": "Dividend"},
"b": {"type": "number", "description": "Divisor"}}
)
async def divide_numbers(args: dict[str, Any]) -> dict[str, Any]:
"""计算两个数的商"""
if args["b"] == 0:
return {
"content": [{"type": "text", "text": "错误:除数不能为零"}]
}
result = args["a"] / args["b"]
return {
"content": [{"type": "text", "text": f"计算结果: {args['a']} ÷ {args['b']} = {result}"}]
}
- 创建MCP服务器
# 创建计算器MCP服务器
calculator_server = create_sdk_mcp_server(
name="calculator",
version="1.0.0",
tools=[add_numbers, subtract_numbers, multiply_numbers, divide_numbers]
)
- 集成到AI客户端
async def use_calculator():
# 配置客户端选项
options = ClaudeAgentOptions(
mcp_servers={"calc": calculator_server},
allowed_tools=[
"mcp__calc__add",
"mcp__calc__subtract",
"mcp__calc__multiply",
"mcp__calc__divide"
],
system_prompt="你是一个数学助手,当遇到计算问题时,使用提供的计算器工具进行精确计算"
)
# 使用计算器工具
async with ClaudeSDKClient(options=options) as client:
# 发送复杂计算请求
await client.query("计算 (25 + 37) × (18 - 6) ÷ 4 的结果")
# 处理响应
async for message in client.receive_response():
print(message, end="")
# 运行示例
anyio.run(use_calculator)
测试验证
运行上述代码,你应该能看到AI正确使用计算器工具进行分步计算,并返回最终结果。测试不同的数学表达式,确保所有工具都能正常工作,特别是边界情况(如除数为零)。
开发小贴士
开发自定义工具时,确保提供清晰的描述和参数定义。良好的文档不仅帮助AI正确使用工具,也便于其他开发者理解和扩展你的工具。
六、常见故障排查指南
| 症状 | 可能原因 | 解决方案 |
|---|---|---|
| 安装失败:找不到Claude Code | 未安装Node.js或Claude Code | 1. 确认Node.js已安装并添加到PATH 2. 运行 npm install -g @anthropic-ai/claude-code |
| 连接错误:无法连接到Claude | 网络问题或API密钥配置错误 | 1. 检查网络连接 2. 验证API密钥是否正确 3. 检查防火墙设置 |
| 工具调用失败:权限被拒绝 | 权限模式设置或钩子函数限制 | 1. 检查permission_mode配置 2. 检查PreToolUse钩子函数 3. 确认工具名称在allowed_tools列表中 |
| 响应格式错误:无法解析JSON | Claude Code版本不兼容 | 1. 升级Claude Code到最新版本 2. 检查SDK与Claude Code版本兼容性 |
| 性能问题:响应缓慢 | 网络延迟或复杂计算 | 1. 优化网络连接 2. 考虑使用本地缓存 3. 分解复杂查询为多个简单查询 |
| 内存泄漏:长时间运行后内存增长 | 未正确关闭客户端连接 | 1. 使用async with上下文管理器 2. 确保正确处理异常 3. 定期重启长时间运行的服务 |
开发小贴士
实现详细的日志记录,记录所有API调用、工具使用和错误信息。这将极大简化故障排查过程,特别是在生产环境中。
七、高级功能与最佳实践
流模式优化
流模式允许你实时处理AI的响应,提供更流畅的用户体验:
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
async def streaming_example():
options = ClaudeAgentOptions(
system_prompt="你是一位技术作家,擅长解释复杂概念"
)
async with ClaudeSDKClient(options=options) as client:
await client.query("详细解释Python中的协程概念,分点说明")
# 实时处理流式响应
response_buffer = []
async for message in client.receive_response():
# 处理每个响应块
response_buffer.append(message)
# 实时输出到控制台
print(message, end="", flush=True)
# 响应完成后可以进行后续处理
full_response = "".join(response_buffer)
print(f"\n\n响应长度: {len(full_response)} 字符")
anyio.run(streaming_example)
性能优化建议:
- 对于长时间运行的流,考虑实现心跳机制
- 大文本响应可分块处理,避免内存占用过高
- 实现背压机制,防止过快的响应淹没处理能力
安全最佳实践
- 输入验证:始终验证用户输入,防止注入攻击
- 权限最小化:仅授予AI完成任务所需的最小工具权限
- 敏感信息过滤:实现响应内容过滤,防止敏感信息泄露
- 审计日志:记录所有AI交互和工具使用情况
- 速率限制:实现API调用速率限制,防止滥用
混合MCP服务器配置
你可以同时使用进程内和外部MCP服务器,构建灵活的工具生态:
options = ClaudeAgentOptions(
mcp_servers={
# 进程内工具服务器
"internal_tools": calculator_server,
# 外部工具服务器
"external_tools": {
"type": "stdio",
"command": "path/to/external/tool/server"
}
},
allowed_tools=[
"mcp__internal_tools__add",
"mcp__external_tools__weather"
]
)
开发小贴士
设计工具时遵循单一职责原则,每个工具专注于完成一项特定任务。这不仅使工具更易于开发和测试,也让AI更容易理解和正确使用它们。
八、项目资源导航
学习路径
-
入门阶段
- 基础示例:examples/quick_start.py
- 核心概念:README.md
- 基础API文档:src/claude_agent_sdk/client.py
-
进阶阶段
- 工具开发指南:examples/mcp_calculator.py
- 钩子系统:examples/hooks.py
- 流模式:examples/streaming_mode.py
-
高级阶段
- 测试用例:tests/
- 内部实现:src/claude_agent_sdk/_internal/
- 错误处理:src/claude_agent_sdk/_errors.py
工具推荐
- 开发环境:VS Code + Python插件
- 测试工具:pytest (参见tests/目录)
- 代码格式化:black
- 类型检查:mypy
- 文档生成:pdoc
社区支持
- 问题跟踪:项目issue系统
- 代码贡献:CONTRIBUTING.md
- 版本历史:CHANGELOG.md
- 许可证信息:LICENSE
开发小贴士
定期查看CHANGELOG.md,了解最新功能和API变更。参与社区讨论,分享你的使用经验和问题解决方案。
通过本指南,你已经掌握了Claude Agent SDK Python的核心功能和最佳实践。无论是构建简单的AI交互应用,还是开发复杂的自定义工具,这些知识都将帮助你高效地实现目标。开始探索Claude Agent SDK的无限可能吧!
GLM-5智谱 AI 正式发布 GLM-5,旨在应对复杂系统工程和长时域智能体任务。Jinja00
GLM-5.1GLM-5.1是智谱迄今最智能的旗舰模型,也是目前全球最强的开源模型。GLM-5.1大大提高了代码能力,在完成长程任务方面提升尤为显著。和此前分钟级交互的模型不同,它能够在一次任务中独立、持续工作超过8小时,期间自主规划、执行、自我进化,最终交付完整的工程级成果。Jinja00
LongCat-AudioDiT-1BLongCat-AudioDiT 是一款基于扩散模型的文本转语音(TTS)模型,代表了当前该领域的最高水平(SOTA),它直接在波形潜空间中进行操作。00- QQwen3.5-397B-A17BQwen3.5 实现了重大飞跃,整合了多模态学习、架构效率、强化学习规模以及全球可访问性等方面的突破性进展,旨在为开发者和企业赋予前所未有的能力与效率。Jinja00
AtomGit城市坐标计划AtomGit 城市坐标计划开启!让开源有坐标,让城市有星火。致力于与城市合伙人共同构建并长期运营一个健康、活跃的本地开发者生态。01
CAP基于最终一致性的微服务分布式事务解决方案,也是一种采用 Outbox 模式的事件总线。C#00