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（进程间通信）开销和部署复杂性。

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

1. 定义工具：使用 @tool 装饰器定义工具名称、描述和输入参数模式。
2. 创建 MCP 服务器：使用 create_sdk_mcp_server() 将工具函数打包成进程内服务器。
3. 集成到 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 绝对是你的首选框架。快去尝试这些超赞的功能吧！