AI エージェントの Tool Use 本番設計：Claude と OpenAI の関数呼び出しを、冪等・安全・可観測に組む

[ユーザー入力]
     │
     ▼
┌─────────────────────────────────────────────┐
│  LLM ──(tool_use: name + input)──▶ アプリ    │
│   ▲                                  │       │
│   │                                  ▼       │
│   └──(tool_result)── アプリがツールを実行   │
└─────────────────────────────────────────────┘
     │ stop_reason が tool_use でなくなったら抜ける
     ▼
[最終応答]

import anthropic

client = anthropic.Anthropic()  # APIキーは環境変数 ANTHROPIC_API_KEY から

TOOLS = [
    {
        "name": "get_product_stock",
        # description は「いつ呼ぶか」まで書くと発火精度が上がる（後述）
        "description": (
            "指定した商品IDの在庫数・価格・納期を社内マスタから取得する。"
            "ユーザーが在庫・価格・納期・適合可否を尋ねたときに必ず呼ぶこと。"
            "推測で答えてはならない。マスタにない値は捏造しない。"
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "product_id": {
                    "type": "string",
                    "description": "商品ID（例: SKU-00123）。形式は SKU- + 5桁数字。",
                },
                "warehouse": {
                    "type": "string",
                    "enum": ["tokyo", "osaka", "fukuoka"],
                    "description": "在庫を確認する倉庫拠点",
                },
            },
            "required": ["product_id"],
            "additionalProperties": False,  # 余計なキーを構造段階で拒否
        },
    }
]

import json
import anthropic

client = anthropic.Anthropic()

MODEL = "claude-opus-4-8"  # 権威ある最新モデルID


# 名前 → 実装関数 のディスパッチテーブル（SRP: ルーティングと実装を分離）
def get_product_stock(product_id: str, warehouse: str = "tokyo") -> dict:
    # ここは「あなたのコード」。後述の検証・冪等化はこの中／手前で行う
    return {"product_id": product_id, "stock": 12, "price_jpy": 49800, "lead_days": 3}


TOOL_IMPL = {"get_product_stock": get_product_stock}


def run_agent(user_text: str, max_turns: int = 6) -> str:
    """ユーザー入力を受け、tool_use ループを回して最終応答テキストを返す。"""
    messages: list[dict] = [{"role": "user", "content": user_text}]

    for _ in range(max_turns):  # 無限ループ防止の上限は必須（飾りではない）
        resp = client.messages.create(
            model=MODEL,
            max_tokens=1024,
            tools=TOOLS,
            messages=messages,
        )
        # アシスタントの応答（tool_use ブロックを含む）をそのまま履歴に積む
        messages.append({"role": "assistant", "content": resp.content})

        if resp.stop_reason != "tool_use":
            # ツール呼び出しが終わった → 最終応答を結合して返す
            return "".join(b.text for b in resp.content if b.type == "text")

        # tool_use ブロックを全件処理し、結果を1つの user メッセージにまとめて返す
        tool_results: list[dict] = []
        for block in resp.content:
            if block.type != "tool_use":
                continue
            try:
                impl = TOOL_IMPL[block.name]      # 未知ツールは KeyError → 下で is_error
                output = impl(**block.input)      # ← 検証前の素通しは危険（第3章で直す）
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(output, ensure_ascii=False),
                })
            except Exception as e:
                # 失敗は is_error で返す → LLM が立て直せる（公式仕様）
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": f"ツール実行エラー: {e}",
                    "is_error": True,
                })

        messages.append({"role": "user", "content": tool_results})

    # 上限到達 = ツールループが収束しなかった。黙って続けず、明示的に止める
    raise RuntimeError("最大ターン数に到達しました（ツールループが収束せず）")

from openai import OpenAI

client = OpenAI()  # APIキーは環境変数 OPENAI_API_KEY から

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_product_stock",
            "description": "指定した商品IDの在庫数・価格・納期を社内マスタから取得する。",
            "parameters": {  # ← Claude の input_schema に相当（JSON Schema）
                "type": "object",
                "properties": {
                    "product_id": {"type": "string", "description": "商品ID"},
                    "warehouse": {
                        "type": "string",
                        "enum": ["tokyo", "osaka", "fukuoka"],
                    },
                },
                "required": ["product_id", "warehouse"],
                "additionalProperties": False,
            },
            "strict": True,  # スキーマ厳守（strict時は全 properties を required + additionalProperties:false）
        },
    }
]

