Claude Agent SDK Python：从环境搭建到自定义工具的全流程指南

2026-04-07 12:16:07作者：裘旻烁

5分钟启动指南：如何快速验证你的AI助手开发环境？

当你拿到一个新的开发工具包时，最想知道的一定是"如何在最短时间内让它跑起来"。Claude Agent SDK Python提供了零侵入式的快速启动方案，让你在5分钟内完成从环境配置到首次AI交互的全过程。

🔧 环境准备三步法

首先确认你的开发环境是否满足要求：

Python 3.10或更高版本（推荐3.11以获得最佳异步性能）
Node.js运行环境（用于Claude Code后端支持）
网络连接（用于下载必要依赖）

💡 兼容性提示：Windows用户需要确保已安装Visual C++构建工具，Mac用户需安装Xcode命令行工具。

🔧 安装执行双命令

通过pip安装SDK核心包：

pip install claude-agent-sdk

然后安装Claude Code运行时：

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

🔧 验证环境是否就绪

创建一个简单的测试文件test_sdk.py：

# 场景说明：验证SDK基础通信能力
# 运行注意事项：确保网络通畅，首次运行会下载必要组件
import anyio
from claude_agent_sdk import query

async def main():
    # 发送一个简单查询，验证SDK是否正常工作
    async for message in query(prompt="请用一句话介绍Claude Agent SDK的核心功能"):
        print(f"AI响应: {message}")

if __name__ == "__main__":
    anyio.run(main)

运行这个脚本，如果看到AI返回的响应，恭喜你已经成功搭建了开发环境！

核心能力解析：如何用SDK构建智能交互系统？

为什么选择Claude Agent SDK而不是直接调用API？这个SDK究竟解决了哪些开发痛点？让我们通过实际场景来解析其核心价值。

基础应用：异步流式响应处理

想象你正在开发一个需要实时显示AI思考过程的应用，传统的请求-响应模式会让用户等待完整结果。而SDK的异步迭代器就像一条流水线上的包裹传送带，AI生成一段内容就立即传送给你，实现实时交互。

# 场景说明：实时处理AI的流式响应，适用于聊天应用或实时内容生成
# 运行注意事项：需要处理可能的网络中断和重连逻辑
import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def stream_chat_demo():
    # 创建客户端选项，配置系统提示
    options = ClaudeAgentOptions(
        system_prompt="你是一个帮助用户整理文件的助手，可以分析文件内容并生成摘要",
        max_turns=5  # 限制对话轮次，防止无限对话
    )
    
    # 使用上下文管理器确保资源正确释放
    async with ClaudeSDKClient(options=options) as client:
        # 发送用户查询
        await client.query("请分析当前目录下的README.md文件并生成3点核心内容")
        
        # 异步迭代获取流式响应
        async for message in client.receive_response():
            # 实时处理每一段响应
            print(f"AI正在思考: {message}", end="")
            
            # 实际应用中可在此处更新UI或进行其他处理

anyio.run(stream_chat_demo)

💡 性能提示：使用异步迭代器时，避免在迭代循环中执行阻塞操作，这会影响实时性。

进阶技巧：工具权限精细控制

企业环境中，如何确保AI只能访问授权的工具和资源？SDK的权限控制系统就像一个智能门禁，让你精确控制AI可以使用哪些工具，以及在什么条件下使用。

# 场景说明：限制AI只能使用特定工具并审核敏感操作
# 运行注意事项：权限逻辑应根据实际安全需求调整
from claude_agent_sdk import ClaudeAgentOptions, HookMatcher

async def file_security_check(input_data, tool_use_id, context):
    """检查文件操作是否符合安全策略"""
    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]
    
    # 只允许特定工具
    if tool_name not in ["Read", "Write"]:
        return {
            "hookSpecificOutput": {
                "permissionDecision": "deny",
                "permissionDecisionReason": f"工具 {tool_name} 未在授权列表中"
            }
        }
        
    # 检查Write操作是否涉及敏感目录
    if tool_name == "Write" and "secret" in tool_input.get("path", ""):
        return {
            "hookSpecificOutput": {
                "permissionDecision": "deny",
                "permissionDecisionReason": "禁止写入敏感目录"
            }
        }
        
    # 允许安全操作
    return {"hookSpecificOutput": {"permissionDecision": "allow"}}

# 配置选项，只允许特定工具并应用安全检查
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write"],
    permission_mode='review',  # 所有工具使用需审核
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="*", hooks=[file_security_check]),
        ],
    }
)

实战案例：构建文件分析助手

结合流式响应和工具控制，我们可以构建一个实用的文件分析助手，它能够读取指定文件并生成结构化摘要。

