関数

query()

Claude Code とやり取りするための主要な非同期関数です。メッセージが到着するたびにそれらを生成する非同期イテレータを返します。
async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeCodeOptions | None = None
) -> AsyncIterator[Message]

パラメータ

パラメータ説明
promptstr | AsyncIterable[dict]文字列としての入力プロンプト、またはストリーミングモード用の非同期イテラブル
optionsClaudeCodeOptions | Noneオプションの設定オブジェクト(None の場合は ClaudeCodeOptions() がデフォルト)

戻り値

会話からのメッセージを生成する AsyncIterator[Message] を返します。

例 - シンプルなクエリ

import asyncio
from claude_code_sdk import query

async def main():
    async for message in query(prompt="2+2は何ですか?"):
        print(message)

asyncio.run(main())

例 - オプション付き


import asyncio
from claude_code_sdk import query, ClaudeCodeOptions

async def main():
    options = ClaudeCodeOptions(
        system_prompt="あなたは専門的なPython開発者です",
        permission_mode='acceptEdits',
        cwd="/home/user/project"
    )

    async for message in query(
        prompt="Python ウェブサーバーを作成してください",
        options=options
    ):
        print(message)


asyncio.run(main())

tool()

型安全性を持つ MCP ツールを定義するためのデコレータです。
def tool(
    name: str,
    description: str,
    input_schema: type | dict[str, Any]
) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]

パラメータ

パラメータ説明
namestrツールの一意識別子
descriptionstrツールが何をするかの人間が読める説明
input_schematype | dict[str, Any]ツールの入力パラメータを定義するスキーマ(下記参照)

入力スキーマオプション

  1. シンプルな型マッピング(推奨):
    {"text": str, "count": int, "enabled": bool}
    
  2. JSON スキーマ形式(複雑な検証用):
    {
        "type": "object",
        "properties": {
            "text": {"type": "string"},
            "count": {"type": "integer", "minimum": 0}
        },
        "required": ["text"]
    }
    

戻り値

ツール実装をラップし、SdkMcpTool インスタンスを返すデコレータ関数です。

from claude_code_sdk import tool
from typing import Any

@tool("greet", "ユーザーに挨拶する", {"name": str})
async def greet(args: dict[str, Any]) -> dict[str, Any]:
    return {
        "content": [{
            "type": "text",
            "text": f"こんにちは、{args['name']}さん!"
        }]
    }

create_sdk_mcp_server()

Python アプリケーション内で実行されるインプロセス MCP サーバーを作成します。
def create_sdk_mcp_server(
    name: str,
    version: str = "1.0.0",
    tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig

パラメータ

パラメータデフォルト説明
namestr-サーバーの一意識別子
versionstr"1.0.0"サーバーのバージョン文字列
toolslist[SdkMcpTool[Any]] | NoneNone@tool デコレータで作成されたツール関数のリスト

戻り値

ClaudeCodeOptions.mcp_servers に渡すことができる McpSdkServerConfig オブジェクトを返します。

from claude_code_sdk import tool, create_sdk_mcp_server

@tool("add", "2つの数値を足す", {"a": float, "b": float})
async def add(args):
    return {
        "content": [{
            "type": "text",
            "text": f"合計: {args['a'] + args['b']}"
        }]
    }

@tool("multiply", "2つの数値を掛ける", {"a": float, "b": float})
async def multiply(args):
    return {
        "content": [{
            "type": "text",
            "text": f"積: {args['a'] * args['b']}"
        }]
    }

calculator = create_sdk_mcp_server(
    name="calculator",
    version="2.0.0",
    tools=[add, multiply]  # デコレートされた関数を渡す
)

# Claude で使用
options = ClaudeCodeOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=["mcp__calc__add", "mcp__calc__multiply"]
)

クラス

ClaudeSDKClient

