Claude Code SDK

claude -p "分析系统性能" \
  --append-system-prompt "您是一名性能工程师" \
  --allowedTools "Bash,Read,WebSearch" \
  --permission-mode acceptEdits \
  --cwd /path/to/project

import { query } from "@anthropic-ai/claude-code";

for await (const message of query({
  prompt: "分析系统性能",
  abortController: new AbortController(),
  options: {
    maxTurns: 5,
    systemPrompt: "您是一名性能工程师",
    allowedTools: ["Bash", "Read", "WebSearch"]
  }
})) {
  if (message.type === "result") {
    console.log(message.result);
  }
}

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()

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 和其他常见格式

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)

# 继续最近的对话
claude --continue "现在重构这个以获得更好的性能"

# 通过会话 ID 恢复特定对话
claude --resume 550e8400-e29b-41d4-a716-446655440000 "更新测试"

# 在非交互模式下恢复
claude --resume 550e8400-e29b-41d4-a716-446655440000 "修复所有 linting 问题" --no-interactive

import { query } from "@anthropic-ai/claude-code";

// 继续最近的对话
for await (const message of query({
  prompt: "现在重构这个以获得更好的性能",
  options: { continueSession: true }
})) {
  if (message.type === "result") console.log(message.result);
}

// 恢复特定会话
for await (const message of query({
  prompt: "更新测试",
  options: { 
    resumeSessionId: "550e8400-e29b-41d4-a716-446655440000",
    maxTurns: 3
  }
})) {
  if (message.type === "result") console.log(message.result);
}

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())

claude -p "审查此代码" --permission-mode plan

import { query } from "@anthropic-ai/claude-code";

for await (const message of query({
  prompt: "您的提示在这里",
  options: {
    permissionMode: 'plan'
  }
})) {
  if (message.type === "result") {
    console.log(message.result);
  }
}

from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async with ClaudeSDKClient(
    options=ClaudeCodeOptions(permission_mode='plan')
) as client:
    await client.query("您的提示在这里")

# SRE 事件响应代理
claude -p "API 宕机，调查" \
  --append-system-prompt "您是 SRE 专家。系统性地诊断问题并提供可操作的解决方案。"

# 法律文档审查代理
claude -p "审查此合同" \
  --append-system-prompt "您是企业律师。识别风险，建议改进，并确保合规性。"

# 附加到默认系统提示
claude -p "重构此函数" \
  --append-system-prompt "始终包含全面的错误处理和单元测试。"

import { query } from "@anthropic-ai/claude-code";

// S RE 事件响应代理
for await (const message of query({
  prompt: "API 宕机，调查",
  options: {
    systemPrompt: "您是 SRE 专家。系统性地诊断问题并提供可操作的解决方案。",
    maxTurns: 3
  }
})) {
  if (message.type === "result") console.log(message.result);
}

// 附加到默认系统提示
for await (const message of query({
  prompt: "重构此函数",
  options: {
    appendSystemPrompt: "始终包含全面的错误处理和单元测试。",
    maxTurns: 2
  }
})) {
  if (message.type === "result") console.log(message.result);
}

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())

# 带有监控工具的 SRE 代理
claude -p "调查支付服务中断" \
  --mcp-config sre-tools.json \
  --allowedTools "mcp__datadog,mcp__pagerduty,mcp__kubernetes" \
  --append-system-prompt "您是 SRE。使用监控数据诊断问题。"

# 带有 CRM 访问的客户支持代理
claude -p "帮助解决客户工单 #12345" \
  --mcp-config support-tools.json \
  --allowedTools "mcp__zendesk,mcp__stripe,mcp__user_db" \
  --append-system-prompt "您是技术支持专家。"

import { query } from "@anthropic-ai/claude-code";

// 带有监控工具的 SRE 代理
for await (const message of query({
  prompt: "调查支付服务中断",
  options: {
    mcpConfig: "sre-tools.json",
    allowedTools: ["mcp__datadog", "mcp__pagerduty", "mcp__kubernetes"],
    systemPrompt: "您是 SRE。使用监控数据诊断问题。",
    maxTurns: 4
  }
})) {
  if (message.type === "result") console.log(message.result);
}

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())

# 与您的 MCP 服务器配置一起使用
claude -p "分析并修复安全问题" \
  --permission-prompt-tool mcp__security__approval_prompt \
  --mcp-config security-tools.json \
  --allowedTools "Read,Grep" \
  --disallowedTools "Bash(rm*),Write"

# 使用自定义权限规则
claude -p "重构代码库" \
  --permission-prompt-tool mcp__custom__permission_check \
  --mcp-config custom-config.json \
  --output-format json

const server = new McpServer({
  name: "Test permission prompt MCP Server",
  version: "0.0.1",
});

server.tool(
  "approval_prompt",
  '模拟权限检查 - 如果输入包含"allow"则批准，否则拒绝',
  {
    tool_name: z.string().describe("请求权限的工具名称"),
    input: z.object({}).passthrough().describe("工具的输入"),
    tool_use_id: z.string().optional().describe("唯一的工具使用请求 ID"),
  },
  async ({ tool_name, input }) => {
    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(
            JSON.stringify(input).includes("allow")
              ? {
                  behavior: "allow",
                  updatedInput: input,
                }
              : {
                  behavior: "deny",
                  message: "测试 approval_prompt 工具拒绝权限",
                }
          ),
        },
      ],
    };
  }
);