import json

messages = [{"role": "user", "content": "SKU-00123 の東京在庫を教えて"}]

MAX_TURNS = 6
for _ in range(MAX_TURNS):  # Claude 側と同じく上限ガードを必ず置く
    resp = client.chat.completions.create(
        model="gpt-5.5",  # モデル名は利用環境に合わせて
        messages=messages,
        tools=tools,
    )
    msg = resp.choices[0].message
    messages.append(msg)  # アシスタント応答（tool_calls 含む）を履歴に積む

    if not msg.tool_calls:
        print(msg.content)  # 最終応答
        break

    for call in msg.tool_calls:
        args = json.loads(call.function.arguments)   # ← 文字列を必ずパース
        output = TOOL_IMPL[call.function.name](**args)
        messages.append({
            "role": "tool",                          # ← Claude と違い専用ロール
            "tool_call_id": call.id,                 # ← Claude の tool_use_id 相当
            "content": json.dumps(output, ensure_ascii=False),
        })
else:
    raise RuntimeError("最大ターン数に到達しました（OpenAI 側ループ）")

from pydantic import BaseModel, Field, ValidationError
from typing import Literal


class StockQuery(BaseModel):
    """ツール引数の契約。ここを通った値だけが「信用できる入力」になる。"""
    model_config = {"extra": "forbid"}  # 未知フィールドは拒否（余計な注入を弾く）

    product_id: str = Field(pattern=r"^SKU-\d{5}$")  # 形式を厳格に固定
    warehouse: Literal["tokyo", "osaka", "fukuoka"] = "tokyo"


def get_product_stock_safe(raw_input: dict) -> dict:
    # 1) 境界で検証（失敗は LLM に返して立て直させる）
    try:
        args = StockQuery.model_validate(raw_input)
    except ValidationError as e:
        # is_error=True で返す前提の構造化エラー
        return {"_error": f"引数が不正です: {e.errors()}"}

    # 2) ここから先は型が保証された世界。安心して読み取り副作用を起こせる
    return _query_stock_master(args.product_id, args.warehouse)

import hashlib
import json
from functools import wraps


