Claude Agent SDK 实战指南：打造可落地的 AI 智能体

你是否厌倦了仅仅"聊天"的 AI 助手？是否梦想着让 AI 拥有类似人类程序员的能力，能够自己看文件、跑代码、执行复杂任务？Anthropic 发布的 Claude Agent SDK 正是帮你实现这一目标的强大工具！

这款 SDK（Software Development Kit）是 Anthropic 官方推出的 Python/TypeScript 工具包，用于与 Claude Agent（原 Claude Code）进行深度交互。它将 Claude Agent 背后强大的智能体框架能力，包括上下文管理、工具调用、错误处理等，带入了你的 Python 应用程序中。

今天，我们就来手把手学习如何使用 Claude Agent SDK，并深入了解它那些让你事半功倍的开发细节！

一、快速上手：安装与环境准备

要开始使用 Claude Agent SDK 强大的智能体功能，准备工作非常简单：

1. 基础要求

使用 Python 版本的 Claude Agent SDK 需要满足以下先决条件：

Python 3.10+
Node.js
Claude Code CLI (2.0.0+): 需要通过 npm install -g @anthropic-ai/claude-code 进行安装。

2. 安装 Claude Agent SDK

使用 pip 即可安装官方的 Python 工具包（注意：原 claude-code-sdk 已更名为 claude-agent-sdk）：

# 安装 Claude Agent SDK Python 包
pip install claude-agent-sdk

# 设置 API 密钥（必须）
export ANTHROPIC_API_KEY="your-key-here"

二、使用教程：两种核心交互模式与代码细节

Claude Agent SDK 提供了两种主要的交互方式，以适应不同的应用场景：query() 函数（单次任务）和 ClaudeSDKClient 类（持续对话）。

1. 模式一：单次交互与流式输出 (`query()`)

query() 函数是开始使用 Claude Agent SDK 最简单快捷的方式。它为每一次交互创建新的会话，没有历史记忆，并返回一个异步迭代器，非常适合处理流式响应。

核心代码示例：基础查询

import asyncio
from claude_agent_sdk import query, AssistantMessage, TextBlock

async def basic_query_example():
    # 使用 query() 发送一个单次问题
    async for message in query(prompt="你好，请问你是谁？"):
        # 实时处理流式消息
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text, end="")
    print()

if __name__ == "__main__":
    asyncio.run(basic_query_example())

2. 模式二：持续对话与高级控制 (`ClaudeSDKClient`)

如果你需要构建一个能"记住"之前对话内容、支持中断、并且能使用自定义工具的智能体，那么你需要使用 ClaudeSDKClient。

ClaudeSDKClient 的关键特性：

会话连续性：在多次 query() 调用中保持对话上下文。
支持中断：可以在智能体执行中途停止任务（通过 client.interrupt()）。
自定义工具与钩子（Hooks）：支持我们在应用中定义进程内工具和钩子，这是 query() 模式所不具备的。

核心代码示例：持续对话

import asyncio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

async def continuous_conversation_example():
    # 使用上下文管理器自动管理连接和断开
    async with ClaudeSDKClient() as client:
        # 第一次查询
        await client.query("告诉我法国的首都是哪里？")
        # 处理响应
        async for message in client.receive_response():
            # ... 打印消息逻辑 ...
            pass

        # 第二次查询，Claude Agent SDK 会记住之前的上下文
        await client.query("那个城市的人口是多少？")
        # 处理后续响应
        async for message in client.receive_response():
            # ... 打印消息逻辑 ...
            pass

if __name__ == "__main__":
    asyncio.run(continuous_conversation_example())

三、 Claude Agent SDK 深入开发细节：赋能智能体

Claude Agent SDK 的强大之处在于它允许开发者通过配置 ClaudeAgentOptions、自定义工具（MCP Server）和钩子（Hooks）来精细控制智能体的行为和安全。

1. 配置智能体行为：`ClaudeAgentOptions`

ClaudeAgentOptions 是配置 Claude Agent SDK 行为的核心数据类。你可以通过它来设置系统提示、允许使用的工具列表、权限模式和工作目录。

代码细节：权限与环境设置

from claude_agent_sdk import ClaudeAgentOptions, query
import asyncio

async def options_example():
    options = ClaudeAgentOptions(
        # 1. 设置智能体的角色和行为
        system_prompt="你是一名专业的 Python 开发者",
        # 2. 允许使用内置工具 (Read, Write, Bash)
        allowed_tools=["Read", "Write", "Bash"],
        # 3. 设置权限模式：'acceptEdits' 自动接受文件编辑，加速自动化
        permission_mode='acceptEdits',
        # 4. 设置当前工作目录（CWD），方便文件操作
        cwd="/home/user/my_agent_project"
    )

    async for message in query(
        prompt="创建一个名为 'server.py' 的 Python Web 服务器",
        options=options
    ):
        # ... 处理响应 ...
        pass

