Claude API ストリーミング実装｜失敗しないPython×FastAPI完全ガイド

AIアプリケーション開発をしていると、ユーザーに長めの回答をリアルタイムで表示したいニーズが出てきますよね。Claude APIを使っているなら、なおさら「なぜかAPIからの応答が遅く感じる」「一文字ずつ表示したい」という悩みが生まれます。その理由は、ストリーミング機能を活用していないからかもしれません。

本記事では、Claude APIのストリーミング機能をPythonとFastAPIで実装する方法を、コード例を交えながら完全に解説します。リアルタイムでトークンを受け取り、ユーザーに段階的に回答を表示する仕組みは、モダンなAIアプリケーション開発の必須スキルです。

Claude APIストリーミングとは
前提条件と環境構築
ステップ1：基本的なストリーミング実装
関連記事

Claude APIストリーミングとは

Claude APIのストリーミングは、APIからの応答をトークン単位で分割して、リアルタイムに受け取る機能です。通常の非ストリーミングAPIでは、Claude全体の回答が完成してからまとめて返されます。対してストリーミングでは、1トークン生成されるたびにクライアントへ送信されます。

これにより以下のメリットが生まれます。

ユーザー体験の向上：回答が「生成されている」リアルタイム感が得られる
応答時間の短縮：最初のトークンが届くまでの時間が短くなる
帯域幅の効率化：大容量応答の場合、段階的に処理できる
UIの豊かさ：チャットアプリなど、より自然なUXが実現できる

Claude APIの選択について悩む場合は、Claude API vs OpenAI API徹底比較の記事も参考になります。

前提条件と環境構築

必要なライブラリのインストール

まず、Python環境にClaudeクライアントとFastAPIをインストールしましょう。

pip install anthropic fastapi uvicorn python-dotenv

バージョン確認として、次のコマンドで正しくインストールされたか確認できます。

python -c "import anthropic; print(anthropic.__version__)"

APIキーの設定

AnthropicのコンソールからAPI キーを取得し、環境変数として設定します。

export ANTHROPIC_API_KEY="your-api-key-here"

Pythonコード内では、python-dotenvを使って読み込みます。

from dotenv import load_dotenv
import os

load_dotenv()
api_key = os.getenv("ANTHROPIC_API_KEY")

ステップ1：基本的なストリーミング実装

シンプルなストリーミング例

まずは、最もシンプルなストリーミング実装から始めましょう。以下のコードは、Claude APIのストリーミングを使用して、プロンプトに対する回答をトークン単位で取得します。

from anthropic import Anthropic

client = Anthropic()

def stream_response(user_message: str):
    with client.messages.stream(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": user_message}
        ]
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
        print()

if __name__ == "__main__":
    stream_response("Pyth