Python × AI SDK実践

約5分

PythonとAPIの基礎を理解していて、実際にClaudeをアプリに組み込みたい方

Anthropic SDKとは、Claude APIをPythonから簡単に呼び出すための公式ライブラリです。メッセージの送受信、ストリーミングレスポンス、ツール使用（tool use）など、Claude APIの主要機能をシンプルなコードで実装できます。このページでは、実際に動くコード例を通じて、AI SDKの使い方を体系的に学びます。

セットアップ

SDKのインストール

pip install anthropic

APIキーの設定

APIキーはコードに直接書かず、環境変数で管理します。

# macOS / Linux
export ANTHROPIC_API_KEY="sk-ant-..."

# Windows（PowerShell）
$env:ANTHROPIC_API_KEY = "sk-ant-..."

.env ファイルを使う場合は python-dotenv が便利です：

pip install python-dotenv

# .env ファイル（.gitignoreに追加してください）
ANTHROPIC_API_KEY=sk-ant-...

from dotenv import load_dotenv
load_dotenv()  # .envファイルを読み込む

APIキーを os.environ.get() や .env 経由で読み込むことで、GitHubなどにコードを公開しても秘密情報が漏洩しません。

基本的なメッセージ送受信

最もシンプルなClaudeへの問い合わせです。

import anthropic

client = anthropic.Anthropic()
# ANTHROPIC_API_KEY 環境変数が自動で読み込まれる

message = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Pythonとは何か、3行で説明してください"}
    ]
)

# レスポンスのテキストを取り出す
print(message.content[0].text)

レスポンスの構造を理解する

import anthropic

client = anthropic.Anthropic()

message = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "1+1は？"}]
)

# レスポンスオブジェクトの各フィールド
print(message.id)               # メッセージの一意なID
print(message.model)            # 使用したモデル名
print(message.role)             # "assistant"
print(message.stop_reason)      # "end_turn"（正常終了）

# トークン使用量
print(message.usage.input_tokens)   # 入力トークン数
print(message.usage.output_tokens)  # 出力トークン数

# テキストの取り出し方
text = message.content[0].text
print(text)  # "2です。"

システムプロンプトを使う

import anthropic

client = anthropic.Anthropic()

message = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    system="あなたはPythonの専門家です。コードの質問には、動作する具体的なコード例を必ず含めて回答してください。",
    messages=[
        {"role": "user", "content": "リストの重複を除去するには？"}
    ]
)

print(message.content[0].text)

会話履歴を維持する

import anthropic

client = anthropic.Anthropic()

# 会話履歴を蓄積するリスト
conversation = []

def chat(user_message: str) -> str:
    """会話履歴を維持しながらClaudeと対話する"""
    conversation.append({"role": "user", "content": user_message})

    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        system="あなたは親切なPythonの学習アシスタントです。",
        messages=conversation
    )

    assistant_message = response.content[0].text
    conversation.append({"role": "assistant", "content": assistant_message})

    return assistant_message

# 会話を続ける
print(chat("Pythonのリスト内包表記を教えてください"))
print(chat("もう少し詳しい例を見せてください"))
print(chat("辞書内包表記との違いは？"))

ストリーミングレスポンス

LLMの応答は生成に時間がかかります。ストリーミングを使うと、生成中のテキストをリアルタイムに表示でき、ユーザー体験が向上します。

import anthropic

client = anthropic.Anthropic()

# with ブロックでストリームを管理する
with client.messages.stream(
    model="claude-opus-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "クラウドアーキテクチャの基本を説明してください"}]
) as stream:
    # テキストが生成されるたびに出力する
    for text in stream.text_stream:
        print(text, end="", flush=True)

print()  # 最後に改行

# ストリーム終了後に最終的なメッセージオブジェクトを取得する
final_message = stream.get_final_message()
print(f"\n使用トークン: {final_message.usage.input_tokens} in / {final_message.usage.output_tokens} out")

ストリームのイベントを細かく制御する

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-opus-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "3つのプログラミング言語を比較してください"}]
) as stream:
    for event in stream:
        # イベントの種類に応じて処理する
        if hasattr(event, "type"):
            if event.type == "content_block_start":
                print("[生成開始]")
            elif event.type == "content_block_delta":
                if hasattr(event.delta, "text"):
                    print(event.delta.text, end="", flush=True)
            elif event.type == "content_block_stop":
                print("\n[生成終了]")
            elif event.type == "message_stop":
                print("[メッセージ完了]")