if __name__ == "__main__":
    asyncio.run(options_example())

2. 自定义工具：进程内 MCP 服务器（In-Process SDK MCP Servers）

Claude Agent SDK 最超赞的功能之一是能够通过 @tool 装饰器定义自定义 Python 函数，并在你的应用程序进程内运行它们，无需外部进程（MCP 服务器），极大地降低了 IPC（进程间通信）开销和部署复杂性。

代码细节：定义和集成自定义工具

定义工具：使用 @tool 装饰器定义工具名称、描述和输入参数模式。
创建 MCP 服务器：使用 create_sdk_mcp_server() 将工具函数打包成进程内服务器。
集成到 Options：将服务器配置传递给 ClaudeAgentOptions 的 mcp_servers 属性。

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

# 1. 定义自定义工具 'get_time'
@tool("get_time", "获取当前的系统时间", {})
async def get_time(args: dict[str, Any]) -> dict[str, Any]:
    """工具输入参数为空 {}"""
    from datetime import datetime
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return {
        "content": [{
            "type": "text",
            "text": f"当前时间：{current_time}"
        }]
    }

# 2. 创建 SDK MCP 服务器
time_server = create_sdk_mcp_server(
    name="system_utilities",
    version="1.0.0",
    tools=[get_time] # 传入自定义工具函数
)

async def custom_tool_session():
    # 3. 配置 Claude Agent Options
    options = ClaudeAgentOptions(
        # 映射服务器名（'utils'）到服务器配置
        mcp_servers={"utils": time_server},
        # 允许 Claude 使用我们定义的工具。名称格式为: mcp__<server_name>__<tool_name>
        allowed_tools=["mcp__utils__get_time"]
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("现在是几点几分？")
        # 智能体将识别并调用我们自定义的 get_time 工具
        async for message in client.receive_response():
            print(message)

if __name__ == "__main__":
    asyncio.run(custom_tool_session())

3. 智能体安全与控制：钩子（Hooks）

钩子（Hooks）是 Claude Agent SDK 中用于在智能体执行周期的特定点（例如，工具使用前或用户提示提交时）拦截和修改行为的确定性回调函数。这是确保安全性和增强可观察性的关键。

代码细节：在工具使用前进行安全检查 (PreToolUse)

我们可以定义一个钩子，用于在 Claude Agent SDK 尝试执行 Bash 命令前，阻止危险的命令（如 rm -rf /）。

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, HookMatcher, HookContext
import asyncio
from typing import Any

async def block_dangerous_bash(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """在 Bash 工具执行前进行验证"""
    tool_name = input_data.get('tool_name', '')
    if tool_name == "Bash":
        command = input_data['tool_input'].get('command', '')
        if 'rm -rf /' in command:
            print("[安全警告] 危险命令被阻止！")
            return {
                'hookSpecificOutput': {
                    'hookEventName': 'PreToolUse',
                    # 阻止操作
                    'permissionDecision': 'deny',
                    'permissionDecisionReason': '危险命令被阻止'
                }
            }
    return {}

async def hooks_example():
    options = ClaudeAgentOptions(
        allowed_tools=["Bash"],
        hooks={
            # 在工具使用前调用钩子
            'PreToolUse': [
                HookMatcher(matcher='Bash', hooks=[block_dangerous_bash]) # 仅匹配 Bash 工具
            ]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        # 尝试运行一个将被阻止的危险命令
        await client.query("使用 Bash 执行命令 rm -rf /")
        async for message in client.receive_response():
            # 智能体将被钩子阻止，并返回系统消息
            print(message)

if __name__ == "__main__":
    asyncio.run(hooks_example())

四、总结：为什么选择 Claude Agent SDK？

Claude Agent SDK 不仅仅是一个 API 包装器，它是一套用于构建真正自主 AI 系统的基础设施。它建立在为 Claude Code 提供支持的强大代理框架之上。

真正的自主性：Claude Agent SDK 实现了"给 AI 一台电脑"的核心理念，支持金融、客服、SRE 运维等多种复杂应用场景。
工程级就绪：它提供了生产环境所需的内置错误处理、会话管理、权限控制和自动上下文压缩功能，让你无需担心上下文溢出或不可靠的执行。
灵活的工具生态系统：通过 Model Context Protocol (MCP) 和进程内 SDK MCP 服务器，你可以无缝地将自定义 Python 函数暴露给 Claude Agent 使用，实现低开销、高效率的工具调用。

如果你希望构建一个能够持久运行、安全可靠、并且能够通过工具集成到你的现有系统的 AI 智能体，那么 Claude Agent SDK 绝对是你的首选框架。快去尝试这些超赞的功能吧！