一文读懂 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）：

<span leaf=""># 安装 Claude Agent SDK Python 包<span leaf="">  
<span leaf="">pip install claude-agent-sdk<span leaf="">  
<span leaf="">  
<span leaf=""># 设置 API 密钥（必须）<span leaf="">  
<span leaf="">export ANTHROPIC_API_KEY="your-key-here"<span leaf="">

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

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

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

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

核心代码示例：基础查询

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

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

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

ClaudeSDKClient 的关键特性：

会话连续性

：在多次 query() 调用中保持对话上下文。
支持中断

：可以在智能体执行中途停止任务（通过 client.interrupt()）。
自定义工具与钩子（Hooks）

：支持我们在应用中定义进程内工具和钩子，这是 query() 模式所不具备的。

核心代码示例：持续对话

<span leaf="">import asyncio<span leaf="">  
<span leaf="">from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock<span leaf="">  
<span leaf="">  
<span leaf="">async def continuous_conversation_example():<span leaf="">  
<span leaf="">    # 使用上下文管理器自动管理连接和断开<span leaf="">  
<span leaf="">    async with ClaudeSDKClient() as client:<span leaf="">  
<span leaf="">        # 第一次查询<span leaf="">  
<span leaf="">        await client.query("告诉我法国的首都是哪里？")<span leaf="">  
<span leaf="">        # 处理响应<span leaf="">  
<span leaf="">        async for message in client.receive_response():<span leaf="">  
<span leaf="">            # ... 打印消息逻辑 ...<span leaf="">  
<span leaf="">            pass<span leaf="">  
<span leaf="">  
<span leaf="">        # 第二次查询，Claude Agent SDK 会记住之前的上下文<span leaf="">  
<span leaf="">        await client.query("那个城市的人口是多少？")<span leaf="">  
<span leaf="">        # 处理后续响应<span leaf="">  
<span leaf="">        async for message in client.receive_response():<span leaf="">  
<span leaf="">            # ... 打印消息逻辑 ...<span leaf="">  
<span leaf="">            pass<span leaf="">  
<span leaf="">  
<span leaf="">if __name__ == "__main__":<span leaf="">  
<span leaf="">    asyncio.run(continuous_conversation_example())<span leaf="">

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

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

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

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

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

<span leaf="">from claude_agent_sdk import ClaudeAgentOptions, query<span leaf="">  
<span leaf="">import asyncio<span leaf="">  
<span leaf="">  
<span leaf="">async def options_example():<span leaf="">  
<span leaf="">    options = ClaudeAgentOptions(<span leaf="">  
<span leaf="">        # 1. 设置智能体的角色和行为<span leaf="">  
<span leaf="">        system_prompt="你是一名专业的 Python 开发者",<span leaf="">  
<span leaf="">        # 2. 允许使用内置工具 (Read, Write, Bash)<span leaf="">  
<span leaf="">        allowed_tools=["Read", "Write", "Bash"],<span leaf="">  
<span leaf="">        # 3. 设置权限模式：'acceptEdits' 自动接受文件编辑，加速自动化<span leaf="">  
<span leaf="">        permission_mode='acceptEdits',<span leaf="">  
<span leaf="">        # 4. 设置当前工作目录（CWD），方便文件操作<span leaf="">  
<span leaf="">        cwd="/home/user/my_agent_project"<span leaf="">  
<span leaf="">    )<span leaf="">  
<span leaf="">  
<span leaf="">    async for message in query(<span leaf="">  
<span leaf="">        prompt="创建一个名为 'server.py' 的 Python Web 服务器",<span leaf="">  
<span leaf="">        options=options<span leaf="">  
<span leaf="">    ):<span leaf="">  
<span leaf="">        # ... 处理响应 ...<span leaf="">  
<span leaf="">        pass<span leaf="">  
<span leaf="">  
<span leaf="">if __name__ == "__main__":<span leaf="">  
<span leaf="">    asyncio.run(options_example())<span leaf="">

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

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

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

定义工具

：使用 @tool 装饰器定义工具名称、描述和输入参数模式。
创建 MCP 服务器

：使用 create_sdk_mcp_server() 将工具函数打包成进程内服务器。
集成到 Options

：将服务器配置传递给 ClaudeAgentOptions 的 mcp_servers 属性。