ツール使用（Tool Use / Function Calling）

ツール使用とは、Claudeが外部の関数やAPIを呼び出せる仕組みです。天気情報の取得、データベース検索、計算など、LLMだけでは難しい処理を実装できます。

シンプルなツールの定義と呼び出し

import anthropic
import json

client = anthropic.Anthropic()

# ツールの定義（どんな関数が使えるかClaudeに伝える）
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"]
        }
    }
]

def get_weather(city: str, unit: str = "celsius") -> dict:
    """実際の天気APIを呼び出す（この例ではダミーデータを返す）"""
    # 実際の実装では OpenWeatherMap などのAPIを呼び出す
    weather_data = {
        "Tokyo": {"temp": 22, "condition": "晴れ", "humidity": 60},
        "Osaka": {"temp": 25, "condition": "曇り", "humidity": 70},
    }
    data = weather_data.get(city, {"temp": 20, "condition": "不明", "humidity": 50})

    if unit == "fahrenheit":
        data["temp"] = data["temp"] * 9 / 5 + 32

    return {"city": city, "temperature": data["temp"], "unit": unit,
            "condition": data["condition"], "humidity": data["humidity"]}

def process_tool_call(tool_name: str, tool_input: dict) -> str:
    """ツール名と引数に基づいて実際の関数を呼び出す"""
    if tool_name == "get_weather":
        result = get_weather(**tool_input)
        return json.dumps(result, ensure_ascii=False)
    return json.dumps({"error": "Unknown tool"})

def chat_with_tools(user_message: str) -> str:
    """ツールを使いながらClaudeと対話する"""
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )

        # ツール使用が不要な場合、回答を返す
        if response.stop_reason == "end_turn":
            return response.content[0].text

        # ツール使用が必要な場合
        if response.stop_reason == "tool_use":
            # assistantの応答（ツール呼び出しを含む）を会話に追加
            messages.append({"role": "assistant", "content": response.content})

            # 各ツール呼び出しを処理する
            tool_results = []
            for content_block in response.content:
                if content_block.type == "tool_use":
                    tool_result = process_tool_call(
                        content_block.name,
                        content_block.input
                    )
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": content_block.id,
                        "content": tool_result
                    })

            # ツール結果を会話に追加して継続
            messages.append({"role": "user", "content": tool_results})

# 実行
answer = chat_with_tools("東京と大阪の天気を比較してください")
print(answer)

エラーハンドリングとリトライ

AI APIへの呼び出しでは、レート制限やネットワークエラーが発生することがあります。

import anthropic
import time

client = anthropic.Anthropic()

def call_with_retry(messages: list, max_retries: int = 3) -> str:
    """レート制限に対応した再試行ロジック"""
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model="claude-opus-4-5",
                max_tokens=1024,
                messages=messages
            )
            return response.content[0].text

        except anthropic.RateLimitError:
            if attempt < max_retries - 1:
                wait_seconds = 2 ** attempt  # 指数バックオフ: 1秒, 2秒, 4秒
                print(f"レート制限に達しました。{wait_seconds}秒後に再試行します...")
                time.sleep(wait_seconds)
            else:
                raise

        except anthropic.APIConnectionError:
            print("ネットワーク接続エラーが発生しました")
            raise

        except anthropic.AuthenticationError:
            print("APIキーが無効です。環境変数 ANTHROPIC_API_KEY を確認してください")
            raise

        except anthropic.APIStatusError as e:
            print(f"APIエラー: {e.status_code} - {e.message}")
            raise

    return ""

result = call_with_retry([{"role": "user", "content": "こんにちは"}])
print(result)

ベストプラクティス

APIキー管理

import os
import anthropic

# 環境変数から読み込む（必須）
api_key = os.environ.get("ANTHROPIC_API_KEY")
if not api_key:
    raise ValueError("ANTHROPIC_API_KEY 環境変数が設定されていません")

client = anthropic.Anthropic(api_key=api_key)

