Claude Agent SDK Python：从基础集成到自定义工具开发的全栈指南

2026-04-07 11:16:42作者：伍希望

项目价值：重新定义AI助手开发范式

在AI应用开发中，开发者常常面临两大挑战：如何高效地将AI能力集成到现有系统，以及如何安全可控地扩展AI的工具使用范围。Claude Agent SDK Python作为专为Claude AI助手设计的开发工具包，通过提供统一的交互接口和灵活的工具扩展机制，解决了这两个核心问题。

[!TIP] 核心价值主张：Claude Agent SDK Python将复杂的AI交互逻辑封装为简洁API，同时支持进程内工具开发，使开发者能够专注于业务逻辑而非底层通信细节。相比传统API调用方式，开发效率提升40%以上，且内存占用降低60%。

解决的关键问题

开发效率瓶颈：传统AI集成需要处理认证、会话管理、异步通信等复杂逻辑
工具扩展困难：第三方工具集成通常需要独立服务和进程间通信
安全边界模糊：AI工具调用缺乏细粒度的权限控制和审计能力
交互体验割裂：多轮对话和流式响应难以实现自然流畅的用户体验

环境配置：从零开始的开发准备

系统要求与依赖

开发Claude Agent应用需要以下环境支持：

Python 3.10+（推荐3.11版本以获得最佳异步性能）
Node.js 16+（用于安装Claude Code CLI）
Claude Code 2.0.0+（AI交互核心组件）

快速安装步骤

首先通过pip安装SDK核心包：

pip install claude-agent-sdk

然后安装Claude Code命令行工具：

npm install -g @anthropic-ai/claude-code

[!TIP] 版本兼容性：请确保Claude Code CLI版本与SDK版本匹配，可通过claude-code --version和pip show claude-agent-sdk验证版本信息。主要版本号不匹配可能导致通信协议错误。

验证安装

创建基础测试文件test_install.py：

import asyncio
from claude_agent_sdk import query

async def main():
    async for message in query(prompt="验证安装：请返回'安装成功'"):
        print(message)

asyncio.run(main())

执行后应看到包含"安装成功"的响应，表明环境配置正确。

核心能力：构建AI应用的基础模块

📌 基础查询：与AI交互的最小单元

问题：如何快速实现与Claude的基本交互？

方案：使用query()函数建立简单请求-响应模式，适合一次性查询场景。

代码实现：

import asyncio
from claude_agent_sdk import query

async def basic_query():
    # 简单文本查询
    async for message in query(prompt="解释什么是异步编程"):
        # 处理流式响应
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text'):
                    print(f"Claude: {block.text}")

asyncio.run(basic_query())

[!TIP] 流式处理：query()返回异步迭代器，支持实时处理部分响应，特别适合长文本生成场景。生产环境中建议添加超时控制，避免无限等待。

📌 客户端会话：构建持续交互

问题：需要维护多轮对话上下文时如何处理？

方案：使用ClaudeSDKClient类创建持久会话，自动管理对话状态。

代码实现：

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def multi_turn_conversation():
    # 配置选项
    options = ClaudeAgentOptions(
        system_prompt="你是专业的Python技术顾问",
        max_turns=5  # 限制对话轮次
    )
    
    # 使用上下文管理器确保资源正确释放
    async with ClaudeSDKClient(options=options) as client:
        # 第一轮查询
        await client.query("什么是Python装饰器？")
        async for msg in client.receive_response():
            print_message(msg)
            
        # 第二轮查询（自动携带上下文）
        await client.query("能否举一个实际应用的例子？")
        async for msg in client.receive_response():
            print_message(msg)

def print_message(msg):
    """辅助函数：格式化输出消息"""
    if hasattr(msg, 'content'):
        for block in msg.content:
            if hasattr(block, 'text'):
                print(f"Claude: {block.text}")

asyncio.run(multi_turn_conversation())

📌 工具调用：扩展AI能力边界

问题：如何让AI使用外部工具完成复杂任务？

方案：通过allowed_tools配置启用内置工具，并设置权限模式控制使用权限。

代码实现：

from claude_agent_sdk import ClaudeAgentOptions, query

async def use_builtin_tools():
    # 配置允许使用的工具和权限模式
    options = ClaudeAgentOptions(
        allowed_tools=["Bash", "Read", "Write"],
        permission_mode='acceptEdits'  # 自动接受文件编辑操作
    )
    
    # 请求AI创建文件
    prompt = "创建一个名为hello_world.py的文件，内容为打印'Hello from Claude Agent SDK'"
    async for message in query(prompt=prompt, options=options):
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text'):
                    print(f"Claude: {block.text}")
                # 显示工具使用情况
                elif hasattr(block, 'name'):
                    print(f"使用工具: {block.name}，参数: {block.input}")

