ツール使用パターン：信頼性の高いエージェント-ツールインターフェースの構築

import anthropic

client = anthropic.Anthropic()

tools = [
    {
        "name": "search_products",
        "description": (
            "キーワードで商品カタログを検索する。"
            "IDと名前と価格を含む一致する商品のリストを返す。"
            "ユーザーが商品を探したりブラウズしたいときに使う。"
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "検索するキーワード"
                },
                "category": {
                    "type": "string",
                    "enum": ["electronics", "clothing", "food", "home", "all"],
                    "description": "フィルタリングする商品カテゴリ。未指定の場合は'all'を使う。"
                },
                "max_results": {
                    "type": "integer",
                    "minimum": 1,
                    "maximum": 20,
                    "description": "返す結果の数。デフォルト：5"
                }
            },
            "required": ["query", "category"]
        }
    }
]

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "1万円以下の電子機器を探して"}]
)

import json
from dataclasses import dataclass, asdict
from typing import Any, Optional


@dataclass
class ToolResult:
    success: bool
    data: Optional[Any] = None
    error: Optional[str] = None

    def to_content(self) -> str:
        return json.dumps(asdict(self), ensure_ascii=False)


def search_products(
    query: str,
    category: str,
    max_results: int = 5,
) -> ToolResult:
    try:
        raw_results = _query_database(query, category, limit=max_results)
        products = [
            {"id": r["product_id"], "name": r["title"], "price": r["price_usd"]}
            for r in raw_results
        ]
        return ToolResult(success=True, data={"products": products, "count": len(products)})
    except ConnectionError as e:
        return ToolResult(success=False, error=f"データベース利用不可：{e}")
    except Exception as e:
        return ToolResult(success=False, error=f"検索失敗：{type(e).__name__}: {e}")


def _query_database(query, category, limit):
    return []

from concurrent.futures import ThreadPoolExecutor, as_completed


TOOL_REGISTRY = {
    "search_products": search_products,
}


def dispatch_tool(name: str, inputs: dict) -> ToolResult:
    handler = TOOL_REGISTRY.get(name)
    if not handler:
        return ToolResult(success=False, error=f"不明なツール：{name}")
    return handler(**inputs)


def process_tool_calls(response: anthropic.types.Message) -> list[dict]:
    tool_uses = [
        block for block in response.content
        if block.type == "tool_use"
    ]
    if not tool_uses:
        return []

    def execute(tool_use):
        result = dispatch_tool(tool_use.name, tool_use.input)
        return {
            "type": "tool_result",
            "tool_use_id": tool_use.id,
            "content": result.to_content(),
        }

    with ThreadPoolExecutor(max_workers=len(tool_uses)) as executor:
        futures = {executor.submit(execute, tu): tu for tu in tool_uses}
        results = []
        for future in as_completed(futures):
            results.append(future.result())

    return results


def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            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":
            tool_results = process_tool_calls(response)
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
        else:
            break

    return ""

import signal


def timeout_handler(signum, frame):
    raise TimeoutError("ツール実行がタイムアウト")


def safe_tool_call(
    name: str,
    inputs: dict,
    timeout_seconds: int = 30,
) -> ToolResult:
    """
    タイムアウト付きでツールを実行し、すべての例外をキャッチする。
    常にToolResultを返す——例外を投げることはない。
    """
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout_seconds)
    try:
        return dispatch_tool(name, inputs)
    except TimeoutError:
        return ToolResult(
            success=False,
            error=f"ツール'{name}'が{timeout_seconds}秒後にタイムアウト"
        )
    except Exception as e:
        return ToolResult(
            success=False,
            error=f"ツール'{name}'が{type(e).__name__}を発生：{e}"
        )
    finally:
        signal.alarm(0)

MAX_TOOL_RESULT_CHARS = 8000


def validate_result(result: ToolResult, expected_keys: list[str]) -> ToolResult:
    if not result.success or not isinstance(result.data, dict):
        return result
    missing = [k for k in expected_keys if k not in result.data]
    if missing:
        return ToolResult(
            success=False,
            error=f"ツールレスポンスに期待されるフィールドが欠けている：{missing}"
        )
    return result


def truncate_result(result: ToolResult) -> ToolResult:
    content = result.to_content()
    if len(content) <= MAX_TOOL_RESULT_CHARS:
        return result

    truncated_data = {
        "truncated": True,
        "chars_omitted": len(content) - MAX_TOOL_RESULT_CHARS,
        "content": content[:MAX_TOOL_RESULT_CHARS],
    }
    return ToolResult(
        success=result.success,
        data=truncated_data,
        error="結果を切り詰めた——コンテキストウィンドウに対して大きすぎる",
    )


def safe_tool_call_validated(
    name: str,
    inputs: dict,
    expected_keys: list[str] | None = None,
) -> ToolResult:
    result = safe_tool_call(name, inputs)
    if expected_keys:
        result = validate_result(result, expected_keys)
    result = truncate_result(result)
    return result