Claude API 高度なツール使用：Tool Search・Programmatic Tool Calling・Tool Use Examples で大規模エージェントを構築する

# ❌ 非効率なアプローチ（ツール数が増えるとトークンが爆発的に増加）
response = client.messages.create(
    model=MODEL,
    max_tokens=4096,
    tools=[tool_1, tool_2, ..., tool_100],  # 全100ツールを毎回送信
    messages=[{"role": "user", "content": user_query}]
)

import anthropic
import json
 
client = anthropic.Anthropic()
 
# ── ツールカタログ定義（大規模環境を想定）──
TOOL_CATALOG = {
    "get_weather": {
        "name": "get_weather",
        "description": "指定された都市の現在の天気情報を取得する",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "都市名（例: Tokyo, Osaka）"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"}
            },
            "required": ["city"]
        }
    },
    "search_database": {
        "name": "search_database",
        "description": "製品データベースを検索して関連情報を返す",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "検索クエリ"},
                "limit": {"type": "integer", "description": "返す結果の最大数", "default": 10}
            },
            "required": ["query"]
        }
    },
    "send_email": {
        "name": "send_email",
        "description": "指定の宛先にメールを送信する",
        "input_schema": {
            "type": "object",
            "properties": {
                "to": {"type": "string"},
                "subject": {"type": "string"},
                "body": {"type": "string"}
            },
            "required": ["to", "subject", "body"]
        }
    },
    # ... 実際には100以上のツールが存在する想定
}
 
def tool_search_handler(query: str, limit: int = 5) -> list[dict]:
    """
    BM25 または正規表現でツールカタログを検索する関数
    本番環境では Elasticsearch や pgvector を使うことを推奨
    """
    results = []
    query_lower = query.lower()
 
    for tool_name, tool_def in TOOL_CATALOG.items():
        # シンプルなキーワードマッチング（本番では意味検索を推奨）
        score = 0
        if query_lower in tool_def["description"].lower():
            score += 2
        if any(word in tool_def["name"] for word in query_lower.split()):
            score += 1
 
        if score > 0:
            results.append({"tool": tool_def, "score": score})
 
    # スコア順にソートして上位 N 件を返す
    results.sort(key=lambda x: x["score"], reverse=True)
    return [r["tool"] for r in results[:limit]]
 
def run_agent_with_tool_search(user_query: str) -> str:
    """
    Tool Search Tool を使ったエージェントループ
    """
    # Tool Search Tool の定義
    tool_search_tool = {
        "name": "tool_search",
        "description": "利用可能なツールカタログを検索して関連ツールを発見する。大量のツールがある場合に使用。",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "どんなツールを探しているか（例: 天気を取得, データベースを検索）"
                },
                "limit": {
                    "type": "integer",
                    "description": "返すツール数の上限",
                    "default": 5
                }
            },
            "required": ["query"]
        }
    }
 
    messages = [{"role": "user", "content": user_query}]
    available_tools = [tool_search_tool]  # 最初は Tool Search Tool だけ提供
    loaded_tools = {}  # 動的にロードされたツールを追跡
 
    max_iterations = 10
    for _ in range(max_iterations):
        response = client.messages.create(
            model="claude-opus-4-6-20260205",
            max_tokens=4096,
            tools=available_tools,
            messages=messages
        )
 
        if response.stop_reason == "end_turn":
            # 最終テキストレスポンスを返す
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return "完了"
 
        # ツール呼び出しを処理
        tool_results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
 
            tool_name = block.name
            tool_input = block.input
 
            if tool_name == "tool_search":
                # ツール検索を実行して新しいツールをロード
                found_tools = tool_search_handler(
                    tool_input["query"],
                    tool_input.get("limit", 5)
                )
 
                # 見つかったツールを available_tools に追加
                for tool in found_tools:
                    if tool["name"] not in loaded_tools:
                        loaded_tools[tool["name"]] = tool
                        available_tools.append(tool)
 
                result = f"{len(found_tools)} 個のツールを発見してロードしました: {[t['name'] for t in found_tools]}"
 
            elif tool_name in loaded_tools:
                # 実際のツール実行（本番では外部APIやDBに接続）
                result = execute_tool(tool_name, tool_input)
 
            else:
                result = f"ツール '{tool_name}' が見つかりません"
 
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": str(result)
            })
 
        # メッセージ履歴を更新
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})
 
    return "最大イテレーション数に達しました"
 
