Python

前提条件

Python 3.10+
来自 PyPI 的 claude-code-sdk
Node.js 18+
来自 NPM 的 @anthropic-ai/claude-code

要查看 Python SDK 源代码，请参阅 claude-code-sdk 仓库。

对于交互式开发，请使用 IPython：pip install ipython

安装

从 PyPI 安装 claude-code-sdk 并从 NPM 安装 @anthropic-ai/claude-code：

pip install claude-code-sdk
npm install -g @anthropic-ai/claude-code  # 必需的依赖项

（可选）为交互式开发安装 IPython：

pip install ipython

快速开始

创建您的第一个代理：

# legal-agent.py
import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def main():
    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            system_prompt="您是一名法律助理。识别风险并提出改进建议。",
            max_turns=2
        )
    ) as client:
        # 发送查询
        await client.query(
            "审查此合同条款的潜在问题：'当事方同意承担无限责任...'"
        )

        # 流式传输响应
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                # 在内容到达时打印流式内容
                for block in message.content:
                    if hasattr(block, 'text'):
                        print(block.text, end='', flush=True)

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

将上述代码保存为 legal-agent.py，然后运行：

python legal-agent.py

对于 IPython/Jupyter notebooks，您可以直接在单元格中运行代码：

await main()

此页面上的 Python 示例使用 asyncio，但您也可以使用 anyio。

基本用法

Python SDK 提供两个主要接口：

1. `ClaudeSDKClient` 类（推荐）

最适合流式响应、多轮对话和交互式应用程序：

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def main():
    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            system_prompt="您是一名性能工程师",
            allowed_tools=["Bash", "Read", "WebSearch"],
            max_turns=5
        )
    ) as client:
        await client.query("分析系统性能")

        # 流式响应
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        print(block.text, end='', flush=True)

# 作为脚本运行
asyncio.run(main())

# 或在 IPython/Jupyter 中：await main()

2. `query` 函数

用于简单的一次性查询：

from claude_code_sdk import query, ClaudeCodeOptions

async for message in query(
    prompt="分析系统性能",
    options=ClaudeCodeOptions(system_prompt="您是一名性能工程师")
):
    if type(message).__name__ == "ResultMessage":
        print(message.result)

配置选项

Python SDK 通过 ClaudeCodeOptions 类接受命令行支持的所有参数。

ClaudeCodeOptions 参数

from claude_code_sdk import ClaudeCodeOptions

options = ClaudeCodeOptions(
    # 核心配置
    system_prompt="您是一个有用的助手",
    append_system_prompt="附加系统指令",
    max_turns=5,
    model="claude-3-5-sonnet-20241022",
    max_thinking_tokens=8000,
    
    # 工具管理
    allowed_tools=["Bash", "Read", "Write"],
    disallowed_tools=["WebSearch"],
    
    # 会话管理
    continue_conversation=False,
    resume="session-uuid",
    
    # 环境
    cwd="/path/to/working/directory",
    add_dirs=["/additional/context/dir"],
    settings="/path/to/settings.json",
    
    # 权限
    permission_mode="acceptEdits",  # "default", "acceptEdits", "plan", "bypassPermissions"
    permission_prompt_tool_name="mcp__approval_tool",
    
    # MCP 集成
    mcp_servers={
        "my_server": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-example"],
            "env": {"API_KEY": "your-key"}
        }
    },
    
    # 高级
    extra_args={"--verbose": None, "--custom-flag": "value"}
)

参数详情

system_prompt: str | None - 定义代理角色的自定义系统提示
append_system_prompt: str | None - 附加到系统提示的额外文本
max_turns: int | None - 最大对话轮数（如果为 None 则无限制）
model: str | None - 要使用的特定 Claude 模型
max_thinking_tokens: int - Claude 思考过程的最大令牌数（默认：8000）
allowed_tools: list[str] - 特别允许使用的工具
disallowed_tools: list[str] - 不应使用的工具
continue_conversation: bool - 继续最近的对话（默认：False）
resume: str | None - 要恢复特定对话的会话 UUID
cwd: str | Path | None - 会话的工作目录
add_dirs: list[str | Path] - 要包含在上下文中的附加目录
settings: str | None - 设置文件路径或设置 JSON 字符串
permission_mode: str | None - 权限处理模式
permission_prompt_tool_name: str | None - 自定义权限提示工具名称
mcp_servers: dict | str | Path - MCP 服务器配置
extra_args: dict[str, str | None] - 向底层 Claude Code CLI 传递任意 CLI 标志