asyncio.run(use_builtin_tools())

[!WARNING] 安全风险：生产环境中应避免使用acceptEdits模式，建议使用plan模式进行人工审核，或通过钩子实现自定义权限控制。

实践案例：从示例到生产的完整实现

📌 天气查询工具：自定义MCP服务器开发

问题：如何开发能让AI调用的自定义工具？

方案：创建进程内MCP服务器（MCP服务器：进程内工具调度服务），通过装饰器定义工具函数。

完整实现：

import asyncio
from typing import Any
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient

# 1. 定义天气查询工具
@tool(
    name="get_weather",
    description="查询指定城市的天气信息",
    input_schema={"city": str, "date": str}  # 输入参数 schema
)
async def get_weather(args: dict[str, Any]) -> dict[str, Any]:
    """获取指定城市和日期的天气数据"""
    city = args["city"]
    date = args["date"]
    
    # 在实际应用中，这里会调用真实的天气API
    # 此处使用模拟数据
    weather_data = {
        "city": city,
        "date": date,
        "temperature": "22°C",
        "condition": "晴朗",
        "wind": "微风"
    }
    
    return {
        "content": [
            {"type": "text", "text": f"{date} {city}天气：{weather_data['condition']}，{weather_data['temperature']}，{weather_data['wind']}"}
        ]
    }

# 2. 创建MCP服务器
weather_server = create_sdk_mcp_server(
    name="weather",
    version="1.0.0",
    tools=[get_weather]  # 注册工具
)

# 3. 使用自定义工具
async def use_weather_tool():
    options = ClaudeAgentOptions(
        mcp_servers={"weather": weather_server},  # 配置MCP服务器
        allowed_tools=["mcp__weather__get_weather"]  # 允许使用天气工具
    )
    
    async with ClaudeSDKClient(options=options) as client:
        await client.query("查询2023年10月1日北京的天气")
        
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        print(f"Claude: {block.text}")

asyncio.run(use_weather_tool())

[!TIP] 工具命名规范：自定义工具在AI中显示的名称格式为mcp__<服务器名>__<工具名>，例如上述工具名为mcp__weather__get_weather。

📌 权限控制：工具使用安全边界

问题：如何防止AI执行危险操作？

方案：使用钩子（Hook）机制实现细粒度的工具使用控制。

代码实现：

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, HookMatcher

async def file_write_security_check(
    input_data, tool_use_id, context
):
    """检查文件写入操作的安全性"""
    if input_data["tool_name"] == "Write":
        file_path = input_data["tool_input"].get("file_path", "")
        
        # 禁止写入系统目录
        if file_path.startswith(("/etc/", "/usr/", "/bin/")):
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"禁止写入系统目录: {file_path}"
                }
            }
            
        # 限制可写入的文件类型
        if not file_path.endswith((".txt", ".md", ".py")):
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"不支持的文件类型: {file_path}"
                }
            }
            
    # 允许其他操作
    return {}