def execute_tool(tool_name: str, tool_input: dict) -> Any:
    """ツールを実際に実行する（本番では外部APIに接続）"""
    if tool_name == "get_weather":
        return {"city": tool_input["city"], "temp": 22, "condition": "晴れ"}
    elif tool_name == "search_database":
        return {"results": [f"製品{i}" for i in range(tool_input.get("limit", 3))]}
    elif tool_name == "send_email":
        return {"status": "sent", "message_id": "msg_001"}
    return {"error": "未実装のツール"}
 
# 実行例
if __name__ == "__main__":
    result = run_agent_with_tool_search("東京の天気を調べて、製品データベースを検索してください")
    print(result)
# 期待する出力例:
# 東京の天気は22°C、晴れです。
# データベースの検索結果: 製品0, 製品1, 製品2 が見つかりました。

import anthropic
from anthropic.types import ToolResultBlockParam
import json
 
client = anthropic.Anthropic()
 
def run_programmatic_tool_calling(task: str) -> str:
    """
    Programmatic Tool Calling を使った複雑なデータ処理ワークフロー
 
    Claude がコードを書いてツールを内部から呼び出し、
    中間結果はコンテキストに含めずに最終結果のみを返す。
    """
 
    # ツール定義（関数として実装）
    tools = [
        {
            "name": "fetch_sales_data",
            "description": "指定期間の売上データをCSV形式で取得する",
            "input_schema": {
                "type": "object",
                "properties": {
                    "start_date": {"type": "string", "description": "開始日 (YYYY-MM-DD)"},
                    "end_date": {"type": "string", "description": "終了日 (YYYY-MM-DD)"},
                    "region": {"type": "string", "description": "地域（all, north, south, east, west）", "default": "all"}
                },
                "required": ["start_date", "end_date"]
            }
        },
        {
            "name": "calculate_statistics",
            "description": "数値リストの統計（平均、中央値、標準偏差）を計算する",
            "input_schema": {
                "type": "object",
                "properties": {
                    "values": {"type": "array", "items": {"type": "number"}, "description": "数値のリスト"},
                    "metrics": {"type": "array", "items": {"type": "string"}, "description": "計算する指標"}
                },
                "required": ["values"]
            }
        },
        {
            "name": "generate_report",
            "description": "分析結果からレポートを生成してSlackに送信する",
            "input_schema": {
                "type": "object",
                "properties": {
                    "title": {"type": "string"},
                    "summary": {"type": "string"},
                    "data": {"type": "object"}
                },
                "required": ["title", "summary"]
            }
        }
    ]
 
    messages = [{"role": "user", "content": task}]
 
    while True:
        response = client.messages.create(
            model="claude-opus-4-6-20260205",
            max_tokens=8192,
            tools=tools,
            messages=messages,
            # Programmatic Tool Calling では code_execution も有効化
            # これにより Claude がコード内でツールを呼び出せる
        )
 
        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":
            break
 
        # ツール呼び出しを処理（並列実行対応）
        tool_results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
 
            result = dispatch_tool(block.name, block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps(result, ensure_ascii=False)
            })
 
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})
 
    return "処理完了"
 