权限模式

"default": CLI 提示危险工具（默认行为）
"acceptEdits": 自动接受文件编辑而不提示
"plan": 计划模式 - 分析而不进行更改
"bypassPermissions": 允许所有工具而不提示（谨慎使用）

高级配置示例

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def advanced_agent():
    """展示高级配置选项的示例"""
    
    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            # 自定义工作目录和附加上下文
            cwd="/project/root",
            add_dirs=["/shared/libs", "/common/utils"],
            
            # 模型和思考配置
            model="claude-3-5-sonnet-20241022",
            max_thinking_tokens=12000,
            
            # 高级工具控制
            allowed_tools=["Read", "Write", "Bash", "Grep"],
            disallowed_tools=["WebSearch", "Bash(rm*)"],
            
            # 自定义设置和 CLI 参数
            settings='{"editor": "vim", "theme": "dark"}',
            extra_args={
                "--verbose": None,
                "--timeout": "300"
            }
        )
    ) as client:
        await client.query("分析代码库结构")
        
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        print(block.text, end='', flush=True)

asyncio.run(advanced_agent())

结构化消息和图像输入

SDK 支持传递结构化消息和图像输入：

from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async with ClaudeSDKClient() as client:
    # 文本消息
    await client.query("分析此代码的安全问题")

    # 带有图像引用的消息（图像将由 Claude 的 Read 工具读取）
    await client.query("解释 screenshot.png 中显示的内容")

    # 按顺序发送多条消息
    messages = [
        "首先，分析 diagram.png 中的架构图",
        "现在根据图表建议改进",
        "最后，生成实现代码"
    ]

    for msg in messages:
        await client.query(msg)
        async for response in client.receive_response():
            # 处理每个响应
            pass

# SDK 通过 Claude 的内置 Read 工具处理图像文件
# 支持的格式：PNG、JPG、PDF 和其他常见格式

多轮对话

方法 1：使用 ClaudeSDKClient 进行持久对话

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions, query

# 方法 1：使用 ClaudeSDKClient 进行持久对话
async def multi_turn_conversation():
    async with ClaudeSDKClient() as client:
        # 第一个查询
        await client.query("让我们重构支付模块")
        async for msg in client.receive_response():
            # 处理第一个响应
            pass

        # 在同一会话中继续
        await client.query("现在添加全面的错误处理")
        async for msg in client.receive_response():
            # 处理继续
            pass

        # 对话上下文在整个过程中保持

# 方法 2：使用带有会话管理的 query 函数
async def resume_session():
    # 继续最近的对话
    async for message in query(
        prompt="现在重构这个以获得更好的性能",
        options=ClaudeCodeOptions(continue_conversation=True)
    ):
        if type(message).__name__ == "ResultMessage":
            print(message.result)

    # 恢复特定会话
    async for message in query(
        prompt="更新测试",
        options=ClaudeCodeOptions(
            resume="550e8400-e29b-41d4-a716-446655440000",
            max_turns=3
        )
    ):
        if type(message).__name__ == "ResultMessage":
            print(message.result)

# 运行示例
asyncio.run(multi_turn_conversation())

自定义系统提示

系统提示定义您的代理的角色、专业知识和行为：

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def specialized_agents():
    # 带有流式传输的 SRE 事件响应代理
    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            system_prompt="您是 SRE 专家。系统地诊断问题并提供可操作的解决方案。",
            max_turns=3
        )
    ) as sre_agent:
        await sre_agent.query("API 宕机，请调查")

        # 流式传输诊断过程
        async for message in sre_agent.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        print(block.text, end='', flush=True)

    # 带有自定义提示的法律审查代理
    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            append_system_prompt="始终包含全面的错误处理和单元测试。",
            max_turns=2
        )
    ) as dev_agent:
        await dev_agent.query("重构这个函数")

        # 收集完整响应
        full_response = []
        async for message in dev_agent.receive_response():
            if type(message).__name__ == "ResultMessage":
                print(message.result)

asyncio.run(specialized_agents())

通过 MCP 的自定义工具

