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";

// SRE 事件回應代理
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()