def dispatch_tool(name: str, inputs: dict) -> dict:
    """ツール実行ディスパッチャー（本番では各サービスに接続）"""
    import statistics
    import random
 
    if name == "fetch_sales_data":
        # 本番では実際のDBやAPIから取得
        data = [{"date": f"2026-03-{i:02d}", "amount": random.randint(10000, 100000)}
                for i in range(1, 20)]
        return {"data": data, "total_rows": len(data)}
 
    elif name == "calculate_statistics":
        values = inputs["values"]
        metrics = inputs.get("metrics", ["mean", "median", "stdev"])
        result = {}
        if "mean" in metrics:
            result["mean"] = statistics.mean(values)
        if "median" in metrics:
            result["median"] = statistics.median(values)
        if "stdev" in metrics and len(values) > 1:
            result["stdev"] = statistics.stdev(values)
        return result
 
    elif name == "generate_report":
        # 本番では Slack API や Email に送信
        print(f"📊 レポート生成: {inputs['title']}")
        return {"status": "sent", "channel": "#sales-reports"}
 
    return {"error": f"未知のツール: {name}"}
 
# 実行例
if __name__ == "__main__":
    result = run_programmatic_tool_calling(
        "2026年3月1日〜19日の売上データを取得して統計を計算し、"
        "結果をレポートにまとめてSlackに送信してください"
    )
    print(result)
# 期待する出力例:
# 3月の売上分析が完了しました。
# 平均売上: ¥55,234, 中央値: ¥52,000, 標準偏差: ¥18,432
# レポートを #sales-reports に送信しました。

import anthropic
 
client = anthropic.Anthropic()
 
def create_tool_with_examples() -> dict:
    """
    Tool Use Examples を含むツール定義の作成
    input_examples フィールドに最小・部分・完全仕様のパターンを定義
    """
    return {
        "name": "create_chart",
        "description": "データからグラフを生成してファイルに保存する。"
                       "折れ線グラフ、棒グラフ、円グラフに対応。",
        "input_schema": {
            "type": "object",
            "properties": {
                "chart_type": {
                    "type": "string",
                    "enum": ["line", "bar", "pie"],
                    "description": "グラフの種類"
                },
                "data": {
                    "type": "array",
                    "description": "グラフデータ。各要素は {label: string, value: number} 形式",
                    "items": {
                        "type": "object",
                        "properties": {
                            "label": {"type": "string"},
                            "value": {"type": "number"}
                        }
                    }
                },
                "title": {"type": "string", "description": "グラフのタイトル"},
                "output_path": {"type": "string", "description": "出力ファイルパス（省略可）"},
                "options": {
                    "type": "object",
                    "description": "追加オプション（色、サイズ等）",
                    "properties": {
                        "colors": {"type": "array", "items": {"type": "string"}},
                        "width": {"type": "integer"},
                        "height": {"type": "integer"},
                        "legend": {"type": "boolean"}
                    }
                }
            },
            "required": ["chart_type", "data"]
        },
        # ── Tool Use Examples: 最小・部分・完全 の3パターン ──
        "input_examples": [
            {
                # 最小仕様（必須パラメータのみ）
                "description": "シンプルな棒グラフ",
                "input": {
                    "chart_type": "bar",
                    "data": [
                        {"label": "Q1", "value": 150},
                        {"label": "Q2", "value": 230},
                        {"label": "Q3", "value": 190}
                    ]
                }
            },
            {
                # 部分仕様（タイトルあり、オプションなし）
                "description": "タイトル付き折れ線グラフ",
                "input": {
                    "chart_type": "line",
                    "data": [
                        {"label": "Jan", "value": 100},
                        {"label": "Feb", "value": 120},
                        {"label": "Mar", "value": 90}
                    ],
                    "title": "月次売上推移"
                }
            },
            {
                # 完全仕様（すべてのオプションを含む）
                "description": "カスタマイズされた円グラフ",
                "input": {
                    "chart_type": "pie",
                    "data": [
                        {"label": "製品A", "value": 40},
                        {"label": "製品B", "value": 35},
                        {"label": "製品C", "value": 25}
                    ],
                    "title": "製品別シェア（2026年Q1）",
                    "output_path": "/reports/q1_share.png",
                    "options": {
                        "colors": ["#FF6B6B", "#4ECDC4", "#45B7D1"],
                        "width": 800,
                        "height": 600,
                        "legend": True
                    }
                }
            }
        ]
    }
 