模型上下文协议（MCP）让您为代理提供自定义工具和功能：

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def mcp_enabled_agent():
    # 带有文档访问和流式传输的法律代理
    # 注意：根据需要配置您的 MCP 服务器
    mcp_servers = {
        # 示例配置 - 根据需要取消注释并配置：
        # "docusign": {
        #     "command": "npx",
        #     "args": ["-y", "@modelcontextprotocol/server-docusign"],
        #     "env": {"API_KEY": "your-key"}
        # }
    }

    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            mcp_servers=mcp_servers,
            allowed_tools=["mcp__docusign", "mcp__compliance_db"],
            system_prompt="您是专门从事合同审查的企业律师。",
            max_turns=4
        )
    ) as client:
        await client.query("审查此合同的合规风险")

        # 监控工具使用和响应
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'type'):
                        if block.type == 'tool_use':
                            print(f"\n[使用工具：{block.name}]\n")
                        elif hasattr(block, 'text'):
                            print(block.text, end='', flush=True)
                    elif hasattr(block, 'text'):
                        print(block.text, end='', flush=True)

            if type(message).__name__ == "ResultMessage":
                print(f"\n\n审查完成。总成本：${message.total_cost_usd:.4f}")

asyncio.run(mcp_enabled_agent())

自定义权限提示工具

为工具调用实现自定义权限处理：

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def use_permission_prompt():
    """使用自定义权限提示工具的示例"""

    # MCP 服务器配置
    mcp_servers = {
        # 示例配置 - 根据需要取消注释并配置：
        # "security": {
        #     "command": "npx",
        #     "args": ["-y", "@modelcontextprotocol/server-security"],
        #     "env": {"API_KEY": "your-key"}
        # }
    }

    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            permission_prompt_tool_name="mcp__security__approval_prompt",  # 从 permission_prompt_tool 更改
            mcp_servers=mcp_servers,
            allowed_tools=["Read", "Grep"],
            disallowed_tools=["Bash(rm*)", "Write"],
            system_prompt="您是安全审计员"
        )
    ) as client:
        await client.query("分析并修复安全问题")

        # 监控工具使用和权限
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'type'):  # 添加了对 'type' 属性的检查
                        if block.type == 'tool_use':
                            print(f"[工具：{block.name}] ", end='')
                    if hasattr(block, 'text'):
                        print(block.text, end='', flush=True)

            # 检查错误消息中的权限拒绝
            if type(message).__name__ == "ErrorMessage":
                if hasattr(message, 'error') and "Permission denied" in str(message.error):
                    print(f"\n⚠️ 权限被拒绝：{message.error}")

# MCP 服务器实现示例（Python）
# 这将在您的 MCP 服务器代码中
async def approval_prompt(tool_name: str, input: dict, tool_use_id: str = None):
    """自定义权限提示处理程序"""
    # 您的自定义逻辑在这里
    if "allow" in str(input):
        return json.dumps({
            "behavior": "allow",
            "updatedInput": input
        })
    else:
        return json.dumps({
            "behavior": "deny",
            "message": f"权限被拒绝：{tool_name}"
        })

asyncio.run(use_permission_prompt())

输出格式

带有流式传输的文本输出

# 带有流式传输的默认文本输出
async with ClaudeSDKClient() as client:
    await client.query("解释文件 src/components/Header.tsx")

    # 在文本到达时流式传输
    async for message in client.receive_response():
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text'):
                    print(block.text, end='', flush=True)
                    # 输出实时流式传输：这是一个显示...的 React 组件

带有元数据的 JSON 输出

# 收集所有带有元数据的消息
async with ClaudeSDKClient() as client:
    await client.query("数据层是如何工作的？")

    messages = []
    result_data = None

    async for message in client.receive_messages():
        messages.append(message)

        # 捕获带有元数据的结果消息
        if type(message).__name__ == "ResultMessage":
            result_data = {
                "result": message.result,
                "cost": message.total_cost_usd,
                "duration": message.duration_ms,
                "num_turns": message.num_turns,
                "session_id": message.session_id
            }
            break

    print(result_data)

输入格式

import asyncio
from claude_code_sdk import ClaudeSDKClient

