「Claude Codeに頼んでいた作業を、Pythonスクリプトから自動で回せないか？」——そう思ったことがあるなら、Claude Agent SDKのバッチ処理が答えになる。
Claude Agent SDKは2026年初頭にAnthropicが公開したライブラリで、Claude Codeを動かしている内部のエージェントループをPython/TypeScript APIとして提供する。ファイルの読み書き、コマンド実行、Web検索——Claude Codeが使えるツールをそのままプログラムから呼び出せる。
この記事では特に「バッチ処理」に焦点を当てる。複数のファイルを並列で処理したい、定期的なデータパイプラインを組みたい、大量のテキストを自律的に分析させたい——そういったユースケースで、Claude Agent SDKがどう使えるかをコード例つきで示す。
まず5分で動かすセットアップ
Claude Agent SDKのインストールは1コマンドだ。
# Python版
pip install claude-agent-sdk
# TypeScript版
npm install @anthropic-ai/claude-agent-sdk
# APIキーを環境変数に設定
export ANTHROPIC_API_KEY=your-api-key
最小限の動作確認コードを見てみよう。
# hello_agent.py — 最小構成の動作確認
# 動作環境: Python 3.11+, claude-agent-sdk>=0.1
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
async for message in query(
prompt="カレントディレクトリのファイル一覧を表示して",
options=ClaudeAgentOptions(allowed_tools=["Bash", "Glob"]),
):
if hasattr(message, "result"):
print(message.result)
asyncio.run(main())
ポイント:
query()はasync generatorで、メッセージをストリームとして返すallowed_toolsでエージェントが使えるツールを制限できる（セキュリティ上重要）message.resultが最終的な実行結果
ここまでが基本。次からバッチ処理の実装に入る。
Claude Agent SDKの全体像
バッチ処理の実装を始める前に、SDKの主要な概念を整理しておく。
利用可能な組み込みツール
| ツール名 | できること | 主なユースケース |
|---|---|---|
| Read | ファイルを読む | コードレビュー、ドキュメント分析 |
| Write | ファイルを作る | レポート生成、コード生成 |
| Edit | ファイルを部分編集 | バグ修正、リファクタリング |
| Bash | コマンド実行 | テスト実行、ビルド、データ変換 |
| Glob | ファイルパターン検索 | 対象ファイルの特定 |
| Grep | ファイル内容を正規表現検索 | パターン抽出、コード分析 |
| WebSearch | Webを検索 | 情報収集、事実確認 |
| WebFetch | URLのコンテンツを取得 | ページ解析、スクレイピング |
| Agent | サブエージェントを起動 | 並列処理、専門エージェントへの委任 |
（参照: Claude Agent SDK Overview — Anthropic Docs / 最終確認日: 2026-04-14）
AIエージェントの設計パターンの基礎はAIエージェント構築完全ガイドでまとめている。
Client SDKとの違い
Client SDK（従来の`anthropic`ライブラリ）と比較すると、役割の違いが明確だ。
# Client SDK: ツールループを自分で実装する
from anthropic import Anthropic
client = Anthropic()
response = client.messages.create(
model="claude-opus-4-6",
messages=[{"role": "user", "content": "auth.pyのバグを直して"}],
tools=[...], # ツール定義を手書き
)
# stop_reason == "tool_use" なら自分でツールを実行して再送信
while response.stop_reason == "tool_use":
result = execute_tool(response.tool_use) # 自前実装
response = client.messages.create(tool_result=result, ...)
# Agent SDK: ツールループはSDKが担う
from claude_agent_sdk import query, ClaudeAgentOptions
async for msg in query("auth.pyのバグを直して", ClaudeAgentOptions(...)):
print(msg) # Claudeがファイルを読み、バグを見つけ、修正する
バッチ処理パターン1: ファイル一括処理
最も典型的なバッチ処理が「複数ファイルへの一括適用」だ。例として、Pythonコードのリポジトリ全体にコメントを追加するケースを実装する。
# batch_file_processor.py — ファイル一括処理
# 動作環境: Python 3.11+, claude-agent-sdk>=0.1
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
# 警告: allow_tools=["Edit"] を指定するとファイルが書き換えられます。
# 必ず git でコミットした状態から実行してください。
import asyncio
import glob
from claude_agent_sdk import query, ClaudeAgentOptions
async def process_file(file_path: str) -> tuple[str, str]:
"""1つのファイルを処理してresultを返す"""
result_text = ""
async for message in query(
prompt=f"""
以下のファイルを読み、各関数・クラスにdocstringを追加してください。
- 既存のdocstringは変更しないこと
- Google Style docstringを使用すること
- ファイルパス: {file_path}
""",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Edit"],
permission_mode="acceptEdits", # 確認なしで編集を許可
),
):
if hasattr(message, "result"):
result_text = message.result
return file_path, result_text
async def batch_process(pattern: str = "src/**/*.py", max_concurrent: int = 3):
"""複数ファイルを並列バッチ処理"""
files = glob.glob(pattern, recursive=True)
print(f"{len(files)}件のファイルを処理します")
# semaphoreで同時実行数を制限（API rate limit対策）
semaphore = asyncio.Semaphore(max_concurrent)
async def limited_process(fp):
async with semaphore:
return await process_file(fp)
# 並列実行
tasks = [limited_process(f) for f in files]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 結果レポート
success, failed = [], []
for r in results:
if isinstance(r, Exception):
failed.append(str(r))
else:
fp, msg = r
success.append(fp)
print(f" OK: {fp}")
print(f"n完了: 成功 {len(success)}件 / 失敗 {len(failed)}件")
if failed:
for err in failed:
print(f" ERR: {err}")
if __name__ == "__main__":
asyncio.run(batch_process(pattern="src/**/*.py", max_concurrent=3))
ポイント:
asyncio.Semaphore(max_concurrent)で同時リクエスト数を制限。Claude APIにはレートリミットがあるため必須return_exceptions=Trueを使うと1つのエラーで全体が止まらないpermission_mode="acceptEdits"は確認なしで編集を許可するモード。本番では慎重に使うこと
バッチ処理パターン2: データパイプライン
次は「入力データを処理して構造化出力を生成する」パイプラインだ。例として、複数のWebページからの情報抽出を実装する。
# data_pipeline.py — URLリストからデータを並列抽出
# 動作環境: Python 3.11+, claude-agent-sdk>=0.1
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
import asyncio
import json
from pathlib import Path
from claude_agent_sdk import query, ClaudeAgentOptions
URLS = [
"https://example.com/product-a",
"https://example.com/product-b",
"https://example.com/product-c",
]
async def extract_product_info(url: str) -> dict:
"""URLからプロダクト情報を抽出"""
collected_result = ""
async for message in query(
prompt=f"""
以下のURLのページを取得し、以下の情報をJSON形式で返してください:
- name: 製品名
- price: 価格（数値のみ、円なし）
- features: 特徴のリスト（最大5つ）
- availability: 在庫状況（"available" or "unavailable"）
URL: {url}
必ずJSON形式のみで返すこと。説明文は不要。
""",
options=ClaudeAgentOptions(
allowed_tools=["WebFetch"],
),
):
if hasattr(message, "result"):
collected_result = message.result
# JSON部分を抽出してパース
try:
# ```json ... ``` ブロックがあれば取り出す
if "```json" in collected_result:
json_str = collected_result.split("```json")[1].split("```")[0].strip()
elif "```" in collected_result:
json_str = collected_result.split("```")[1].split("```")[0].strip()
else:
json_str = collected_result.strip()
return {"url": url, "data": json.loads(json_str), "error": None}
except (json.JSONDecodeError, IndexError) as e:
return {"url": url, "data": None, "error": str(e)}
async def run_pipeline(urls: list[str], output_path: str = "output.json"):
"""データパイプラインの実行"""
semaphore = asyncio.Semaphore(5) # 同時5リクエストまで
async def limited_extract(url):
async with semaphore:
return await extract_product_info(url)
print(f"{len(urls)}件のURLを処理中...")
results = await asyncio.gather(*[limited_extract(u) for u in urls])
# 結果を保存
Path(output_path).write_text(
json.dumps(results, ensure_ascii=False, indent=2),
encoding="utf-8"
)
print(f"結果を {output_path} に保存しました")
# サマリー表示
success = [r for r in results if r["error"] is None]
print(f"成功: {len(success)}/{len(results)}件")
if __name__ == "__main__":
asyncio.run(run_pipeline(URLS, output_path="products.json"))
バッチ処理パターン3: サブエージェントで専門分業
Claude Agent SDKの真骨頂は「エージェントがサブエージェントを起動する」再帰的な構成だ。大きなタスクを分割して専門エージェントに委任できる。
# multi_agent_review.py — コードレビューを専門エージェントで分業
# 動作環境: Python 3.11+, claude-agent-sdk>=0.1
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition
async def multi_agent_code_review(target_dir: str):
"""セキュリティ・パフォーマンス・スタイルの3専門エージェントでコードレビュー"""
result_text = ""
async for message in query(
prompt=f"""
{target_dir} のPythonコードを3つの観点でレビューしてください:
1. security-reviewer エージェントでセキュリティ問題を検出
2. performance-reviewer エージェントでパフォーマンス問題を検出
3. style-reviewer エージェントでコードスタイル問題を検出
各エージェントの結果を統合して最終レポートを作成してください。
""",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Glob", "Grep", "Agent"],
agents={
"security-reviewer": AgentDefinition(
description="セキュリティ専門のコードレビュアー。SQLインジェクション、XSS、秘密情報の漏洩等を検出。",
prompt="コードのセキュリティ脆弱性を分析し、問題点と修正案をMarkdownで報告してください。",
tools=["Read", "Glob", "Grep"],
),
"performance-reviewer": AgentDefinition(
description="パフォーマンス専門のコードレビュアー。N+1クエリ、メモリリーク、非効率なアルゴリズムを検出。",
prompt="コードのパフォーマンス問題を分析し、改善提案をMarkdownで報告してください。",
tools=["Read", "Glob", "Grep"],
),
"style-reviewer": AgentDefinition(
description="コードスタイル専門のレビュアー。PEP8準拠、命名規則、ドキュメント不足を検出。",
prompt="コードのスタイル問題を分析し、具体的な改善案をMarkdownで報告してください。",
tools=["Read", "Glob", "Grep"],
),
},
),
):
if hasattr(message, "result"):
result_text = message.result
return result_text
if __name__ == "__main__":
result = asyncio.run(multi_agent_code_review("./src"))
print(result)
# レポートをファイルに保存する場合
# Path("review_report.md").write_text(result, encoding="utf-8")
ポイント:
AgentDefinitionでサブエージェントの専門性を定義する- 親エージェントが`Agent`ツールを使って子エージェントを呼び出す設計
- 各サブエージェントには独立したtoolsリストを設定でき、権限を最小化できる
セッション管理: 状態を引き継いでバッチを継続
長時間のバッチ処理では、中断・再開が重要だ。Claude Agent SDKはセッションIDで会話コンテキストを保持できる。
# session_batch.py — セッションIDを使った継続バッチ
# 動作環境: Python 3.11+, claude-agent-sdk>=0.1
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, SystemMessage, ResultMessage
async def analyze_codebase_in_steps():
"""2段階で大規模コードベースを分析"""
# ステップ1: コードベースの概要を把握
session_id = None
async for message in query(
prompt="src/ ディレクトリの構造を把握して、主要コンポーネントを説明してください",
options=ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep"]),
):
# セッションIDを取得（初期化メッセージから）
if isinstance(message, SystemMessage) and message.subtype == "init":
session_id = message.data["session_id"]
print(f"セッションID取得: {session_id}")
if isinstance(message, ResultMessage):
print("ステップ1完了:", message.result[:200], "...")
# ステップ2: 前のコンテキストを保持したまま詳細分析
if session_id:
async for message in query(
prompt="把握した構造を踏まえて、認証フローのセキュリティリスクを分析してください",
options=ClaudeAgentOptions(
resume=session_id, # 前のセッションを再開
allowed_tools=["Read", "Grep"],
),
):
if isinstance(message, ResultMessage):
print("ステップ2完了:", message.result)
asyncio.run(analyze_codebase_in_steps())
【要注意】バッチ処理でよくある失敗パターン
失敗1: レートリミットを無視した並列実行
❌ よくある間違い: Semaphoreなしで大量のタスクを一斉にasyncio.gather()
⭕ 正しいアプローチ: asyncio.Semaphore(3〜5)で同時実行数を制御
なぜ重要か: AnthropicのAPIにはTPM（Tokens Per Minute）とRPM（Requests Per Minute）の制限がある。Semaphoreなしで100タスクを同時に走らせると、RateLimitErrorが頻発してバッチが途中で止まる。
失敗2: max_turnsを設定しない
❌ 無限ループのリスク: 複雑なタスクでエージェントが終了条件を見つけられず延々ループする
⭕ 必ず設定: ClaudeAgentOptions(max_turns=30)で上限を設ける
# 安全なオプション設定例
options = ClaudeAgentOptions(
allowed_tools=["Read", "Edit", "Bash"],
max_turns=30, # ループ上限
permission_mode="default", # 危険な操作は確認を求める
)
失敗3: エラーハンドリングを省略する
❌ `asyncio.gather()`でエラーを拾わない → 1つの失敗で全体のリザルトが壊れる
⭕ return_exceptions=True + 個別のtry/exceptで安全に処理する
# エラーハンドリングの正しいパターン
async def safe_process(item):
try:
return await process_item(item)
except Exception as e:
return {"item": item, "error": str(e), "status": "failed"}
results = await asyncio.gather(*[safe_process(i) for i in items], return_exceptions=True)
# return_exceptions=True でException自体もresultsに含まれる
errors = [r for r in results if isinstance(r, Exception)]
successes = [r for r in results if not isinstance(r, Exception)]
失敗4: allowed_toolsを広く設定しすぎる
❌ ファイル読み取りだけのタスクにallowed_tools=["Read", "Write", "Edit", "Bash"]
⭕ 必要最小限のツールだけ許可する（最小権限の原則）
なぜ重要か: Claudeはプロンプトインジェクション攻撃に対してある程度の防御を持つが、不必要なツールを与えると意図しないファイル操作が起きる可能性がある。WebFetchで取得したコンテンツに悪意あるプロンプトが含まれている場合のリスクを最小化するためにも、ツールは絞ること。
コスト管理: APIの使いすぎを防ぐ
バッチ処理はトークン消費が大きい。コスト管理の基本設定を示す。
# コスト制御のオプション
options = ClaudeAgentOptions(
allowed_tools=["Read", "Grep"], # 必要最小限
max_turns=20, # ターン数制限
# モデルを明示的に指定（小さいモデルでコスト削減）
# 2026年4月現在、Agent SDKはデフォルトでclaudeの最新版を使用
# 軽いタスクはclaude-haiku-4-5を検討
)
# 実行前にトークン使用量を概算
# 1ファイル処理: 約1,000〜5,000トークン（ファイルサイズ依存）
# 100ファイルバッチ: 100,000〜500,000トークン
# claude-opus-4-6: $15/1Mトークン → 100ファイルで$1.5〜7.5
参考・出典
- Claude Agent SDK Overview — Anthropic Docs（参照日: 2026-04-14）
- Batch Processing — Claude API Docs（参照日: 2026-04-14）
- claude-agent-sdk-python — GitHub (anthropics)（参照日: 2026-04-14）
- claude-agent-sdk — PyPI（参照日: 2026-04-14）
まとめ：今日から始める3つのアクション
- 今日やること:
pip install claude-agent-sdkを実行して、最小構成のhello_agent.pyを動かしてみる。5分でエージェントが動く感触をつかむ - 今週中: 自分のプロジェクトで「繰り返しClaude Codeに頼んでいる作業」を1つ選んでバッチ化を試みる。ファイルが10本以下なら並列処理の恩恵が出始める
- 今月中: サブエージェントパターンを1つ実装する。コードレビューを専門エージェントに分業させる構成は、CI/CDパイプラインへの組み込みにも発展しやすい
あわせて読みたい:
- AIエージェント構築完全ガイド — エージェントループの設計から本番運用まで
- AIエージェント構築ツール比較2026 — Claude Agent SDK vs LangGraph vs CrewAI
著者: 佐藤傑（さとう・すぐる）
株式会社Uravation代表取締役。X（@SuguruKun_ai）フォロワー10万人超。100社以上の企業向けAI研修・導入支援。著書累計3万部突破。SoftBank IT連載7回執筆（NewsPicks最大1,125ピックス）。
ご質問・ご相談は お問い合わせフォーム からお気軽にどうぞ。