def run_with_tool_examples(user_request: str) -> str:
    """Tool Use Examples を含むツールでエージェントを実行"""
    chart_tool = create_tool_with_examples()
 
    response = client.messages.create(
        model="claude-opus-4-6-20260205",
        max_tokens=4096,
        tools=[chart_tool],
        messages=[{"role": "user", "content": user_request}]
    )
 
    # ツール呼び出しを処理
    for block in response.content:
        if block.type == "tool_use" and block.name == "create_chart":
            print(f"✅ グラフ生成: {block.input}")
            # 実際のグラフ生成処理（matplotlib 等）
            return f"グラフを生成しました: {block.input.get('title', '無題')}"
 
    # テキスト応答
    for block in response.content:
        if hasattr(block, "text"):
            return block.text
 
    return "処理完了"
 
# 実行例
if __name__ == "__main__":
    result = run_with_tool_examples(
        "2026年第1四半期の売上データ（製品A:40%, 製品B:35%, 製品C:25%）を"
        "カスタムカラーの円グラフで可視化して、/reports/q1_share.png に保存してください"
    )
    print(result)
# 期待する出力例:
# ✅ グラフ生成: {'chart_type': 'pie', 'data': [...], 'title': '2026年Q1 製品別売上シェア',
#                'output_path': '/reports/q1_share.png', 'options': {'colors': [...], ...}}
# グラフを生成しました: 2026年Q1 製品別売上シェア

class ProductionAgentWithAdvancedTools:
    """
    本番環境向けエージェント：3つの高度ツール機能を統合
    - Tool Search で動的ツール検索（トークン85%削減）
    - Programmatic Tool Calling で中間処理（レイテンシ37%改善）
    - Tool Use Examples で精度向上（72%→90%）
    """
 
    def __init__(self, tool_catalog: dict):
        self.client = anthropic.Anthropic()
        self.tool_catalog = tool_catalog
        self.loaded_tools = {}
        self.model = "claude-opus-4-6-20260205"
 
    def run(self, user_query: str, max_iterations: int = 15) -> str:
        messages = [{"role": "user", "content": user_query}]
 
        # Tool Search Tool を最初から提供
        available_tools = [self._get_tool_search_definition()]
 
        for iteration in range(max_iterations):
            response = self.client.messages.create(
                model=self.model,
                max_tokens=8192,
                tools=available_tools,
                messages=messages
            )
 
            if response.stop_reason == "end_turn":
                return self._extract_text(response)
 
            tool_results = self._process_tool_calls(
                response.content,
                available_tools
            )
 
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
 
        return "最大イテレーション到達"
 
    def _process_tool_calls(self, content, available_tools: list) -> list:
        results = []
        for block in content:
            if block.type != "tool_use":
                continue
 
            if block.name == "tool_search":
                # ツールを動的ロード
                found = self._search_and_load_tools(
                    block.input["query"],
                    available_tools
                )
                result = f"ロード完了: {[t['name'] for t in found]}"
            else:
                result = self._execute_tool(block.name, block.input)
 
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps(result, ensure_ascii=False)
            })
        return results
 
    def _search_and_load_tools(self, query: str, available_tools: list) -> list:
        found = []
        for name, tool in self.tool_catalog.items():
            if query.lower() in tool["description"].lower():
                if name not in self.loaded_tools:
                    self.loaded_tools[name] = tool
                    available_tools.append(tool)
                    found.append(tool)
        return found
 
    def _execute_tool(self, name: str, inputs: dict) -> Any:
        # 実際のツール実行ロジック
        if name in self.tool_catalog:
            return {"result": f"{name} を実行しました", "inputs": inputs}
        return {"error": f"ツール '{name}' が見つかりません"}
 
    def _get_tool_search_definition(self) -> dict:
        return {
            "name": "tool_search",
            "description": "利用可能なツールを検索してロードする",
            "input_schema": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "検索クエリ"}
                },
                "required": ["query"]
            }
        }
 
    def _extract_text(self, response) -> str:
        for block in response.content:
            if hasattr(block, "text"):
                return block.text
        return "完了"

