AIエージェントをPythonで作る完全ガイド — Anthropic SDK × tool_use でゼロから実装する【2026年版】
PythonでAIエージェントをゼロから実装する方法をAnthropicの公式SDKを使って解説。tool_useの仕様・エージェントループの設計・エラーハンドリング・コスト管理まで、LangChainなしの直実装で理解する。
エンジニアのゆとです。
「AIエージェントの作り方」で検索すると、LangChainやLangGraphを使った記事ばかり出てくる。フレームワークを使えば確かに速く作れるが、「エージェントが内部で何をやっているか」が見えにくくなる。
この記事ではAnthropicの公式PythonSDKだけを使って、AIエージェントをゼロから実装する。フレームワークなしで実装すると、tool_useの仕様・エージェントループの設計・トークンの流れが全部見えるようになる。それを理解した上でフレームワークに移行する方が、結果的に早く本番で動くものを作れる。
AIエージェントとは何か(3行で)
- 通常のLLM呼び出し: 人間がプロンプトを書く → AIが返答する(1往復)
- AIエージェント: AIが自律的にツールを使いながら目標を達成するまでループを回す(N往復)
例えば「今日の東京の天気を教えて」という質問に答えるとき、通常のLLMは学習データの知識しか使えない。エージェントは「天気検索ツール」を呼び出して最新情報を取得してから回答できる。
Anthropic SDKのtool_use仕様
ツールの定義
AnthropicのPython SDKでツールを定義する方法:
tools = [
{
"name": "get_weather",
"description": "指定した都市の現在の天気を取得する",
"input_schema": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "天気を調べたい都市名(例: Tokyo, Osaka)"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度の単位"
}
},
"required": ["city"]
}
}
]
input_schemaはJSON Schema形式で書く。requiredフィールドで必須パラメータを指定するのを忘れずに。
メッセージのやり取り
import anthropic
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
tools=tools,
messages=[
{"role": "user", "content": "東京の天気を教えて"}
]
)
print(response.stop_reason) # "tool_use" または "end_turn"
print(response.content)
Claudeが「ツールを使う必要がある」と判断すると、stop_reasonが"tool_use"になり、contentにツール呼び出しの情報が入ってくる。
tool_useのレスポンス形式
# response.content の例
[
TextBlock(text="東京の天気を調べてみます。", type="text"),
ToolUseBlock(
id="toolu_01abc...",
name="get_weather",
input={"city": "Tokyo", "unit": "celsius"},
type="tool_use"
)
]
ToolUseBlockのidは後で結果を返すときに必要になる。必ず保持しておく。
エージェントループの実装
これがエージェントの核心部分だ。
import anthropic
import json
client = anthropic.Anthropic()
def get_weather(city: str, unit: str = "celsius") -> str:
"""天気を返すモック関数(実際はAPIを叩く)"""
weather_data = {
"Tokyo": {"temp": 28, "condition": "晴れ"},
"Osaka": {"temp": 30, "condition": "くもり"},
}
data = weather_data.get(city, {"temp": 20, "condition": "不明"})
unit_label = "°C" if unit == "celsius" else "°F"
temp = data["temp"] if unit == "celsius" else data["temp"] * 9/5 + 32
return f"{city}の天気: {data['condition']}, 気温: {temp}{unit_label}"
# ツール実行の振り分け関数
def execute_tool(tool_name: str, tool_input: dict) -> str:
if tool_name == "get_weather":
return get_weather(**tool_input)
raise ValueError(f"Unknown tool: {tool_name}")
def run_agent(user_message: str) -> str:
"""エージェントのメインループ"""
messages = [{"role": "user", "content": user_message}]
while True:
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
tools=tools,
messages=messages
)
# エージェントが完了した場合
if response.stop_reason == "end_turn":
# テキストレスポンスを取得
for block in response.content:
if hasattr(block, "text"):
return block.text
return ""
# ツール呼び出しが必要な場合
if response.stop_reason == "tool_use":
# アシスタントのレスポンスをメッセージ履歴に追加
messages.append({
"role": "assistant",
"content": response.content
})
# 各ツール呼び出しを処理
tool_results = []
for block in response.content:
if block.type == "tool_use":
tool_result = execute_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id, # idの対応付けが重要
"content": tool_result
})
# ツール結果をメッセージ履歴に追加
messages.append({
"role": "user",
"content": tool_results
})
# ループを続ける(Claudeがツール結果を受け取って続きを考える)
else:
# 予期しないstop_reason
break
return "エージェントが予期せず終了しました"
# 実行
result = run_agent("東京と大阪の天気を比べて、どちらが過ごしやすいか教えて")
print(result)
このコードのポイントを整理する:
while Trueでループを回し続けるstop_reason == "end_turn"でClaudeが完了したと判断したらループを抜ける- ツール呼び出しがあったら実行して、結果を
"role": "user"として追加する tool_use_idでツールの呼び出しと結果を対応付ける(複数ツールがあるとき重要)
エージェントループで一番詰まるのはメッセージ履歴の形式ミスだ。アシスタントのtool_useブロックを履歴に追加した後、そのtool_use_idに対応するtool_resultを次のuserメッセージで返す、という順序が崩れるとAPIがエラーを返す。
複数ツールの同時呼び出し(Parallel Tool Use)
Claudeは1回のレスポンスで複数のツールを同時に呼び出せる。
# Claudeが1回で複数ツールを呼び出す例
# response.content に複数のToolUseBlockが含まれる
[
ToolUseBlock(id="toolu_01", name="get_weather", input={"city": "Tokyo"}),
ToolUseBlock(id="toolu_02", name="get_weather", input={"city": "Osaka"})
]
上のエージェントループの実装はすでにこれに対応している。for block in response.contentでループして、複数のツール結果をまとめてuserメッセージに追加すればいい。
並列実行のメリットは速度だ。上記の例だと東京と大阪を別々に呼び出しても待ち時間は変わらないが、I/O待ちがある実際のAPIならasyncioと組み合わせて並列実行できる。
エラーハンドリング
実運用では必ずエラーが発生する。最低限の対策を入れておく。
import time
def run_agent_with_retry(user_message: str, max_retries: int = 3) -> str:
for attempt in range(max_retries):
try:
return run_agent(user_message)
except anthropic.APIStatusError as e:
if e.status_code == 529: # Overloaded
wait_time = 2 ** attempt # 指数バックオフ
print(f"APIが混雑しています。{wait_time}秒後にリトライします...")
time.sleep(wait_time)
elif e.status_code == 400:
# リクエストの問題(メッセージ形式のミス等)はリトライしない
raise
else:
raise
raise Exception("最大リトライ回数に達しました")
よくあるエラーと対処:
| エラーコード | 原因 | 対処 |
|---|---|---|
| 400 | メッセージ形式のミス | tool_use_idの対応・contentの型を確認 |
| 429 | レートリミット | 指数バックオフでリトライ |
| 529 | API混雑 | 指数バックオフでリトライ |
| 500 | Anthropic側の問題 | リトライ |
コスト管理
エージェントループはLLMを複数回呼び出すので、コストが積み上がる。把握しておかないと意図しない請求になる。
def run_agent_with_cost_tracking(user_message: str) -> tuple[str, dict]:
"""コストを追跡しながらエージェントを実行"""
total_input_tokens = 0
total_output_tokens = 0
api_call_count = 0
messages = [{"role": "user", "content": user_message}]
while True:
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
tools=tools,
messages=messages
)
# トークン使用量を累積
total_input_tokens += response.usage.input_tokens
total_output_tokens += response.usage.output_tokens
api_call_count += 1
# 以降は通常のエージェントループ...
if response.stop_reason == "end_turn":
result_text = ""
for block in response.content:
if hasattr(block, "text"):
result_text += block.text
# Claude Sonnet 4.5の料金(2026年時点)
# input: $3/1M tokens, output: $15/1M tokens
cost = (total_input_tokens / 1_000_000 * 3) + \
(total_output_tokens / 1_000_000 * 15)
return result_text, {
"input_tokens": total_input_tokens,
"output_tokens": total_output_tokens,
"api_calls": api_call_count,
"estimated_cost_usd": round(cost, 6)
}
# ツール処理(省略)...
break
return "", {}
ツール定義(tools配列)やシステムプロンプトは毎回同じ内容を送ることになるが、Anthropic Prompt Cachingを使うと最初のリクエスト以降はキャッシュヒットして入力トークン料金が90%削減される。リクエスト数が多いエージェントなら導入を検討する価値がある。
実用的なエージェントの例: GitHub Issue要約エージェント
ここまでの実装をベースに、実際に使えるエージェントを作る。
import anthropic
import os
client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
# GitHubのissue情報を取得するツール(実際はGitHub APIを叩く)
tools = [
{
"name": "get_github_issues",
"description": "GitHubリポジトリのopen issueを取得する",
"input_schema": {
"type": "object",
"properties": {
"repo": {
"type": "string",
"description": "オーナー/リポジトリ名(例: anthropics/anthropic-sdk-python)"
},
"limit": {
"type": "integer",
"description": "取得するissueの最大数(デフォルト: 10)"
}
},
"required": ["repo"]
}
},
{
"name": "search_issues",
"description": "issueをキーワードで検索する",
"input_schema": {
"type": "object",
"properties": {
"repo": {"type": "string"},
"keyword": {"type": "string", "description": "検索キーワード"}
},
"required": ["repo", "keyword"]
}
}
]
def get_github_issues(repo: str, limit: int = 10) -> str:
"""GitHub APIを使ってissueを取得(実装例)"""
import urllib.request
url = f"https://api.github.com/repos/{repo}/issues?state=open&per_page={limit}"
req = urllib.request.Request(url, headers={"User-Agent": "Python"})
try:
with urllib.request.urlopen(req) as response:
import json
issues = json.loads(response.read())
return json.dumps([
{"number": i["number"], "title": i["title"], "body": i.get("body", "")[:200]}
for i in issues
], ensure_ascii=False)
except Exception as e:
return f"エラー: {str(e)}"
def search_issues(repo: str, keyword: str) -> str:
return f"'{keyword}'で検索: {get_github_issues(repo)}" # 簡略化
def execute_tool(name: str, input_data: dict) -> str:
if name == "get_github_issues":
return get_github_issues(**input_data)
if name == "search_issues":
return search_issues(**input_data)
return f"Unknown tool: {name}"
# エージェントの実行(前述のrun_agentと同じループ)
result = run_agent("anthropics/anthropic-sdk-pythonの最新10件のissueを要約して、優先度の高そうなものを3つ教えて")
print(result)
フレームワーク(LangChain等)との比較
LangChainやLangGraphを使えばもっと短くかけるのに、なぜ生実装か?
生実装のメリット:
- tool_use_idの管理・メッセージ形式の理解が身につく
- デバッグがしやすい(どのAPIコールで何が起きているか見える)
- フレームワークのバグや制約に縛られない
- 学習コストが低い(Anthropic公式ドキュメントが唯一の参照先)
LangChain等を使う場面:
- RAG(Retrieval-Augmented Generation)を含む複雑なパイプライン
- OpenAI・Claude・Geminiを同一コードで切り替えたい
- エージェントのメモリ管理を標準化したい
最初は生実装で仕組みを理解して、必要に応じてフレームワークに移行するのが個人的な推奨だ。
まとめ
Anthropic SDK × Pythonでのエージェント実装をまとめると:
- ツールを
input_schemaで定義する while Trueでエージェントループを回すstop_reason == "tool_use"のときツールを実行してresultをuserメッセージで返すtool_use_idの対応付けを正確にやるstop_reason == "end_turn"でループを抜ける
これだけ理解すれば、LangChainなしで実用的なエージェントが作れる。コスト管理とエラーハンドリングを加えれば本番でも動く。