<span leaf="">from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient<span leaf="">  
<span leaf="">from typing import Any<span leaf="">  
<span leaf="">import asyncio<span leaf="">  
<span leaf="">  
<span leaf=""># 1. 定义自定义工具 'get_time'<span leaf="">  
<span leaf="">@tool("get_time", "获取当前的系统时间", {})<span leaf="">  
<span leaf="">async def get_time(args: dict[str, Any]) -&gt; dict[str, Any]:<span leaf="">  
<span leaf="">    """工具输入参数为空 {}"""<span leaf="">  
<span leaf="">    from datetime import datetime<span leaf="">  
<span leaf="">    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")<span leaf="">  
<span leaf="">    return {<span leaf="">  
<span leaf="">        "content": [{<span leaf="">  
<span leaf="">            "type": "text",<span leaf="">  
<span leaf="">            "text": f"当前时间：{current_time}"<span leaf="">  
<span leaf="">        }]<span leaf="">  
<span leaf="">    }<span leaf="">  
<span leaf="">  
<span leaf=""># 2. 创建 SDK MCP 服务器<span leaf="">  
<span leaf="">time_server = create_sdk_mcp_server(<span leaf="">  
<span leaf="">    name="system_utilities",<span leaf="">  
<span leaf="">    version="1.0.0",<span leaf="">  
<span leaf="">    tools=[get_time] # 传入自定义工具函数<span leaf="">  
<span leaf="">)<span leaf="">  
<span leaf="">  
<span leaf="">async def custom_tool_session():<span leaf="">  
<span leaf="">    # 3. 配置 Claude Agent Options<span leaf="">  
<span leaf="">    options = ClaudeAgentOptions(<span leaf="">  
<span leaf="">        # 映射服务器名（'utils'）到服务器配置<span leaf="">  
<span leaf="">        mcp_servers={"utils": time_server},<span leaf="">  
<span leaf="">        # 允许 Claude 使用我们定义的工具。名称格式为: mcp__&lt;server_name&gt;__&lt;tool_name&gt;<span leaf="">  
<span leaf="">        allowed_tools=["mcp__utils__get_time"]<span leaf="">  
<span leaf="">    )<span leaf="">  
<span leaf="">  
<span leaf="">    async with ClaudeSDKClient(options=options) as client:<span leaf="">  
<span leaf="">        await client.query("现在是几点几分？")<span leaf="">  
<span leaf="">        # 智能体将识别并调用我们自定义的 get_time 工具<span leaf="">  
<span leaf="">        async for message in client.receive_response():<span leaf="">  
<span leaf="">            print(message)<span leaf="">  
<span leaf="">  
<span leaf="">if __name__ == "__main__":<span leaf="">  
<span leaf="">    asyncio.run(custom_tool_session())<span leaf="">

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

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

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

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

<span leaf="">from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, HookMatcher, HookContext<span leaf="">  
<span leaf="">import asyncio<span leaf="">  
<span leaf="">from typing import Any<span leaf="">  
<span leaf="">  
<span leaf="">async def block_dangerous_bash(<span leaf="">  
<span leaf="">    input_data: dict[str, Any],<span leaf="">  
<span leaf="">    tool_use_id: str | None,<span leaf="">  
<span leaf="">    context: HookContext<span leaf="">  
<span leaf="">) -&gt; dict[str, Any]:<span leaf="">  
<span leaf="">    """在 Bash 工具执行前进行验证"""<span leaf="">  
<span leaf="">    tool_name = input_data.get('tool_name', '')<span leaf="">  
<span leaf="">    if tool_name == "Bash":<span leaf="">  
<span leaf="">        command = input_data['tool_input'].get('command', '')<span leaf="">  
<span leaf="">        if 'rm -rf /' in command:<span leaf="">  
<span leaf="">            print("[安全警告] 危险命令被阻止！")<span leaf="">  
<span leaf="">            return {<span leaf="">  
<span leaf="">                'hookSpecificOutput': {<span leaf="">  
<span leaf="">                    'hookEventName': 'PreToolUse',<span leaf="">  
<span leaf="">                    # 阻止操作<span leaf="">  
<span leaf="">                    'permissionDecision': 'deny',<span leaf="">  
<span leaf="">                    'permissionDecisionReason': '危险命令被阻止'<span leaf="">  
<span leaf="">                }<span leaf="">  
<span leaf="">            }<span leaf="">  
<span leaf="">    return {}<span leaf="">  
<span leaf="">  
<span leaf="">async def hooks_example():<span leaf="">  
<span leaf="">    options = ClaudeAgentOptions(<span leaf="">  
<span leaf="">        allowed_tools=["Bash"],<span leaf="">  
<span leaf="">        hooks={<span leaf="">  
<span leaf="">            # 在工具使用前调用钩子<span leaf="">  
<span leaf="">            'PreToolUse': [<span leaf="">  
<span leaf="">                HookMatcher(matcher='Bash', hooks=[block_dangerous_bash]) # 仅匹配 Bash 工具<span leaf="">  
<span leaf="">            ]<span leaf="">  
<span leaf="">        }<span leaf="">  
<span leaf="">    )<span leaf="">  
<span leaf="">  
<span leaf="">    async with ClaudeSDKClient(options=options) as client:<span leaf="">  
<span leaf="">        # 尝试运行一个将被阻止的危险命令<span leaf="">  
<span leaf="">        await client.query("使用 Bash 执行命令 rm -rf /")<span leaf="">  
<span leaf="">        async for message in client.receive_response():<span leaf="">  
<span leaf="">            # 智能体将被钩子阻止，并返回系统消息<span leaf="">  
<span leaf="">            print(message)<span leaf="">  
<span leaf="">  
<span leaf="">if __name__ == "__main__":<span leaf="">  
<span leaf="">    asyncio.run(hooks_example())<span leaf="">

4. 代理claude codeDEMO

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

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

文章目录

一文读懂 Claude Agent SDK：如何打造你自己的生产级 AI 智能体 --知识铺

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

1. 基础要求

2. 安装 Claude Agent SDK

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

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

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

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

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

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

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

4. 代理claude codeDEMO

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

See Also

最近文章

福利派送

分类

标签

友情链接

其它

文章目录

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

1. 基础要求

2. 安装 Claude Agent SDK

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

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

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

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

1. 配置智能体行为：ClaudeAgentOptions

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

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

4. 代理claude codeDEMO

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

See Also

最近文章

福利派送

分类

标签

友情链接

其它

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

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

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