# 意味検索ベースの Tool Search（本番推奨）
# pip install sentence-transformers faiss-cpu
 
from sentence_transformers import SentenceTransformer
import faiss
import numpy as np
 
class SemanticToolSearch:
    def __init__(self, tool_catalog: dict):
        self.model = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
        self.tools = list(tool_catalog.values())
 
        # ツール説明文をベクトル化してインデックス構築
        descriptions = [t["description"] for t in self.tools]
        embeddings = self.model.encode(descriptions)
 
        self.index = faiss.IndexFlatL2(embeddings.shape[1])
        self.index.add(embeddings.astype(np.float32))
 
    def search(self, query: str, k: int = 5) -> list[dict]:
        query_vector = self.model.encode([query]).astype(np.float32)
        distances, indices = self.index.search(query_vector, k)
        return [self.tools[i] for i in indices[0] if distances[0][list(indices[0]).index(i)] < 1.5]

# ❌ 間違い：ツール呼び出しの後に tool_result を返さない
messages.append({"role": "assistant", "content": response.content})
# ここで直接 user メッセージを追加するとエラー
 
# ✅ 正解：必ず tool_result を返す
tool_results = [{"type": "tool_result", "tool_use_id": block.id, "content": "..."}
                for block in response.content if block.type == "tool_use"]
messages.append({"role": "user", "content": tool_results})

# 検索クエリが具体的すぎる場合に発生
# ❌ 具体的すぎるクエリ
tool_search_handler("東京の天気を摂氏で取得する関数")
 
# ✅ 汎用的なクエリで検索
tool_search_handler("天気")  # 短いキーワードの方が効果的

# 大量のツール結果をそのまま返すとコンテキストが膨らむ
# ✅ 結果を要約してから返す
def truncate_result(result: dict, max_chars: int = 500) -> str:
    result_str = json.dumps(result, ensure_ascii=False)
    if len(result_str) > max_chars:
        return result_str[:max_chars] + "...[省略]"
    return result_str

import anthropic
import json
 
client = anthropic.Anthropic()
 
# ツール定義
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"]
        }
    }
]
 
messages = [{"role": "user", "content": "東京の天気を教えて"}]
 
# Step 1: 初回リクエスト
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=messages
)
 
# Step 2: stop_reasonを確認
if response.stop_reason == "tool_use":
    # tool_useブロックを取り出す
    tool_use_block = next(
        block for block in response.content
        if block.type == "tool_use"
    )
    tool_name = tool_use_block.name
    tool_input = tool_use_block.input
    tool_use_id = tool_use_block.id
    
    # Step 3: 実際にツールを実行
    # （実際にはここで外部APIなどを呼び出す）
    result = {"temperature": "18°C", "condition": "晴れ", "humidity": "55%"}
    
    # Step 4: messagesにassistantとtool_resultを追加
    messages.append({"role": "assistant", "content": response.content})
    messages.append({
        "role": "user",
        "content": [
            {
                "type": "tool_result",
                "tool_use_id": tool_use_id,
                "content": json.dumps(result, ensure_ascii=False)
            }
        ]
    })
    
    # Step 5: 最終回答を取得
    final_response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=tools,
        messages=messages
    )
    
    print(final_response.content[0].text)

# ❌ 悪い例 — 何をするツールか不明確
{
    "name": "search",
    "description": "検索します",
    ...
}
 
# ✅ 良い例 — いつ使うべきかが明確
{
    "name": "search_product_database",
    "description": "ユーザーが商品名・型番・カテゴリで製品を検索したいとき、"
                   "または在庫・価格・仕様について質問したときに呼び出す。"
                   "Claudeが知らない具体的な商品情報が必要な場合に使用する。",
    ...
}