max_tokensの適切な設定

# 短い回答が期待される場合（分類、判断など）
response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=256,   # 短めに設定してコストを抑える
    messages=[{"role": "user", "content": "このテキストのsentimentはPositiveですかNegativeですか？一言で答えてください"}]
)

# 長い回答が必要な場合（文書生成、詳細な説明など）
response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=4096,  # 長めに設定する
    messages=[{"role": "user", "content": "Pythonの非同期処理について詳しく解説してください"}]
)

実践：シンプルなQ&A CLIツール

これまでの内容を統合した、実際に動く小さなCLIアプリを作ります。

"""
simple_qa.py - Claudeを使ったシンプルなQ&A CLIツール

使い方:
  python simple_qa.py
"""

import anthropic
import os
import sys

def create_client() -> anthropic.Anthropic:
    """Anthropicクライアントを作成する"""
    api_key = os.environ.get("ANTHROPIC_API_KEY")
    if not api_key:
        print("エラー: ANTHROPIC_API_KEY 環境変数が設定されていません")
        sys.exit(1)
    return anthropic.Anthropic(api_key=api_key)

def ask_claude(client: anthropic.Anthropic, question: str, history: list) -> str:
    """Claudeに質問してストリーミングで回答を返す"""
    history.append({"role": "user", "content": question})

    full_response = ""
    print("\nClaude: ", end="", flush=True)

    with client.messages.stream(
        model="claude-opus-4-5",
        max_tokens=1024,
        system="あなたは親切で知識豊富なアシスタントです。日本語で回答してください。",
        messages=history
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
            full_response += text

    print()  # 改行
    history.append({"role": "assistant", "content": full_response})
    return full_response

def main():
    """メインループ"""
    client = create_client()
    history = []

    print("=== Claude Q&A ツール ===")
    print("質問を入力してください。終了するには 'quit' または 'exit' と入力してください。\n")

    while True:
        try:
            question = input("あなた: ").strip()

            if not question:
                continue

            if question.lower() in ("quit", "exit", "終了"):
                print("終了します。")
                break

            if question.lower() in ("clear", "クリア"):
                history.clear()
                print("会話履歴をクリアしました。\n")
                continue

            ask_claude(client, question, history)
            print()

        except KeyboardInterrupt:
            print("\n\n終了します。")
            break

        except anthropic.RateLimitError:
            print("レート制限に達しました。しばらく待ってから再試行してください。")

        except anthropic.APIConnectionError:
            print("ネットワーク接続エラーが発生しました。接続を確認してください。")

if __name__ == "__main__":
    main()

実行方法：

export ANTHROPIC_API_KEY="sk-ant-..."
python simple_qa.py

まとめ

anthropic.Anthropic() でクライアントを作成し、client.messages.create() でメッセージを送る
APIキーは環境変数から読み込み、コードに直接書かない
client.messages.stream() でストリーミングレスポンスを実現する
ツール使用（tool use）でClaudeに外部関数を呼び出させられる
stop_reason と content の型を確認してレスポンスを処理する
anthropic.RateLimitError などの例外を適切にハンドリングする

よくある質問

Q: claude-opus-4-5以外のモデルはどう使いますか？

A: model パラメータを変更するだけです。高速・安価なモデル（claude-haiku-4-5など）を使うと、レイテンシを下げたりコストを削減できます。Anthropicのモデル一覧を参照してください。

Q: max_tokensに設定した分だけコストがかかりますか？

A: いいえ。コストは実際に生成されたトークン数に基づきます。max_tokensは「上限」であり、回答が短ければその分だけの課金になります。

Q: ストリーミングとそうでない場合でコストは変わりますか？

A: 変わりません。ストリーミングは表示のタイミングが異なるだけで、生成するトークン数は同じです。

Q: tool useでエラーになった場合、どうデバッグすればいいですか？

A: response.content の中身を print(response.content) で確認し、tool_use タイプのブロックに期待通りの name と input が含まれているか確認します。input_schema の定義が間違っている場合、Claudeが正しいJSON引数を生成できないことがあります。

参考文献

Python Software Foundation, The Python Tutorial
OpenAI, OpenAI API documentation

クイズ

パッケージ管理入門

PythonでのJSON処理