Claude Code SDK
リソース
Claude Code SDK
Python
Claude Code Python SDKでカスタムAIエージェントを構築する
前提条件
- 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
- 特定の会話を再開するセッションUUIDcwd
: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)
関連リソース
- CLI使用法と制御 - 完全なCLIドキュメント
- GitHub Actions統合 - ClaudeでGitHubワークフローを自動化
- 一般的なワークフロー - 一般的な使用例のステップバイステップガイド
Was this page helpful?
On this page
- 前提条件
- インストール
- クイックスタート
- 基本的な使用方法
- 1. ClaudeSDKClientクラス(推奨)
- 2. query関数
- 設定オプション
- ClaudeCodeOptionsパラメータ
- パラメータの詳細
- 権限モード
- 高度な設定例
- 構造化メッセージと画像入力
- マルチターン会話
- 方法1:持続的な会話にClaudeSDKClientを使用
- カスタムシステムプロンプト
- MCPによるカスタムツール
- カスタム権限プロンプトツール
- 出力形式
- ストリーミング付きテキスト出力
- メタデータ付きJSON出力
- 入力形式
- エージェント統合例
- SREインシデント対応エージェント
- 自動セキュリティレビュー
- マルチターン法務アシスタント
- Python固有のベストプラクティス
- 主要パターン
- IPython/Jupyterのヒント
- 関連リソース