前提条件

  • 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は2つの主要なインターフェースを提供します:

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によるカスタムツール

Model Context Protocol(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)

関連リソース