def _idem_key(tool_name: str, args: dict, *, business_key: str | None = None) -> str:
    """業務キーがあればそれを、なければツール名＋正規化引数で決定的キーを作る。"""
    if business_key:
        return hashlib.sha256(f"{tool_name}:{business_key}".encode()).hexdigest()
    payload = json.dumps(args, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(f"{tool_name}:{payload}".encode()).hexdigest()


def idempotent(store):  # store: get(key)->result|None / set(key, result)
    """副作用ツールを冪等化するデコレータ。"""
    def deco(fn):
        @wraps(fn)
        def wrapper(tool_name: str, args: dict, *, business_key: str | None = None):
            key = _idem_key(tool_name, args, business_key=business_key)
            if (cached := store.get(key)) is not None:
                return cached  # 2回目以降は外部APIを叩かない（二重課金・二重発注の回避）
            result = fn(tool_name, args)
            store.set(key, result)  # 実行が成功してから記録（失敗時は次回再実行できる）
            return result
        return wrapper
    return deco

import random
import time
import httpx


def call_with_retry(fn, *, max_attempts=4, base=0.5, timeout=8.0):
    """指数バックオフ＋ジッタ。4xx(入力不正)は即失敗、5xx/タイムアウトのみ再試行。"""
    for attempt in range(1, max_attempts + 1):
        try:
            return fn(timeout=timeout)
        except httpx.HTTPStatusError as e:
            if 400 <= e.response.status_code < 500:
                raise  # 入力不正はリトライしても無駄（fail fast）
            if attempt == max_attempts:
                raise
        except (httpx.TimeoutException, httpx.TransportError):
            if attempt == max_attempts:
                raise
        # 0.5s, 1s, 2s... にジッタを足してリトライ嵐（thundering herd）を避ける
        time.sleep(base * (2 ** (attempt - 1)) + random.uniform(0, 0.3))

import hashlib
import json
import logging
import time
from contextlib import contextmanager

log = logging.getLogger("agent.tool")


def _arg_fingerprint(args: dict) -> str:
    """引数そのものではなく、内容ハッシュを残す（PII・機密を生で出さない）。"""
    payload = json.dumps(args, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(payload.encode()).hexdigest()[:16]


@contextmanager
def trace_tool(tool_name: str, args: dict, trace_id: str):
    """1ツール呼び出し = 1スパン。所要時間・成否・引数ハッシュを構造化ログに残す。"""
    start = time.monotonic()
    record = {
        "trace_id": trace_id,
        "tool": tool_name,
        "arg_hash": _arg_fingerprint(args),  # 生引数は残さない
        "status": "ok",
    }
    try:
        yield record
    except Exception as e:
        record["status"] = "error"
        record["error_type"] = type(e).__name__  # メッセージ本文ではなく型（PII混入を避ける）
        raise
    finally:
        record["duration_ms"] = round((time.monotonic() - start) * 1000, 1)
        log.info("tool_call", extra=record)  # 構造化ログ／OpenTelemetry へ

def execute_traced(tool_name: str, args: dict, trace_id: str) -> dict:
    with trace_tool(tool_name, args, trace_id):
        return TOOL_IMPL[tool_name](args)  # 検証・冪等化を通った実行関数

import re

ALLOWED_RECIPIENTS = {"support@example.com", "sales@example.com"}
DESTRUCTIVE_TOOLS = {"delete_order", "issue_refund", "send_external_email"}
SKU_RE = re.compile(r"^SKU-\d{5}$")


def _sanitize_args(tool_name: str, args: dict) -> dict | None:
    """ポリシー段階での軽量サニタイズ。不正なら None を返して実行を止める。"""
    if "product_id" in args and not SKU_RE.fullmatch(str(args["product_id"])):
        return None  # 形式外の product_id は通さない
    # メールアドレスのように外部送信先になる値は、必ず allowlist と突き合わせる
    return args


def guard_and_execute(tool_name: str, args: dict, *, approver=None) -> dict:
    # 1) 引数サニタイズ：形式チェックは Pydantic 検証（第3章）の前段ガードとして二重化
    cleaned = _sanitize_args(tool_name, args)
    if cleaned is None:
        return {"_error": "引数の形式が不正なため実行を拒否しました"}

    # 2) 許可リスト：LLM が何を言おうと、許可された宛先以外には送らない（deny ではなく allow）
    if tool_name == "send_external_email":
        if cleaned.get("to") not in ALLOWED_RECIPIENTS:
            return {"_error": f"宛先 {cleaned.get('to')} は許可されていません"}

    # 3) 破壊的操作は人間確認ゲート（無人運用でも、ここだけは人を挟む設計が成立する）
    if tool_name in DESTRUCTIVE_TOOLS:
        if approver is None or not approver.confirm(tool_name, cleaned):
            return {"_error": "破壊的操作には承認が必要です。実行を保留しました。"}

    return TOOL_IMPL[tool_name](cleaned)

import pytest
from pydantic import ValidationError


def test_stock_query_rejects_bad_sku():
    with pytest.raises(ValidationError):
        StockQuery.model_validate({"product_id": "INVALID"})  # 形式違反


def test_stock_query_forbids_extra_key():
    with pytest.raises(ValidationError):
        # extra=forbid: LLM が余計なキーを足してきても拒否する
        StockQuery.model_validate({"product_id": "SKU-00123", "evil": "x"})


def test_idempotent_skips_second_call():
    store, calls = {}, []

    @idempotent(store={"get": store.get, "set": store.__setitem__})
    def charge(tool_name, args):
        calls.append(args)
        return {"ok": True}

    charge("issue_refund", {"order": "A-1"}, business_key="A-1")
    charge("issue_refund", {"order": "A-1"}, business_key="A-1")
    assert len(calls) == 1  # 2回目は外部を叩かない