async def process_inputs():
    async with ClaudeSDKClient() as client:
        # 文本输入
        await client.query("解释这段代码")
        async for message in client.receive_response():
            # 处理流式响应
            pass

        # 图像输入（Claude 将自动使用 Read 工具）
        await client.query("这个图表中有什么？screenshot.png")
        async for message in client.receive_response():
            # 处理图像分析
            pass

        # 混合内容的多个输入
        inputs = [
            "分析 diagram.png 中的架构",
            "将其与最佳实践进行比较",
            "生成改进版本"
        ]

        for prompt in inputs:
            await client.query(prompt)
            async for message in client.receive_response():
                # 处理每个响应
                pass

asyncio.run(process_inputs())

代理集成示例

SRE 事件响应代理

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def investigate_incident(incident_description: str, severity: str = "medium"):
    """带有实时流式传输的自动化事件响应代理"""

    # 监控工具的 MCP 服务器配置
    mcp_servers = {
        # 示例配置 - 根据需要取消注释并配置：
        # "datadog": {
        #     "command": "npx",
        #     "args": ["-y", "@modelcontextprotocol/server-datadog"],
        #     "env": {"API_KEY": "your-datadog-key", "APP_KEY": "your-app-key"}
        # }
    }

    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            system_prompt="您是 SRE 专家。系统地诊断问题并提供可操作的解决方案。",
            max_turns=6,
            allowed_tools=["Bash", "Read", "WebSearch", "mcp__datadog"],
            mcp_servers=mcp_servers
        )
    ) as client:
        # 发送事件详情
        prompt = f"事件：{incident_description}（严重性：{severity}）"
        print(f"🚨 调查中：{prompt}\n")
        await client.query(prompt)

        # 流式传输调查过程
        investigation_log = []
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'type'):
                        if block.type == 'tool_use':
                            print(f"[{block.name}] ", end='')
                    if hasattr(block, 'text'):
                        text = block.text
                        print(text, end='', flush=True)
                        investigation_log.append(text)

            # 捕获最终结果
            if type(message).__name__ == "ResultMessage":
                return {
                    'analysis': ''.join(investigation_log),
                    'cost': message.total_cost_usd,
                    'duration_ms': message.duration_ms
                }

# 用法
result = await investigate_incident("支付 API 返回 500 错误", "high")
print(f"\n\n调查完成。成本：${result['cost']:.4f}")

自动化安全审查

import subprocess
import asyncio
import json
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def audit_pr(pr_number: int):
    """带有流式反馈的拉取请求安全审计代理"""
    # 获取 PR 差异
    pr_diff = subprocess.check_output(
        ["gh", "pr", "diff", str(pr_number)],
        text=True
    )

    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            system_prompt="您是安全工程师。审查此 PR 的漏洞、不安全模式和合规问题。",
            max_turns=3,
            allowed_tools=["Read", "Grep", "WebSearch"]
        )
    ) as client:
        print(f"🔍 审计 PR #{pr_number}\n")
        await client.query(pr_diff)

        findings = []
        async for message in client.receive_response():
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        # 在发现时流式传输发现
                        print(block.text, end='', flush=True)
                        findings.append(block.text)

            if type(message).__name__ == "ResultMessage":
                return {
                    'pr_number': pr_number,
                    'findings': ''.join(findings),
                    'metadata': {
                        'cost': message.total_cost_usd,
                        'duration': message.duration_ms,
                        'severity': 'high' if 'vulnerability' in ''.join(findings).lower() else 'medium'
                    }
                }

# 用法
report = await audit_pr(123)
print(f"\n\n审计完成。严重性：{report['metadata']['severity']}")
print(json.dumps(report, indent=2))

多轮法律助理

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def legal_review():
    """带有持久会话和流式传输的法律文档审查"""

    async with ClaudeSDKClient(
        options=ClaudeCodeOptions(
            system_prompt="您是企业律师。提供详细的法律分析。",
            max_turns=2
        )
    ) as client:
        # 在同一会话中进行多步骤审查
        steps = [
            "审查 contract.pdf 的责任条款",
            "检查 GDPR 要求的合规性",
            "生成风险的执行摘要"
        ]

        review_results = []

        for step in steps:
            print(f"\n📋 {step}\n")
            await client.query(step)

            step_result = []
            async for message in client.receive_response():
                if hasattr(message, 'content'):
                    for block in message.content:
                        if hasattr(block, 'text'):
                            text = block.text
                            print(text, end='', flush=True)
                            step_result.append(text)

                if type(message).__name__ == "ResultMessage":
                    review_results.append({
                        'step': step,
                        'analysis': ''.join(step_result),
                        'cost': message.total_cost_usd
                    })

        # 摘要
        total_cost = sum(r['cost'] for r in review_results)
        print(f"\n\n✅ 法律审查完成。总成本：${total_cost:.4f}")
        return review_results