# 特定のツールを必ず使わせる
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "tool", "name": "get_weather"},  # このツールを必ず使う
    messages=messages
)
 
# いずれかのツールを必ず使わせる
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "any"},  # ツールを必ず使う（どれでもよい）
    messages=messages
)

# ❌ よくある間違い①：tool_resultをassistantロールで返している
messages.append({
    "role": "assistant",  # ← ここが間違い！tool_resultはuserロールで渡す
    "content": [{"type": "tool_result", ...}]
})
 
# ❌ よくある間違い②：tool_use_idを省略している
messages.append({
    "role": "user",
    "content": [
        {
            "type": "tool_result",
            # "tool_use_id": tool_use_id  ← これがないと紐付けできない！
            "content": "結果のテキスト"
        }
    ]
})
 
# ❌ よくある間違い③：assistantのレスポンスをmessagesに追加していない
# （tool_resultだけを追加して前のassistantメッセージを省いている）
messages.append({
    "role": "user",
    "content": [{"type": "tool_result", ...}]
})
# この前にmessages.append({"role": "assistant", "content": response.content})が必要
 
# ✅ 正しい形式
messages.append({"role": "assistant", "content": response.content})  # 必須！
messages.append({
    "role": "user",
    "content": [
        {
            "type": "tool_result",
            "tool_use_id": tool_use_block.id,   # 必須！
            "content": json.dumps(result)         # 文字列またはcontent配列
        }
    ]
})

# ❌ requiredがないと、Claudeは全フィールドが任意と解釈する可能性がある
{
    "type": "object",
    "properties": {
        "city": {"type": "string"},
        "unit": {"type": "string"}
    }
    # "required": ["city"]  ← これがないと cityが渡されないことがある
}
 
# ✅ 必須パラメータは明示する
{
    "type": "object",
    "properties": {
        "city": {"type": "string", "description": "都市名"},
        "unit": {
            "type": "string",
            "enum": ["celsius", "fahrenheit"],
            "default": "celsius"
        }
    },
    "required": ["city"]
}

def execute_tool_safely(tool_name: str, tool_input: dict) -> str:
    """ツール実行前に入力を検証する"""
    
    if tool_name == "calculate_price":
        # Claudeが文字列で渡してくることがある
        quantity = tool_input.get("quantity")
        if isinstance(quantity, str):
            try:
                quantity = int(quantity)
            except ValueError:
                return json.dumps({"error": "quantityは整数値を指定してください"})
        
        unit_price = tool_input.get("unit_price", 0)
        return json.dumps({"total": quantity * unit_price})
    
    return json.dumps({"error": f"未知のツール: {tool_name}"})

def process_response_with_parallel_tools(response, messages, tools):
    """並列ツール呼び出しを処理する"""
    
    if response.stop_reason != "tool_use":
        return response
    
    # すべてのtool_useブロックを取り出す
    tool_use_blocks = [
        block for block in response.content
        if block.type == "tool_use"
    ]
    
    # tool_resultを並列実行（実際のアプリではasyncioやThreadPoolExecutorを使う）
    tool_results = []
    for tool_block in tool_use_blocks:
        result = execute_tool_safely(tool_block.name, tool_block.input)
        tool_results.append({
            "type": "tool_result",
            "tool_use_id": tool_block.id,  # 各ツールのIDに対応させる
            "content": result
        })
    
    # assistantメッセージとすべてのtool_resultを追加
    messages.append({"role": "assistant", "content": response.content})
    messages.append({
        "role": "user",
        "content": tool_results  # 全ての結果をまとめて返す
    })
    
    # 再度Claudeに送信
    return client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        tools=tools,
        messages=messages
    )