# 场景说明：智能文件分析助手，读取文件内容并生成结构化摘要
# 运行注意事项：确保程序有读取目标文件的权限
import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def file_analyzer(file_path):
    # 配置AI行为：专注于文件分析任务
    options = ClaudeAgentOptions(
        system_prompt="你是文件分析专家，需要读取文件内容并生成包含以下要素的结构化摘要："
                      "1. 文件用途 2. 核心功能 3. 使用示例 4. 注意事项",
        allowed_tools=["Read"],
        permission_mode='accept',  # 自动允许Read工具使用
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # 发送分析请求
        await client.query(f"请分析文件: {file_path}，并按要求生成结构化摘要")
        
        # 收集完整响应
        full_response = []
        async for message in client.receive_response():
            full_response.append(message)
            print(message, end="")
            
        return "".join(full_response)

# 分析项目中的示例文件
async def main():
    try:
        result = await file_analyzer("examples/quick_start.py")
        # 可以将结果保存到文件或进一步处理
        with open("analysis_result.txt", "w") as f:
            f.write(result)
        print("\n分析结果已保存到 analysis_result.txt")
    except Exception as e:
        print(f"分析过程出错: {str(e)}")

anyio.run(main)

工具生态开发：如何为AI助手打造专属能力扩展？

当内置工具无法满足特定业务需求时，SDK提供了低代码扩展能力，让你能够为AI助手开发专属工具。这种扩展就像给智能手机安装新APP，让AI具备原本没有的专业能力。

基础应用：创建你的第一个自定义工具

让我们从一个实用的"文件搜索工具"开始，这个工具能帮助AI在项目中快速定位包含特定内容的文件。

# 场景说明：创建文件内容搜索工具，帮助AI定位项目中的关键信息
# 运行注意事项：避免在包含大量文件的目录中执行深度搜索
import os
import re
from claude_agent_sdk import tool, create_sdk_mcp_server

# 使用@tool装饰器定义工具元数据
@tool(
    name="file_search",
    description="搜索指定目录下包含特定内容的文件",
    parameters={
        "directory": {"type": "string", "description": "要搜索的目录路径"},
        "pattern": {"type": "string", "description": "要搜索的文本模式"}
    }
)
async def search_files(args: dict) -> dict:
    """搜索目录下包含指定文本模式的文件"""
    results = []
    directory = args.get("directory", ".")
    pattern = args.get("pattern", "")
    
    # 验证输入参数
    if not pattern:
        return {"content": [{"type": "text", "text": "错误：搜索模式不能为空"}]}
        
    try:
        # 简单的文件搜索实现
        for root, _, files in os.walk(directory):
            for file in files:
                # 只处理文本文件
                if file.endswith(('.py', '.md', '.txt')):
                    file_path = os.path.join(root, file)
                    try:
                        with open(file_path, 'r', encoding='utf-8') as f:
                            content = f.read()
                            if re.search(pattern, content, re.IGNORECASE):
                                results.append(f"- {file_path}: 包含匹配内容")
                    except Exception as e:
                        results.append(f"- {file_path}: 无法读取文件 ({str(e)})")
        
        if results:
            return {"content": [{"type": "text", "text": "搜索结果:\n" + "\n".join(results)}]}
        else:
            return {"content": [{"type": "text", "text": "未找到匹配文件"}]}
            
    except Exception as e:
        return {"content": [{"type": "text", "text": f"搜索过程出错: {str(e)}"}]}

# 创建工具服务器
file_tools_server = create_sdk_mcp_server(
    name="file_tools",
    version="1.0.0",
    tools=[search_files]
)

进阶技巧：构建工具集合与类型安全

随着工具数量增加，我们需要更好的组织方式和类型安全保障。下面是一个完整的"项目助手工具集"示例。

# 场景说明：构建项目管理工具集合，包含文件搜索、代码分析和文档生成功能
# 运行注意事项：代码分析功能可能消耗较多系统资源
from typing import List, Dict, Any
from claude_agent_sdk import tool, create_sdk_mcp_server

# === 文件工具 ===
@tool(
    name="search",
    description="搜索项目中的文件内容",
    parameters={
        "pattern": {"type": "string", "description": "搜索模式"},
        "file_types": {"type": "array", "items": {"type": "string"}, "description": "文件类型列表，如['.py', '.md']"}
    }
)
async def search_project(args: Dict[str, Any]) -> Dict[str, Any]:
    # 实现搜索逻辑...
    return {"content": [{"type": "text", "text": "搜索结果..."}]}

# === 代码分析工具 ===
@tool(
    name="analyze_code",
    description="分析Python代码结构",
    parameters={
        "file_path": {"type": "string", "description": "Python文件路径"}
    }
)
async def analyze_python_code(args: Dict[str, Any]) -> Dict[str, Any]:
    # 实现代码分析逻辑...
    return {"content": [{"type": "text", "text": "代码分析结果..."}]}

# === 文档生成工具 ===
@tool(
    name="generate_doc",
    description="为Python函数生成文档字符串",
    parameters={
        "file_path": {"type": "string", "description": "Python文件路径"},
        "function_name": {"type": "string", "description": "函数名"}
    }
)
async def generate_docstring(args: Dict[str, Any]) -> Dict[str, Any]:
    # 实现文档生成逻辑...
    return {"content": [{"type": "text", "text": "生成的文档字符串..."}]}

# 创建工具服务器，整合所有工具
project_assistant_server = create_sdk_mcp_server(
    name="project_assistant",
    version="1.0.0",
    tools=[search_project, analyze_python_code, generate_docstring]
)

# 在客户端中使用工具集合
from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient

async def use_project_assistant():
    options = ClaudeAgentOptions(
        mcp_servers={"project": project_assistant_server},
        allowed_tools=[
            "mcp__project__search",
            "mcp__project__analyze_code",
            "mcp__project__generate_doc"
        ]
    )
    
    async with ClaudeSDKClient(options=options) as client:
        await client.query("请搜索项目中所有包含'create_sdk_mcp_server'的Python文件，"
                          "然后分析这些文件中该函数的使用方式，并为每个使用示例生成文档字符串")
        
        async for message in client.receive_response():
            print(message)

# anyio.run(use_project_assistant)

实战案例：构建智能日志分析工具

现在我们来构建一个更复杂的实战工具：智能日志分析器，它能够解析日志文件、识别错误模式并提供修复建议。

# 场景说明：智能日志分析工具，帮助开发人员快速定位和解决问题
# 运行注意事项：大型日志文件可能需要较长处理时间
import re
from datetime import datetime
from claude_agent_sdk import tool, create_sdk_mcp_server

# 错误模式数据库 - 实际应用中可从文件加载
ERROR_PATTERNS = {
    r"FileNotFoundError: (.*)": {
        "category": "文件系统错误",
        "solution": "检查文件路径是否正确，确保文件存在"
    },
    r"PermissionError: (.*)": {
        "category": "权限错误",
        "solution": "检查应用程序对目标资源的访问权限"
    },
    r"ImportError: (.*)": {
        "category": "依赖错误",
        "solution": "确保所有依赖包已安装，版本兼容"
    }
}

@tool(
    name="analyze_log",
    description="分析日志文件并提供错误诊断和解决方案",
    parameters={
        "log_path": {"type": "string", "description": "日志文件路径"},
        "time_range": {
            "type": "object",
            "properties": {
                "start": {"type": "string", "format": "date-time", "description": "开始时间"},
                "end": {"type": "string", "format": "date-time", "description": "结束时间"}
            },
            "description": "可选的时间范围过滤"
        }
    }
)
async def analyze_log_file(args: dict) -> dict:
    """分析日志文件，识别错误并提供解决方案"""
    log_path = args.get("log_path")
    time_range = args.get("time_range", {})
    
    if not log_path:
        return {"content": [{"type": "text", "text": "错误：日志文件路径不能为空"}]}
    
    try:
        # 解析时间范围
        start_time = datetime.fromisoformat(time_range["start"]) if "start" in time_range else None
        end_time = datetime.fromisoformat(time_range["end"]) if "end" in time_range else None
        
        # 读取日志文件
        with open(log_path, 'r', encoding='utf-8') as f:
            logs = f.readlines()
        
        # 分析日志
        error_count = 0
        error_types = {}
        solutions = []
        
        for line in logs:
            # 简单的时间戳解析（假设日志行以ISO时间戳开头）
            line_time = None
            if len(line) >= 20 and line[10] == 'T' and line[19] == 'Z':
                try:
                    line_time = datetime.fromisoformat(line[:19])
                except ValueError:
                    pass
            
            # 时间范围过滤
            if line_time and start_time and line_time < start_time:
                continue
            if line_time and end_time and line_time > end_time:
                continue
            
            # 错误模式匹配
            for pattern, info in ERROR_PATTERNS.items():
                match = re.search(pattern, line)
                if match:
                    error_count += 1
                    category = info["category"]
                    error_types[category] = error_types.get(category, 0) + 1
                    
                    # 记录解决方案
                    solutions.append({
                        "error": match.group(0),
                        "category": category,
                        "solution": info["solution"]
                    })
        
        # 生成分析报告
        report = []
        report.append(f"日志分析报告: {log_path}")
        report.append(f"总错误数: {error_count}")
        report.append("\n错误类型分布:")
        for category, count in error_types.items():
            report.append(f"- {category}: {count}次")
        
        if solutions:
            report.append("\n主要错误及解决方案:")
            # 去重并只显示前5个解决方案
            unique_solutions = {s["error"]: s for s in solutions}.values()
            for sol in list(unique_solutions)[:5]:
                report.append(f"- {sol['error']}")
                report.append(f"  解决方案: {sol['solution']}")
        
        return {"content": [{"type": "text", "text": "\n".join(report)}]}
        
    except Exception as e:
        return {"content": [{"type": "text", "text": f"分析日志时出错: {str(e)}"}]}

# 创建日志分析工具服务器
log_analyzer_server = create_sdk_mcp_server(
    name="log_analyzer",
    version="1.0.0",
    tools=[analyze_log_file]
)

常见陷阱：工具开发中的3个典型错误

在开发自定义工具时，即使经验丰富的开发者也可能遇到以下常见问题：

1. 参数验证缺失

问题描述：未对工具输入参数进行验证，导致处理无效输入时崩溃。

错误示例：

@tool("divide", "除法运算", {"a": float, "b": float})
async def divide_numbers(args: dict) -> dict:
    # 缺少对除数为零的检查
    result = args["a"] / args["b"]  # 当b=0时会抛出异常
    return {"content": [{"type": "text", "text": f"结果: {result}"}]}

修复方案：

@tool("divide", "除法运算", {"a": float, "b": float})
async def divide_numbers(args: dict) -> dict:
    a = args.get("a")
    b = args.get("b")
    
    # 参数验证
    if a is None or b is None:
        return {"content": [{"type": "text", "text": "错误：缺少参数a或b"}]}
    
    # 除数为零检查
    if b == 0:
        return {"content": [{"type": "text", "text": "错误：除数不能为零"}]}
        
    result = a / b
    return {"content": [{"type": "text", "text": f"结果: {result}"}]}

2. 同步阻塞操作

问题描述：在异步工具函数中执行长时间同步操作，阻塞事件循环。

错误示例：

@tool("slow_operation", "执行耗时操作", {"path": str})
async def slow_operation(args: dict) -> dict:
    # 同步读取大文件，阻塞事件循环
    with open(args["path"], "r") as f:
        data = f.read()  # 对于大文件会阻塞整个事件循环
    
    # 复杂计算也会阻塞
    result = process_large_data(data)  # 长时间运行的同步函数
    return {"content": [{"type": "text", "text": f"结果: {result}"}]}

修复方案：

import anyio
from anyio import to_thread

@tool("slow_operation", "执行耗时操作", {"path": str})
async def slow_operation(args: dict) -> dict:
    # 使用线程池运行阻塞操作
    async with anyio.create_task_group() as tg:
        # 在单独线程中读取文件
        data = await tg.start(
            to_thread.run_sync, 
            lambda: open(args["path"], "r").read()
        )
        
        # 在单独线程中处理数据
        result = await tg.start(
            to_thread.run_sync, 
            lambda: process_large_data(data)
        )
        
    return {"content": [{"type": "text", "text": f"结果: {result}"}]}

3. 安全漏洞

问题描述：工具函数中存在路径遍历漏洞，允许访问系统敏感文件。

错误示例：

@tool("read_file", "读取文件内容", {"path": str})
async def unsafe_read_file(args: dict) -> dict:
    path = args["path"]
    with open(path, "r") as f:  # 直接使用用户提供的路径
        content = f.read()
    return {"content": [{"type": "text", "text": content}]}

修复方案：

import os

@tool("read_file", "读取文件内容", {"path": str})
async def safe_read_file(args: dict) -> dict:
    base_dir = os.path.abspath("allowed_files/")  # 限制访问目录
    user_path = args.get("path", "")
    
    # 解析用户提供的路径
    resolved_path = os.path.abspath(
        os.path.join(base_dir, user_path)
    )
    
    # 检查路径是否在允许的目录内
    if not resolved_path.startswith(base_dir):
        return {"content": [{"type": "text", "text": "错误：访问被拒绝，路径超出允许范围"}]}
        
    try:
        with open(resolved_path, "r") as f:
            content = f.read(1024 * 10)  # 限制读取大小
        return {"content": [{"type": "text", "text": content}]}
    except Exception as e:
        return {"content": [{"type": "text", "text": f"读取文件失败: {str(e)}"}]}