Claude APIでストリーミングレスポンスを実装する方法！PythonでリアルタイムAI出力を表示するチャットアプリ実践ガイド

「Claude APIを使ってチャットアプリを作ったけど、回答が出るまでの待ち時間がつらい…」そんな経験、ありませんか？

通常のAPIリクエストだと、AIが文章を全部生成し終わってから一気に返ってくる仕組みなので、長い回答だと数秒〜十数秒も画面が真っ白なままになってしまいます。これ、ユーザー体験としてはかなり辛いですよね。

実はClaude APIにはストリーミング（streaming）という機能があって、ChatGPTのように文字が少しずつリアルタイムで表示される動作を簡単に実装できます。今回はPythonでストリーミングレスポンスを実装する方法を、実際のコードを動かしながら一緒にマスターしていきましょう！🚀

Contents

ストリーミングって何が違うの？
事前準備を確認しておきましょう
基本のストリーミング実装
最終的なレスポンス全体も取得したい場合
会話履歴を持つチャットアプリへ応用する
ストリーミング実装でよくある落とし穴
まとめ

ストリーミングって何が違うの？

streaming data chat / Photo by Sanket Mishra via Pexels

まず、通常のリクエストとストリーミングの違いをざっくりイメージしてみましょう。

通常のAPIリクエスト（非ストリーミング）

AIが全文を生成し終わるまで待機 → 完成したテキストをまとめて受け取る
料理に例えると「料理が全部できてから運ばれてくる」感じ

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

AIが文字を生成するたびに、少しずつデータが送られてくる
料理に例えると「できた料理から順番にテーブルへ出てくる」感じ

ユーザーから見ると体感の待ち時間がぐっと短くなるので、特にチャットアプリやAIアシスタント系のUIでは必須の実装テクニックです。

事前準備を確認しておきましょう

今回の実装に必要なものはこちらです。

Python 3.8以上
anthropicライブラリ（pip install anthropic）
Claude APIキー（環境変数 ANTHROPIC_API_KEY に設定済み）

APIキーの取得や環境変数の設定がまだの方は、先にそちらを済ませてから進めてみてください。

基本のストリーミング実装

まずはシンプルなストリーミングの基本形から見ていきましょう。ポイントは stream=True を渡す代わりに client.messages.stream() を使うことです。

import anthropic

# クライアントを初期化（APIキーは環境変数から自動取得）
client = anthropic.Anthropic()

def stream_response(user_message: str):
    """ストリーミングでClaudeの回答をリアルタイム表示する"""
    print("Claude: ", end="", flush=True)

    # stream() コンテキストマネージャーを使う
    with client.messages.stream(
        model="claude-opus-4-5",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": user_message}
        ]
    ) as stream:
        # text_stream で1トークンずつ受け取る
        for text in stream.text_stream:
            print(text, end="", flush=True)  # 改行なしで連続表示

    print()  # 最後に改行を入れる

# 試してみよう！
stream_response("Pythonのストリーミング処理について100文字で説明してください")

ポイントをまとめるとこんな感じです👇

client.messages.stream() を with ブロックで使う
stream.text_stream をfor文で回すと、テキストが少しずつ届く
print(text, end="", flush=True) の flush=True が重要！バッファをすぐに吐き出す設定です

最終的なレスポンス全体も取得したい場合

ストリーミング中に表示しつつ、最終的な完全なテキストも変数に保存したい場面があります。そのときは stream.get_final_message() を使います。

import anthropic

client = anthropic.Anthropic()

def stream_and_collect(user_message: str) -> str:
    """ストリーミング表示しながら完全な回答も返す"""
    print("Claude: ", end="", flush=True)
    collected_text = ""

    with client.messages.stream(
        model="claude-opus-4-5",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": user_message}
        ]
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
            collected_text += text  # テキストを手動で結合する方法

    print()

    # あるいは get_final_message() でまとめて取得する方法も
    # final_msg = stream.get_final_message()
    # collected_text = final_msg.content[0].text

    return collected_text

result = stream_and_collect("Pythonの良いところを3つ挙げてください")
print(f"\n--- 保存されたテキスト（{len(result)}文字）---")
print(result[:50] + "...")

これでリアルタイム表示とテキストの保存を同時にこなせます。ログ保存やデータベースへの書き込みをしたい場合に便利ですよ。

会話履歴を持つチャットアプリへ応用する

実際のチャットアプリでは、過去の会話を記憶しながらストリーミングしたいですよね。会話履歴リストにメッセージを積み上げていく方法と組み合わせると、本格的なチャットボットになります。

import anthropic

client = anthropic.Anthropic()

def chat_with_streaming():
    """会話履歴を保持しながらストリーミングチャットを行う"""
    conversation_history = []  # 会話履歴リスト

    print("=== Claudeとストリーミングチャット ===")
    print("終了するには 'quit' と入力してください\n")

    while True:
        # ユーザー入力を受け付ける
        user_input = input("あなた: ").strip()

        if user_input.lower() == "quit":
            print("チャットを終了します。")
            break

        if not user_input:
            continue

        # 会話履歴にユーザーメッセージを追加
        conversation_history.append({
            "role": "user",
            "content": user_input
        })

        # ストリーミングでレスポンスを表示
        print("Claude: ", end="", flush=True)
        assistant_response = ""

        with client.messages.stream(
            model="claude-opus-4-5",
            max_tokens=1024,
            system="あなたは親切なAIアシスタントです。日本語で回答してください。",
            messages=conversation_history  # 履歴ごと渡す
        ) as stream:
            for text in stream.text_stream:
                print(text, end="", flush=True)
                assistant_response += text

        print("\n")

        # Claudeの回答を履歴に追加
        conversation_history.append({
            "role": "assistant",
            "content": assistant_response
        })

# 実行
chat_with_streaming()

ここが重要です✅