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

# インストール
# pip install anthropic>=1.50.0
 
import anthropic
import asyncio
import json
from typing import Any
 
# クライアント初期化
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
# 使用するモデル（2026年3月時点の推奨）
MODEL = "claude-opus-4-6-20260205"
 
# 3機能に共通のベータヘッダー（本記事のコードは全てこれを前提とします）
ADVANCED_TOOL_USE_BETA = "advanced-tool-use-2025-11-20"

response = client.beta.messages.create(
    betas=[ADVANCED_TOOL_USE_BETA],
    model=MODEL,
    max_tokens=4096,
    tools=[...],
    messages=[{"role": "user", "content": user_query}],
)

# ❌ 非効率なアプローチ（ツール数が増えるとトークンが爆発的に増加）
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.beta.messages.create(
            betas=[ADVANCED_TOOL_USE_BETA],   # defer_loading を有効にするために必須
            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.beta.messages.create(
            betas=[ADVANCED_TOOL_USE_BETA],   # allowed_callers を有効にするために必須
            model="claude-opus-4-6-20260205",
            max_tokens=8192,
            tools=tools,
            messages=messages,
            # tools には 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.beta.messages.create(
        betas=[ADVANCED_TOOL_USE_BETA],   # input_examples を有効にするために必須
        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 で動的ツール検索（初期コンテキストのトークンを削減）
    - Programmatic Tool Calling で中間処理（中間結果をコンテキスト外へ）
    - Tool Use Examples で入力例を提示（パラメータ誤りを削減）
    """
 
    def __init__(self, tool_catalog: dict):
        # ベータ機能を使うため beta 名前空間で呼び出す（後述の _call を参照）
        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.beta.messages.create(
                betas=[ADVANCED_TOOL_USE_BETA],
                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

def handle_tool_execution(tool_use_block):
    """ツールを実行し、失敗は is_error で Claude に伝える"""
    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:
        return {
            "type": "tool_result",
            "tool_use_id": tool_use_block.id,
            "content": f"APIエラーが発生しました: {e}。ステータスコード: {e.status_code}",
            "is_error": True,
        }
    except Exception as e:
        return {
            "type": "tool_result",
            "tool_use_id": tool_use_block.id,
            "content": f"予期しないエラー: {e}",
            "is_error": True,
        }

def execute_tool_safely(tool_name: str, tool_input: dict) -> str:
    """ツール実行前に入力の型を検証・補正する"""
    if tool_name == "calculate_price":
        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}"})