# 用法
results = await legal_review()

Python 特定最佳实践

关键模式

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

# 始终使用上下文管理器
async with ClaudeSDKClient() as client:
    await client.query("分析这段代码")
    async for msg in client.receive_response():
        # 处理流式消息
        pass

# 并发运行多个代理
async with ClaudeSDKClient() as reviewer, ClaudeSDKClient() as tester:
    await asyncio.gather(
        reviewer.query("审查 main.py"),
        tester.query("为 main.py 编写测试")
    )

# 错误处理
from claude_code_sdk import CLINotFoundError, ProcessError

try:
    async with ClaudeSDKClient() as client:
        # 您的代码在这里
        pass
except CLINotFoundError:
    print("安装 CLI：npm install -g @anthropic-ai/claude-code")
except ProcessError as e:
    print(f"进程错误：{e}")

# 收集带有元数据的完整响应
async def get_response(client, prompt):
    await client.query(prompt)
    text = []
    async for msg in client.receive_response():
        if hasattr(msg, 'content'):
            for block in msg.content:
                if hasattr(block, 'text'):
                    text.append(block.text)
        if type(msg).__name__ == "ResultMessage":
            return {'text': ''.join(text), 'cost': msg.total_cost_usd}

IPython/Jupyter 提示

# 在 Jupyter 中，直接在单元格中使用 await
client = ClaudeSDKClient()
await client.connect()
await client.query("分析 data.csv")
async for msg in client.receive_response():
    print(msg)
await client.disconnect()

# 创建可重用的辅助函数
async def stream_print(client, prompt):
    await client.query(prompt)
    async for msg in client.receive_response():
        if hasattr(msg, 'content'):
            for block in msg.content:
                if hasattr(block, 'text'):
                    print(block.text, end='', flush=True)

开始使用

使用 Claude Code 构建

Claude 代码 SDK

部署

管理

配置

参考

资源

前提条件

安装

快速开始

基本用法

1. `ClaudeSDKClient` 类（推荐）

2. `query` 函数

配置选项

ClaudeCodeOptions 参数

参数详情

权限模式

高级配置示例

结构化消息和图像输入

多轮对话

方法 1：使用 ClaudeSDKClient 进行持久对话

自定义系统提示

通过 MCP 的自定义工具

自定义权限提示工具

输出格式

带有流式传输的文本输出

带有元数据的 JSON 输出

输入格式

代理集成示例

SRE 事件响应代理

自动化安全审查

多轮法律助理

Python 特定最佳实践

关键模式

IPython/Jupyter 提示

相关资源

开始使用

使用 Claude Code 构建

Claude 代码 SDK

部署

管理

配置

参考

资源

​前提条件

​安装

​快速开始

​基本用法

​1. ClaudeSDKClient 类（推荐）

​2. query 函数

​配置选项

​ClaudeCodeOptions 参数

​参数详情

​权限模式

​高级配置示例

​结构化消息和图像输入

​多轮对话

​方法 1：使用 ClaudeSDKClient 进行持久对话

​自定义系统提示

​通过 MCP 的自定义工具

​自定义权限提示工具

​输出格式

​带有流式传输的文本输出

​带有元数据的 JSON 输出

​输入格式

​代理集成示例

​SRE 事件响应代理

​自动化安全审查

​多轮法律助理

​Python 特定最佳实践

​关键模式

​IPython/Jupyter 提示

​相关资源

前提条件

安装

快速开始

基本用法

1. `ClaudeSDKClient` 类（推荐）

2. `query` 函数

配置选项

ClaudeCodeOptions 参数

参数详情

权限模式

高级配置示例

结构化消息和图像输入

多轮对话

方法 1：使用 ClaudeSDKClient 进行持久对话

自定义系统提示

通过 MCP 的自定义工具

自定义权限提示工具

输出格式

带有流式传输的文本输出

带有元数据的 JSON 输出

输入格式

代理集成示例

SRE 事件响应代理

自动化安全审查

多轮法律助理

Python 特定最佳实践

关键模式

IPython/Jupyter 提示

相关资源