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 筆記本，您可以直接在儲存格中執行程式碼：

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] - 將任意 CLI 標誌傳遞給底層 Claude Code 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", "Rea", "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 提示

相關資源