Claude Code との双方向でインタラクティブな会話のためのクライアントです。ストリーミング、割り込み、動的メッセージ送信をサポートして、会話フローを完全に制御できます。
class ClaudeSDKClient:
    def __init__(self, options: ClaudeCodeOptions | None = None)
    async def connect(self, prompt: str | AsyncIterable[dict] | None = None) -> None
    async def query(self, prompt: str | AsyncIterable[dict], session_id: str = "default") -> None
    async def receive_messages(self) -> AsyncIterator[Message]
    async def receive_response(self) -> AsyncIterator[Message]
    async def interrupt(self) -> None
    async def disconnect(self) -> None

メソッド

メソッド説明
__init__(options)オプションの設定でクライアントを初期化
connect(prompt)オプションの初期プロンプトまたはメッセージストリームで Claude に接続
query(prompt, session_id)ストリーミングモードで新しいリクエストを送信
receive_messages()Claude からのすべてのメッセージを非同期イテレータとして受信
receive_response()ResultMessage を含むまでのメッセージを受信
interrupt()割り込み信号を送信(ストリーミングモードでのみ動作)
disconnect()Claude から切断

コンテキストマネージャーサポート

クライアントは自動接続管理のために非同期コンテキストマネージャーとして使用できます:
async with ClaudeSDKClient() as client:
    await client.query("こんにちは Claude")
    async for message in client.receive_response():
        print(message)
重要: メッセージを反復処理する際は、早期終了のために break を使用することは避けてください。これは asyncio のクリーンアップ問題を引き起こす可能性があります。代わりに、反復を自然に完了させるか、必要なものを見つけたときを追跡するためにフラグを使用してください。

例 - インタラクティブな会話

import asyncio
from claude_code_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage

async def main():
    async with ClaudeSDKClient() as client:
        # 初期メッセージを送信
        await client.query("数学の問題を段階的に解きましょう")
        
        # レスポンスを受信して処理
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"アシスタント: {block.text[:100]}...")
            elif isinstance(message, ResultMessage):
                print("レスポンス完了")
        
        # レスポンスに基づいてフォローアップを送信
        await client.query("80の15%は何ですか?")

asyncio.run(main())

SdkMcpTool

@tool デコレータで作成された SDK MCP ツールの定義です。
@dataclass
class SdkMcpTool(Generic[T]):
    name: str
    description: str
    input_schema: type[T] | dict[str, Any]
    handler: Callable[[T], Awaitable[dict[str, Any]]]
プロパティ説明
namestrツールの一意識別子
descriptionstr人間が読める説明
input_schematype[T] | dict[str, Any]入力検証のためのスキーマ
handlerCallable[[T], Awaitable[dict[str, Any]]]ツール実行を処理する非同期関数

ClaudeCodeOptions

Claude Code クエリの設定データクラスです。
@dataclass
class ClaudeCodeOptions:
    allowed_tools: list[str] = field(default_factory=list)
    max_thinking_tokens: int = 8000
    system_prompt: str | None = None
    append_system_prompt: str | None = None
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
    permission_mode: PermissionMode | None = None
    continue_conversation: bool = False
    resume: str | None = None
    max_turns: int | None = None
    disallowed_tools: list[str] = field(default_factory=list)
    model: str | None = None
    permission_prompt_tool_name: str | None = None
    cwd: str | Path | None = None
    settings: str | None = None
    add_dirs: list[str | Path] = field(default_factory=list)
    env: dict[str, str] = field(default_factory=dict)
    extra_args: dict[str, str | None] = field(default_factory=dict)