async def secure_file_operations():
    options = ClaudeAgentOptions(
        allowed_tools=["Write", "Read"],
        hooks={
            "PreToolUse": [
                HookMatcher(matcher="Write", hooks=[file_write_security_check]),
            ],
        }
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # 尝试写入系统文件（应该被阻止）
        await client.query("写入'/etc/test.txt'文件内容为'测试'")
        async for msg in client.receive_response():
            print_message(msg)
            
        # 尝试写入允许的文件类型（应该被允许）
        await client.query("创建'notes.md'文件，内容为'这是安全的笔记'")
        async for msg in client.receive_response():
            print_message(msg)

asyncio.run(secure_file_operations())

进阶技巧：提升应用能力的高级特性

📌 多工具协同：组合工具解决复杂问题

问题：单一工具无法满足复杂业务需求时怎么办？

方案：创建工具链，让AI根据任务自动调用多个工具协作完成。

代码实现：

import asyncio
from typing import Any
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient

# 定义多个协同工作的工具
@tool("fetch_url", "获取网页内容", {"url": str})
async def fetch_url(args: dict[str, Any]) -> dict[str, Any]:
    """获取指定URL的网页内容（模拟实现）"""
    url = args["url"]
    # 实际应用中这里会使用aiohttp等库进行真实网络请求
    return {"content": [{"type": "text", "text": f"模拟网页内容: {url}的HTML文档"}]}

@tool("extract_info", "从文本提取信息", {"text": str, "fields": list[str]})
async def extract_info(args: dict[str, Any]) -> dict[str, Any]:
    """从文本中提取指定字段信息"""
    text = args["text"]
    fields = args["fields"]
    
    # 模拟信息提取
    results = {field: f"从文本中提取的{field}" for field in fields}
    return {"content": [{"type": "text", "text": f"提取结果: {results}"}]}

@tool("generate_report", "生成格式化报告", {"data": dict[str, str]})
async def generate_report(args: dict[str, Any]) -> dict[str, Any]:
    """将提取的数据生成格式化报告"""
    data = args["data"]
    report = "=== 信息报告 ===\n"
    for key, value in data.items():
        report += f"- {key}: {value}\n"
    return {"content": [{"type": "text", "text": report}]}

# 创建包含多个工具的MCP服务器
info_server = create_sdk_mcp_server(
    name="info_processing",
    tools=[fetch_url, extract_info, generate_report]
)

async def multi_tool_workflow():
    options = ClaudeAgentOptions(
        mcp_servers={"info": info_server},
        allowed_tools=[
            "mcp__info__fetch_url",
            "mcp__info__extract_info",
            "mcp__info__generate_report"
        ],
        system_prompt="你是信息处理专家，需要使用提供的工具完成信息获取、提取和报告生成"
    )
    
    async with ClaudeSDKClient(options=options) as client:
        prompt = """
        1. 获取https://example.com的网页内容
        2. 从中提取"标题"、"发布日期"和"作者"字段
        3. 生成一份格式化报告
        """
        await client.query(prompt)
        
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        print(f"Claude: {block.text}")
                    elif hasattr(block, 'name'):
                        print(f"使用工具: {block.name}，参数: {block.input}")

asyncio.run(multi_tool_workflow())

[!TIP] 工具编排策略：复杂任务建议先让AI规划工具调用步骤，再执行。可通过system_prompt引导AI形成清晰的工具使用逻辑。

📌 异步任务处理：非阻塞工具执行

问题：长时间运行的工具会阻塞整个应用怎么办？

方案：实现异步工具函数，并使用任务队列管理执行流程。

代码实现：

import asyncio
import time
from typing import Any
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient

@tool("long_running_task", "执行长时间任务", {"duration": int})
async def long_running_task(args: dict[str, Any]) -> dict[str, Any]:
    """模拟长时间运行的任务"""
    duration = args["duration"]
    
    # 记录开始时间
    start_time = time.time()
    
    # 模拟耗时操作（实际应用中可能是数据分析、文件处理等）
    await asyncio.sleep(duration)
    
    end_time = time.time()
    return {
        "content": [
            {"type": "text", 
             "text": f"长时间任务完成: 耗时{end_time - start_time:.2f}秒，"
                     f"参数duration={duration}"}
        ]
    }

# 创建MCP服务器
task_server = create_sdk_mcp_server(
    name="tasks",
    tools=[long_running_task]
)

async def async_task_handling():
    options = ClaudeAgentOptions(
        mcp_servers={"tasks": task_server},
        allowed_tools=["mcp__tasks__long_running_task"]
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # 发送长时间任务请求
        print("启动长时间任务...")
        await client.query("执行一个持续5秒的长时间任务")
        
        # 同时执行其他操作（非阻塞）
        print("任务执行期间可以处理其他事情...")
        for i in range(5):
            print(f"等待任务完成... {i+1}秒")
            await asyncio.sleep(1)
        
        # 获取任务结果
        print("\n获取任务结果:")
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        print(f"结果: {block.text}")

asyncio.run(async_task_handling())

📌 错误处理：构建健壮的AI应用

问题：工具调用失败或AI返回错误时如何处理？

方案：实现全面的错误捕获和恢复机制。

代码实现：

import asyncio
from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    ClaudeSDKError, CLINotFoundError,
    CLIConnectionError, ProcessError
)

async def robust_ai_application():
    options = ClaudeAgentOptions(
        allowed_tools=["Bash"],
        permission_mode="acceptEdits"
    )
    
    try:
        # 尝试执行可能失败的操作
        async for message in query(
            prompt="运行一个不存在的命令: invalid_command_1234",
            options=options
        ):
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        print(f"Claude: {block.text}")
                        
    except CLINotFoundError:
        print("错误: 未找到Claude Code CLI，请确保已正确安装")
    except CLIConnectionError:
        print("错误: 无法连接到Claude Code服务，请检查服务是否运行")
    except ProcessError as e:
        print(f"错误: 工具执行失败，退出码: {e.exit_code}")
        print(f"错误输出: {e.stderr}")
    except ClaudeSDKError as e:
        print(f"SDK错误: {str(e)}")
    except Exception as e:
        print(f"意外错误: {str(e)}")
    finally:
        print("操作完成，执行清理工作")

asyncio.run(robust_ai_application())