// 在 SDK 中使用
import { query } from "@anthropic-ai/claude-code";

for await (const message of query({
  prompt: "分析代码库",
  options: {
    permissionPromptTool: "mcp__test-server__approval_prompt",
    mcpConfig: "my-config.json",
    allowedTools: ["Read", "Grep"]
  }
})) {
  if (message.type === "result") console.log(message.result);
}

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())

claude -p "解释文件 src/components/Header.tsx"
# 输出：这是一个显示...的 React 组件

// 默认文本输出
for await (const message of query({
  prompt: "解释文件 src/components/Header.tsx"
})) {
  if (message.type === "result") {
    console.log(message.result);
    // 输出：这是一个显示...的 React 组件
  }
}

# 带有流式传输的默认文本输出
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 组件

claude -p "数据层如何工作？" --output-format json

// 收集所有消息以进行类似 JSON 的访问
const messages = [];
for await (const message of query({
  prompt: "数据层如何工作？"
})) {
  messages.push(message);
}

// 访问带有元数据的结果消息
const result = messages.find(m => m.type === "result");
console.log({
  result: result.result,
  cost: result.total_cost_usd,
  duration: result.duration_ms
});

# 收集所有带有元数据的消息
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)

# 直接参数
claude -p "解释此代码"

# 从 stdin
echo "解释此代码" | claude -p

// 直接提示
for await (const message of query({
  prompt: "解释此代码"
})) {
  if (message.type === "result") console.log(message.result);
}

// 从变量
const userInput = "解释此代码";
for await (const message of query({ prompt: userInput })) {
  if (message.type === "result") console.log(message.result);
}

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())

#!/bin/bash

# 自动化事件响应代理
investigate_incident() {
    local incident_description="$1"
    local severity="${2:-medium}"
    
    claude -p "事件：$incident_description（严重性：$severity）" \
      --append-system-prompt "您是 SRE 专家。诊断问题，评估影响，并提供即时行动项目。" \
      --output-format json \
      --allowedTools "Bash,Read,WebSearch,mcp__datadog" \
      --mcp-config monitoring-tools.json
}

# 用法
investigate_incident "支付 API 返回 500 错误" "high"

import { query } from "@anthropic-ai/claude-code";

// 自动化事件响应代理
async function investigateIncident(
  incidentDescription: string, 
  severity = "medium"
) {
  const messages = [];
  
  for await (const message of query({
    prompt: `事件：${incidentDescription}（严重性：${severity}）`,
    options: {
      systemPrompt: "您是 SRE 专家。诊断问题，评估影响，并提供即时行动项目。",
      maxTurns: 6,
      allowedTools: ["Bash", "Read", "WebSearch", "mcp__datadog"],
      mcpConfig: "monitoring-tools.json"
    }
  })) {
    messages.push(message);
  }
  
  return messages.find(m => m.type === "result");
}

// 用法
const result = await investigateIncident("支付 API 返回 500 错误", "high");
console.log(result.result);

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}")

# 拉取请求的安全审计代理
audit_pr() {
    local pr_number="$1"
    
    gh pr diff "$pr_number" | claude -p \
      --append-system-prompt "您是安全工程师。审查此 PR 的漏洞、不安全模式和合规问题。" \
      --output-format json \
      --allowedTools "Read,Grep,WebSearch"
}

# 用法并保存到文件
audit_pr 123 > security-report.json

import { query } from "@anthropic-ai/claude-code";
import { execSync } from "child_process";

async function auditPR(prNumber: number) {
  // 获取 PR 差异
  const prDiff = execSync(`gh pr diff ${prNumber}`, { encoding: 'utf8' });
  
  const messages = [];
  for await (const message of query({
    prompt: prDiff,
    options: {
      systemPrompt: "您是安全工程师。审查此 PR 的漏洞、不安全模式和合规问题。",
      maxTurns: 3,
      allowedTools: ["Read", "Grep", "WebSearch"]
    }
  })) {
    messages.push(message);
  }
  
  return messages.find(m => m.type === "result");
}

// 用法
const report = await auditPR(123);
console.log(JSON.stringify(report, null, 2));

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))

# 带有会话持久性的法律文档审查
session_id=$(claude -p "开始法律审查会话" --output-format json | jq -r '.session_id')

# 分多个步骤审查合同
claude -p --resume "$session_id" "审查 contract.pdf 的责任条款"
claude -p --resume "$session_id" "检查 GDPR 要求的合规性" 
claude -p --resume "$session_id" "生成风险的执行摘要"

import { query } from "@anthropic-ai/claude-code";

async function legalReview() {
  // 开始法律审查会话
  let sessionId: string;
  
  for await (const message of query({
    prompt: "开始法律审查会话",
    options: { maxTurns: 1 }
  })) {
    if (message.type === "system" && message.subtype === "init") {
      sessionId = message.session_id;
    }
  }
  
  // 使用同一会话的多步骤审查
  const steps = [
    "审查 contract.pdf 的责任条款",
    "检查 GDPR 要求的合规性",
    "生成风险的执行摘要"
  ];
  
  for (const step of steps) {
    for await (const message of query({
      prompt: step,
      options: { resumeSessionId: sessionId, maxTurns: 2 }
    })) {
      if (message.type === "result") {
        console.log(`步骤：${step}`);
        console.log(message.result);
      }
    }
  }
}

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()