プロパティデフォルト説明
allowed_toolslist[str][]許可されたツール名のリスト
max_thinking_tokensint8000思考プロセスの最大トークン数
system_promptstr | NoneNoneデフォルトのシステムプロンプトを完全に置き換える
append_system_promptstr | NoneNoneデフォルトのシステムプロンプトに追加するテキスト
mcp_serversdict[str, McpServerConfig] | str | Path{}MCP サーバー設定または設定ファイルへのパス
permission_modePermissionMode | NoneNoneツール使用の許可モード
continue_conversationboolFalse最新の会話を続行
resumestr | NoneNone再開するセッション ID
max_turnsint | NoneNone最大会話ターン数
disallowed_toolslist[str][]禁止されたツール名のリスト
modelstr | NoneNone使用する Claude モデル
permission_prompt_tool_namestr | NoneNone許可プロンプト用の MCP ツール名
cwdstr | Path | NoneNone現在の作業ディレクトリ
settingsstr | NoneNone設定ファイルへのパス
add_dirslist[str | Path][]Claude がアクセスできる追加ディレクトリ
extra_argsdict[str, str | None]{}CLI に直接渡す追加の CLI 引数
can_use_toolCanUseTool | NoneNoneツール許可コールバック関数
hooksdict[HookEvent, list[HookMatcher]] | NoneNoneイベントをインターセプトするためのフック設定

PermissionMode

ツール実行を制御するための許可モードです。
PermissionMode = Literal[
    "default",           # 標準の許可動作
    "acceptEdits",       # ファイル編集を自動承認
    "plan",              # 計画モード - 実行なし
    "bypassPermissions"  # すべての許可チェックをバイパス(注意して使用)
]

McpSdkServerConfig

create_sdk_mcp_server() で作成された SDK MCP サーバーの設定です。
class McpSdkServerConfig(TypedDict):
    type: Literal["sdk"]
    name: str
    instance: Any  # MCP サーバーインスタンス

McpServerConfig

MCP サーバー設定のユニオン型です。
McpServerConfig = McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig

McpStdioServerConfig

class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]  # 後方互換性のためオプション
    command: str
    args: NotRequired[list[str]]
    env: NotRequired[dict[str, str]]

McpSSEServerConfig

class McpSSEServerConfig(TypedDict):
    type: Literal["sse"]
    url: str
    headers: NotRequired[dict[str, str]]

McpHttpServerConfig

class McpHttpServerConfig(TypedDict):
    type: Literal["http"]
    url: str
    headers: NotRequired[dict[str, str]]

メッセージ型

Message

すべての可能なメッセージのユニオン型です。
Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage

UserMessage

ユーザー入力メッセージです。
@dataclass
class UserMessage:
    content: str | list[ContentBlock]

AssistantMessage

コンテンツブロックを持つアシスタントレスポンスメッセージです。
@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str

SystemMessage

メタデータを持つシステムメッセージです。
@dataclass
class SystemMessage:
    subtype: str
    data: dict[str, Any]

ResultMessage

コストと使用情報を持つ最終結果メッセージです。
@dataclass
class ResultMessage:
    subtype: str
    duration_ms: int
    duration_api_ms: int
    is_error: bool
    num_turns: int
    session_id: str
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | None = None

コンテンツブロック型

ContentBlock

すべてのコンテンツブロックのユニオン型です。
ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

TextBlock

テキストコンテンツブロックです。
@dataclass
class TextBlock:
    text: str

ThinkingBlock

思考コンテンツブロック(思考機能を持つモデル用)です。
@dataclass
class ThinkingBlock:
    thinking: str
    signature: str

ToolUseBlock

ツール使用リクエストブロックです。
@dataclass
class ToolUseBlock:
    id: str
    name: str
    input: dict[str, Any]

ToolResultBlock

ツール実行結果ブロックです。
@dataclass
class ToolResultBlock:
    tool_use_id: str
    content: str | list[dict[str, Any]] | None = None
    is_error: bool | None = None

エラー型

ClaudeSDKError

すべての SDK エラーのベース例外クラスです。
class ClaudeSDKError(Exception):
    """Claude SDK のベースエラー。"""

CLINotFoundError

