Claude API × Python 実践：ツール呼び出しとストリーミングを組み合わせてAIアシスタントを作る

pip install anthropic>=0.40.0

import anthropic
import json
import math
from datetime import datetime
from typing import Any
 
client = anthropic.Anthropic()
 
# ツール定義（JSON スキーマ形式）
TOOLS = [
    {
        "name": "get_current_time",
        "description": "現在の日時を ISO 8601 形式で返します",
        "input_schema": {
            "type": "object",
            "properties": {},
            "required": []
        }
    },
    {
        "name": "calculate",
        "description": "数式を評価して結果を返します。四則演算・冪乗・sqrt などに対応しています",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "評価する数式（例: '2 + 3 * 4', 'sqrt(16)'）"
                }
            },
            "required": ["expression"]
        }
    }
]
 
def execute_tool(name: str, inputs: dict[str, Any]) -> str:
    """ツールを実際に実行して結果文字列を返す"""
    if name == "get_current_time":
        return datetime.now().isoformat()
 
    elif name == "calculate":
        expr = inputs.get("expression", "")
        try:
            # 安全のため、使用できる関数を明示的に制限する
            allowed_names = {
                "sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
                "tan": math.tan, "log": math.log, "abs": abs, "pi": math.pi
            }
            result = eval(expr, {"__builtins__": {}}, allowed_names)
            return str(result)
        except Exception as e:
            return f"計算エラー: {str(e)}"
 
    return f"未知のツール: {name}"

def chat_with_tools(messages: list, model: str = "claude-sonnet-4-6", depth: int = 0) -> str:
    """
    ストリーミング + Tool Use でチャットを実行する。
    ツール呼び出しが発生した場合は深さを増やして再帰的に処理する。
    """
    if depth > 5:
        return "（ツール呼び出しの上限に達しました）"
 
    collected_text = []
    tool_calls = []
    current_tool = None
    current_tool_input_raw = ""
 
    with client.messages.stream(
        model=model,
        max_tokens=2048,
        tools=TOOLS,
        messages=messages
    ) as stream:
        for event in stream:
            event_type = event.type
 
            if event_type == "content_block_start":
                block = event.content_block
                if block.type == "tool_use":
                    # ツール呼び出しブロックが始まった
                    current_tool = {"id": block.id, "name": block.name}
                    current_tool_input_raw = ""
 
            elif event_type == "content_block_delta":
                delta = event.delta
                if delta.type == "text_delta":
                    # テキストをリアルタイムで表示しながら蓄積
                    print(delta.text, end="", flush=True)
                    collected_text.append(delta.text)
                elif delta.type == "input_json_delta":
                    # ツール入力 JSON の断片を蓄積（完成してから parse する）
                    current_tool_input_raw += delta.partial_json
 
            elif event_type == "content_block_stop":
                if current_tool is not None:
                    # JSON が完成したのでパースしてツール呼び出しリストに追加
                    tool_input = json.loads(current_tool_input_raw) if current_tool_input_raw else {}
                    tool_calls.append({**current_tool, "input": tool_input})
                    current_tool = None
                    current_tool_input_raw = ""
 
    # ツール呼び出しがなければテキストをそのまま返す
    if not tool_calls:
        return "".join(collected_text)
 
    # ツール実行 → 結果を次のリクエストに渡す
    print("\n[ツールを実行中...]")
 
    # アシスタントの発言（テキスト + ツール呼び出し）を履歴に追加
    assistant_content = []
    if collected_text:
        assistant_content.append({"type": "text", "text": "".join(collected_text)})
    for tc in tool_calls:
        assistant_content.append({
            "type": "tool_use",
            "id": tc["id"],
            "name": tc["name"],
            "input": tc["input"]
        })
    messages.append({"role": "assistant", "content": assistant_content})
 
    # 各ツールを実行し、結果を tool_result として追加
    tool_results = []
    for tc in tool_calls:
        result = execute_tool(tc["name"], tc["input"])
        print(f"  {tc['name']}({tc['input']}) → {result}")
        tool_results.append({
            "type": "tool_result",
            "tool_use_id": tc["id"],
            "content": result
        })
    messages.append({"role": "user", "content": tool_results})
 
    # ツール結果を踏まえた最終回答を再帰的に取得
    return chat_with_tools(messages, model, depth + 1)

def main():
    print("Claude AIアシスタント（ツール対応・ストリーミング版）")
    print("終了: 'quit' または 'exit'\n")
 
    messages = []
 
    while True:
        user_input = input("You: ").strip()
        if not user_input or user_input.lower() in ("quit", "exit"):
            break
 
        messages.append({"role": "user", "content": user_input})
        print("Assistant: ", end="", flush=True)
        response = chat_with_tools(messages)
        print()
 
        # 最終応答を履歴に追加
        messages.append({"role": "assistant", "content": response})
 
if __name__ == "__main__":
    main()

# 例：ファイル内容を読み込むツール
def read_file_tool(file_path: str) -> str:
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            content = f.read()
        # 長すぎる場合はトリミング
        return content[:5000] if len(content) > 5000 else content
    except FileNotFoundError:
        return f"ファイルが見つかりません: {file_path}"
    except Exception as e:
        return f"読み込みエラー: {str(e)}"