你是否厌倦了仅仅“聊天”的 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())

4. 代理claude codeDEMO

#!/usr/bin/env python3  
# -*- coding: utf-8 -*-  
"""  
Claude Agent SDK Basic Test Script  
Test the simplest SDK call functionality  
"""  
  
import asyncio  
import os  
from claude_agent_sdk import query, AssistantMessage, TextBlock  
  
  
async def basic_query_example():  
    """Basic query example - test single interaction"""  
    print("Starting Claude Agent SDK test...")  
    print("=" * 60)  
  
    try:  
        # Send a simple test question  
        prompt = "Hello! Please introduce yourself in one sentence."  
        print(f"Prompt: {prompt}\n")  
        print("AI Response:")  
        print("-" * 60)  
  
        # Use query() to send single question and handle streaming response  
        async for message in query(prompt=prompt):  
            # Process streaming messages in real-time  
            if isinstance(message, AssistantMessage):  
                for block in message.content:  
                    if isinstance(block, TextBlock):  
                        print(block.text, end="", flush=True)  
  
        print("\n" + "-" * 60)  
        print("Test completed successfully!")  
  
    except Exception as e:  
        print(f"\nTest failed: {str(e)}")  
        print(f"Error type: {type(e).__name__}")  
        raise  
  
  
async def main():  
    """Main function - set environment variables and run test"""  
  
    # Set environment variables (using user-provided credentials)  
    os.environ["ANTHROPIC_API_KEY"] = "API KEY"  
    os.environ["ANTHROPIC_BASE_URL"] = "代理地址"  
  
    print("\nEnvironment Configuration:")  
    print(f"   API Key: {os.environ['ANTHROPIC_API_KEY'][:20]}...")  
    print(f"   Base URL: {os.environ['ANTHROPIC_BASE_URL']}\n")  
  
    # Run basic test  
    await basic_query_example()  
  
  
if __name__ == "__main__":  
    # Run async main function  
    asyncio.run(main())

四、总结：为什么选择 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 绝对是你的首选框架。快去尝试这些超赞的功能吧！