Claude Code CLI がインストールされていないか見つからない場合に発生します。
class CLINotFoundError(CLIConnectionError):
    def __init__(self, message: str = "Claude Code not found", cli_path: str | None = None):
        """
        Args:
            message: エラーメッセージ(デフォルト: "Claude Code not found")
            cli_path: 見つからなかった CLI へのオプションのパス
        """

CLIConnectionError

Claude Code への接続が失敗した場合に発生します。
class CLIConnectionError(ClaudeSDKError):
    """Claude Code への接続に失敗しました。"""

ProcessError

Claude Code プロセスが失敗した場合に発生します。
class ProcessError(ClaudeSDKError):
    def __init__(self, message: str, exit_code: int | None = None, stderr: str | None = None):
        self.exit_code = exit_code
        self.stderr = stderr

CLIJSONDecodeError

JSON パースが失敗した場合に発生します。
class CLIJSONDecodeError(ClaudeSDKError):
    def __init__(self, line: str, original_error: Exception):
        """
        Args:
            line: パースに失敗した行
            original_error: 元の JSON デコード例外
        """
        self.line = line
        self.original_error = original_error

フック型

HookEvent

サポートされているフックイベント型です。セットアップの制限により、Python SDK は SessionStart、SessionEnd、および Notification フックをサポートしていません。
HookEvent = Literal[
    "PreToolUse",      # ツール実行前に呼び出される
    "PostToolUse",     # ツール実行後に呼び出される
    "UserPromptSubmit", # ユーザーがプロンプトを送信したときに呼び出される
    "Stop",            # 実行停止時に呼び出される
    "SubagentStop",    # サブエージェント停止時に呼び出される
    "PreCompact"       # メッセージ圧縮前に呼び出される
]

HookCallback

フックコールバック関数の型定義です。
HookCallback = Callable[
    [dict[str, Any], str | None, HookContext],
    Awaitable[dict[str, Any]]
]
パラメータ:
  • input_data: フック固有の入力データ(フックドキュメントを参照)
  • tool_use_id: オプションのツール使用識別子(ツール関連フック用)
  • context: 追加情報を持つフックコンテキスト
以下を含む可能性のある辞書を返します:
  • decision: アクションをブロックするための "block"
  • systemMessage: トランスクリプトに追加するシステムメッセージ
  • hookSpecificOutput: フック固有の出力データ

HookContext

フックコールバックに渡されるコンテキスト情報です。
@dataclass
class HookContext:
    signal: Any | None = None  # 将来: 中止シグナルサポート

HookMatcher

特定のイベントやツールにフックをマッチングするための設定です。
@dataclass
class HookMatcher:
    matcher: str | None = None        # マッチするツール名またはパターン(例: "Bash", "Write|Edit")
    hooks: list[HookCallback] = field(default_factory=list)  # 実行するコールバックのリスト

フック使用例

from claude_code_sdk import query, ClaudeCodeOptions, HookMatcher, HookContext
from typing import Any