def handle_tool_execution(tool_use_block):
    """ツール実行とエラーハンドリング"""
    try:
        result = call_external_api(tool_use_block.input)
        return {
            "type": "tool_result",
            "tool_use_id": tool_use_block.id,
            "content": json.dumps(result)
        }
    except ExternalAPIError as e:
        # エラーをClaudeに伝える — is_error: trueを使う
        return {
            "type": "tool_result",
            "tool_use_id": tool_use_block.id,
            "content": f"APIエラーが発生しました: {str(e)}。ステータスコード: {e.status_code}",
            "is_error": True  # Claudeにエラーであることを伝える
        }
    except Exception as e:
        return {
            "type": "tool_result",
            "tool_use_id": tool_use_block.id,
            "content": f"予期しないエラー: {str(e)}",
            "is_error": True
        }

import anthropic
import json
 
client = anthropic.Anthropic()
 
def stream_with_tool_use(messages, tools):
    """ストリーミングでツール使用を処理する"""
    
    input_json_accumulator = {}  # ツールごとの入力JSONを積み上げる
    content_blocks = []
    
    with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=tools,
        messages=messages
    ) as stream:
        for event in stream:
            if event.type == "content_block_start":
                if event.content_block.type == "tool_use":
                    # 新しいtool_useブロックの開始
                    input_json_accumulator[event.index] = ""
                    content_blocks.append({
                        "type": "tool_use",
                        "id": event.content_block.id,
                        "name": event.content_block.name,
                        "input": {}
                    })
            
            elif event.type == "content_block_delta":
                if event.delta.type == "input_json_delta":
                    # JSON入力を積み上げる
                    input_json_accumulator[event.index] += event.delta.partial_json
            
            elif event.type == "content_block_stop":
                # ブロック完了 — JSONをパース
                if event.index in input_json_accumulator:
                    try:
                        content_blocks[event.index]["input"] = json.loads(
                            input_json_accumulator[event.index]
                        )
                    except json.JSONDecodeError:
                        content_blocks[event.index]["input"] = {}
        
        final_message = stream.get_final_message()
        return final_message, content_blocks

import time
import anthropic
from anthropic import APIStatusError, APIConnectionError
 
def robust_tool_call(messages, tools, max_retries=3):
    """リトライロジック付きのツール呼び出し"""
    
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                tools=tools,
                messages=messages
            )
            return response
        
        except APIStatusError as e:
            if e.status_code == 429:  # レートリミット
                wait_time = 2 ** attempt  # 指数バックオフ: 1, 2, 4秒
                print(f"レートリミット到達。{wait_time}秒後にリトライ... (試行 {attempt + 1}/{max_retries})")
                time.sleep(wait_time)
            elif e.status_code in [500, 503]:  # サーバーエラー
                wait_time = 5 * (attempt + 1)
                print(f"サーバーエラー。{wait_time}秒後にリトライ...")
                time.sleep(wait_time)
            else:
                raise  # 4xx系（定義ミスなど）はリトライしない
        
        except APIConnectionError:
            wait_time = 3 * (attempt + 1)
            print(f"接続エラー。{wait_time}秒後にリトライ...")
            time.sleep(wait_time)
    
    raise Exception(f"{max_retries}回のリトライがすべて失敗しました")

import json
 
def debug_tool_interaction(response):
    """ツールインタラクションをデバッグ出力する"""
    
    print(f"=== Claude Response Debug ===")
    print(f"stop_reason: {response.stop_reason}")
    print(f"usage: input={response.usage.input_tokens} output={response.usage.output_tokens}")
    
    for i, block in enumerate(response.content):
        print(f"\n[Block {i}] type={block.type}")
        if block.type == "text":
            print(f"  text: {block.text[:200]}...")
        elif block.type == "tool_use":
            print(f"  tool: {block.name}")
            print(f"  id: {block.id}")
            print(f"  input: {json.dumps(block.input, ensure_ascii=False, indent=2)}")
 
# 使い方
response = client.messages.create(...)
debug_tool_interaction(response)