async def validate_bash_command(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """危険な bash コマンドを検証し、潜在的にブロックします。"""
    if input_data['tool_name'] == 'Bash':
        command = input_data['tool_input'].get('command', '')
        if 'rm -rf /' in command:
            return {
                'hookSpecificOutput': {
                    'hookEventName': 'PreToolUse',
                    'permissionDecision': 'deny',
                    'permissionDecisionReason': '危険なコマンドがブロックされました'
                }
            }
    return {}

async def log_tool_use(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """監査のためにすべてのツール使用をログに記録します。"""
    print(f"使用されたツール: {input_data.get('tool_name')}")
    return {}

options = ClaudeCodeOptions(
    hooks={
        'PreToolUse': [
            HookMatcher(matcher='Bash', hooks=[validate_bash_command]),
            HookMatcher(hooks=[log_tool_use])  # すべてのツールに適用
        ],
        'PostToolUse': [
            HookMatcher(hooks=[log_tool_use])
        ]
    }
)

async for message in query(
    prompt="このコードベースを分析してください",
    options=options
):
    print(message)

ツール入力/出力型

すべての組み込み Claude Code ツールの入力/出力スキーマのドキュメントです。Python SDK はこれらを型としてエクスポートしませんが、メッセージ内のツール入力と出力の構造を表しています。

Task

ツール名: Task 入力:
{
    "description": str,      # タスクの短い(3-5語)説明
    "prompt": str,           # エージェントが実行するタスク
    "subagent_type": str     # 使用する専門エージェントの種類
}
出力:
{
    "result": str,                    # サブエージェントからの最終結果
    "usage": dict | None,             # トークン使用統計
    "total_cost_usd": float | None,  # USD での総コスト
    "duration_ms": int | None         # ミリ秒での実行時間
}

Bash

ツール名: Bash 入力:
{
    "command": str,                  # 実行するコマンド
    "timeout": int | None,           # ミリ秒でのオプションのタイムアウト(最大 600000)
    "description": str | None,       # 明確で簡潔な説明(5-10語)
    "run_in_background": bool | None # バックグラウンドで実行する場合は true に設定
}
出力:
{
    "output": str,              # stdout と stderr の結合出力
    "exitCode": int,            # コマンドの終了コード
    "killed": bool | None,      # タイムアウトによりコマンドが強制終了されたかどうか
    "shellId": str | None       # バックグラウンドプロセスのシェル ID
}

Edit

ツール名: Edit 入力:
{
    "file_path": str,           # 変更するファイルの絶対パス
    "old_string": str,          # 置換するテキスト
    "new_string": str,          # 置換後のテキスト
    "replace_all": bool | None  # すべての出現を置換(デフォルト False)
}
出力:
{
    "message": str,      # 確認メッセージ
    "replacements": int, # 行われた置換の数
    "file_path": str     # 編集されたファイルパス
}

MultiEdit

ツール名: MultiEdit 入力:
{
    "file_path": str,     # 変更するファイルの絶対パス
    "edits": [            # 編集操作の配列
        {
            "old_string": str,          # 置換するテキスト
            "new_string": str,          # 置換後のテキスト
            "replace_all": bool | None  # すべての出現を置換
        }
    ]
}
出力:
{
    "message": str,       # 成功メッセージ
    "edits_applied": int, # 適用された編集の総数
    "file_path": str      # 編集されたファイルパス
}

Read

ツール名: Read 入力:
{
    "file_path": str,       # 読み取るファイルの絶対パス
    "offset": int | None,   # 読み取りを開始する行番号
    "limit": int | None     # 読み取る行数
}
出力(テキストファイル):
{
    "content": str,         # 行番号付きのファイル内容
    "total_lines": int,     # ファイルの総行数
    "lines_returned": int   # 実際に返された行数
}
出力(画像):
{
    "image": str,       # Base64 エンコードされた画像データ
    "mime_type": str,   # 画像の MIME タイプ
    "file_size": int    # バイト単位のファイルサイズ
}

Write

ツール名: Write 入力:
{
    "file_path": str,  # 書き込むファイルの絶対パス
    "content": str     # ファイルに書き込む内容
}
出力:
{
    "message": str,        # 成功メッセージ
    "bytes_written": int,  # 書き込まれたバイト数
    "file_path": str       # 書き込まれたファイルパス
}

Glob

ツール名: Glob 入力:
{
    "pattern": str,       # ファイルをマッチするための glob パターン
    "path": str | None    # 検索するディレクトリ(デフォルトは cwd)
}
出力:
{
    "matches": list[str],  # マッチしたファイルパスの配列
    "count": int,          # 見つかったマッチの数
    "search_path": str     # 使用された検索ディレクトリ
}

Grep

ツール名: Grep 入力:
{
    "pattern": str,                    # 正規表現パターン
    "path": str | None,                # 検索するファイルまたはディレクトリ
    "glob": str | None,                # ファイルをフィルタする glob パターン
    "type": str | None,                # 検索するファイルタイプ
    "output_mode": str | None,         # "content", "files_with_matches", または "count"
    "-i": bool | None,                 # 大文字小文字を区別しない検索
    "-n": bool | None,                 # 行番号を表示
    "-B": int | None,                  # 各マッチの前に表示する行数
    "-A": int | None,                  # 各マッチの後に表示する行数
    "-C": int | None,                  # 前後に表示する行数
    "head_limit": int | None,          # 出力を最初の N 行/エントリに制限
    "multiline": bool | None           # マルチラインモードを有効にする
}
出力(content モード):
{
    "matches": [
        {
            "file": str,
            "line_number": int | None,
            "line": str,
            "before_context": list[str] | None,
            "after_context": list[str] | None
        }
    ],
    "total_matches": int
}
出力(files_with_matches モード):
{
    "files": list[str],  # マッチを含むファイル
    "count": int         # マッチを含むファイルの数
}

NotebookEdit

ツール名: NotebookEdit 入力:
{
    "notebook_path": str,                     # Jupyter ノートブックへの絶対パス
    "cell_id": str | None,                    # 編集するセルの ID
    "new_source": str,                        # セルの新しいソース
    "cell_type": "code" | "markdown" | None,  # セルのタイプ
    "edit_mode": "replace" | "insert" | "delete" | None  # 編集操作タイプ
}
出力:
{
    "message": str, # 成功メッセージ
    "edit_type": "replaced" | "inserted" | "deleted",  # 実行された編集のタイプ
    "cell_id": str | None,                       # 影響を受けたセル ID
    "total_cells": int                           # 編集後のノートブックの総セル数
}

WebFetch

ツール名: WebFetch 入力:
{
    "url": str,     # コンテンツを取得する URL
    "prompt": str   # 取得したコンテンツに対して実行するプロンプト
}
出力:
{
    "response": str,           # プロンプトに対する AI モデルのレスポンス
    "url": str,                # 取得された URL
    "final_url": str | None,   # リダイレクト後の最終 URL
    "status_code": int | None  # HTTP ステータスコード
}

WebSearch

ツール名: WebSearch 入力:
{
    "query": str,                        # 使用する検索クエリ
    "allowed_domains": list[str] | None, # これらのドメインからの結果のみを含める
    "blocked_domains": list[str] | None  # これらのドメインからの結果を含めない
}
出力:
{
    "results": [
        {
            "title": str,
            "url": str,
            "snippet": str,
            "metadata": dict | None
        }
    ],
    "total_results": int,
    "query": str
}

TodoWrite

ツール名: TodoWrite 入力:
{
    "todos": [
        {
            "content": str, # タスクの説明
            "status": "pending" | "in_progress" | "completed",  # タスクのステータス
            "activeForm": str                            # 説明のアクティブフォーム
        }
    ]
}
出力:
{
    "message": str,  # 成功メッセージ
    "stats": {
        "total": int,
        "pending": int,
        "in_progress": int,
        "completed": int
    }
}

BashOutput

ツール名: BashOutput 入力:
{
    "bash_id": str,       # バックグラウンドシェルの ID
    "filter": str | None  # 出力行をフィルタするオプションの正規表現
}
出力:
{
    "output": str, # 最後のチェック以降の新しい出力
    "status": "running" | "completed" | "failed",       # 現在のシェルステータス
    "exitCode": int | None # 完了時の終了コード
}

KillBash

ツール名: KillBash 入力:
{
    "shell_id": str  # 強制終了するバックグラウンドシェルの ID
}
出力:
{
    "message": str,  # 成功メッセージ
    "shell_id": str  # 強制終了されたシェルの ID
}

ExitPlanMode

ツール名: ExitPlanMode 入力:
{
    "plan": str  # ユーザーの承認のために実行する計画
}
出力:
{
    "message": str,          # 確認メッセージ
    "approved": bool | None  # ユーザーが計画を承認したかどうか
}

ListMcpResources

ツール名: ListMcpResources 入力:
{
    "server": str | None  # リソースをフィルタするオプションのサーバー名
}
出力:
{
    "resources": [
        {
            "uri": str,
            "name": str,
            "description": str | None,
            "mimeType": str | None,
            "server": str
        }
    ],
    "total": int
}

ReadMcpResource

ツール名: ReadMcpResource 入力:
{
    "server": str,  # MCP サーバー名
    "uri": str      # 読み取るリソース URI
}
出力:
{
    "contents": [
        {
            "uri": str,
            "mimeType": str | None,
            "text": str | None,
            "blob": str | None
        }
    ],
    "server": str
}

使用例

基本的なファイル操作

from claude_code_sdk import query, ClaudeCodeOptions, AssistantMessage, ToolUseBlock
import asyncio

async def create_project():
    options = ClaudeCodeOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode='acceptEdits',
        cwd="/home/user/project"
    )
    
    async for message in query(
        prompt="setup.py を含む Python プロジェクト構造を作成してください",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    print(f"ツールを使用中: {block.name}")

asyncio.run(create_project())

エラーハンドリング

from claude_code_sdk import (
    query,
    CLINotFoundError,
    ProcessError,
    CLIJSONDecodeError
)

try:
    async for message in query(prompt="こんにちは"):
        print(message)
except CLINotFoundError:
    print("Claude Code をインストールしてください: npm install -g @anthropic-ai/claude-code")
except ProcessError as e:
    print(f"プロセスが終了コード {e.exit_code} で失敗しました")
except CLIJSONDecodeError as e:
    print(f"レスポンスのパースに失敗しました: {e}")

クライアントでのストリーミングモード

from claude_code_sdk import ClaudeSDKClient
import asyncio

async def interactive_session():
    async with ClaudeSDKClient() as client:
        # 初期メッセージを送信
        await client.query("天気はどうですか?")
        
        # レスポンスを処理
        async for msg in client.receive_response():
            print(msg)
        
        # フォローアップを送信
        await client.query("それについてもっと教えてください")
        
        # フォローアップレスポンスを処理
        async for msg in client.receive_response():
            print(msg)

asyncio.run(interactive_session())

カスタムツールの使用

from claude_code_sdk import (
    query,
    ClaudeCodeOptions,
    tool,
    create_sdk_mcp_server,
    AssistantMessage,
    TextBlock
)
import asyncio
from typing import Any

# @tool デコレータでカスタムツールを定義
@tool("calculate", "数学的計算を実行", {"expression": str})
async def calculate(args: dict[str, Any]) -> dict[str, Any]:
    try:
        result = eval(args["expression"], {"__builtins__": {}})
        return {
            "content": [{
                "type": "text",
                "text": f"結果: {result}"
            }]
        }
    except Exception as e:
        return {
            "content": [{
                "type": "text",
                "text": f"エラー: {str(e)}"
            }],
            "is_error": True
        }

@tool("get_time", "現在時刻を取得", {})
async def get_time(args: dict[str, Any]) -> dict[str, Any]:
    from datetime import datetime
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return {
        "content": [{
            "type": "text",
            "text": f"現在時刻: {current_time}"
        }]
    }

async def main():
    # カスタムツールで SDK MCP サーバーを作成
    my_server = create_sdk_mcp_server(
        name="utilities",
        version="1.0.0",
        tools=[calculate, get_time]
    )
    
    # サーバーでオプションを設定
    options = ClaudeCodeOptions(
        mcp_servers={"utils": my_server},
        allowed_tools=[
            "mcp__utils__calculate",
            "mcp__utils__get_time"
        ]
    )
    
    # カスタムツールが利用可能な状態で Claude にクエリ
    async for message in query(
        prompt="123 × 456 は何で、今何時ですか?",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

asyncio.run(main())

関連項目