Claude Streaming API実装ガイド｜リアルタイム応答をアプリに組み込む

message_start
 →content_block_start(type:text)
   →content_block_delta("こん")
   →content_block_delta("にちは")
   →content_block_delta("。Pythonに")
   →content_block_delta("ついて説明")
   →content_block_delta("します。")
 →content_block_stop
→message_delta(stop_reason:end_turn)
→message_stop

importanthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Pythonの非同期処理について説明して"}
    ]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

print()  # 最後に改行

importanthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Pythonの非同期処理について説明して"}
    ]
) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)
        elif event.type == "message_delta":
            print(f"\n--- stop_reason:{event.delta.stop_reason}")
            print(f"--- 出力トークン:{event.usage.output_tokens}")

importanthropic
importasyncio

async defmain():
    client = anthropic.AsyncAnthropic()

    async with client.messages.stream(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": "Pythonの非同期処理について説明して"}
        ]
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)

asyncio.run(main())

importAnthropicfrom"@anthropic-ai/sdk";

constclient=newAnthropic();

asyncfunctionmain(){
 conststream=client.messages.stream({
   model:"claude-sonnet-4-5",
   max_tokens:1024,
   messages:[
     {role:"user",content:"TypeScriptの型システムについて説明して"}
   ]
 });

 stream.on("text",(text)=>{
   process.stdout.write(text);
 });

 constfinalMessage=awaitstream.finalMessage();
 console.log(`\n--- 出力トークン:${finalMessage.usage.output_tokens}`);
}

main();

// Express + SSEの例
importexpressfrom"express";
importAnthropicfrom"@anthropic-ai/sdk";

constapp=express();
constclient=newAnthropic();

app.get("/api/chat",async(req,res)=>{
 res.setHeader("Content-Type","text/event-stream");
 res.setHeader("Cache-Control","no-cache");
 res.setHeader("Connection","keep-alive");

 conststream=client.messages.stream({
   model:"claude-sonnet-4-5",
   max_tokens:1024,
   messages:[
     {role:"user",content:req.query.messageasstring}
   ]
 });

 stream.on("text",(text)=>{
   res.write(`data:${JSON.stringify({text})}\n\n`);
 });

 stream.on("end",()=>{
   res.write("data: [DONE]\n\n");
   res.end();
 });
});

app.listen(3000);

message_start
 →content_block_start(type:text)
   →content_block_delta("天気を調べます")
 →content_block_stop
 →content_block_start(type:tool_use,name:"get_weather")
   →content_block_delta({"location":"東京"}) ←JSONが断片的に届く
 →content_block_stop
→message_delta(stop_reason:tool_use)
→message_stop

importanthropic
importjson

client = anthropic.Anthropic()

tool_input_parts = []

with client.messages.stream(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    tools=[{
        "name": "get_weather",
        "description": "指定した都市の天気を取得する",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "都市名"}
            },
            "required": ["location"]
        }
    }],
    messages=[
        {"role": "user", "content": "東京の天気を教えて"}
    ]
) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)
            elif event.delta.type == "input_json_delta":
                tool_input_parts.append(event.delta.partial_json)

if tool_input_parts:
    tool_input = json.loads("".join(tool_input_parts))
    print(f"\nツール呼び出し:{tool_input}")

importanthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-5",
    max_tokens=8000,
    thinking={
        "type": "enabled",
        "budget_tokens": 5000
    },
    messages=[
        {"role": "user", "content": "量子コンピュータの基本原理を説明して"}
    ]
) as stream:
    for event in stream:
        if event.type == "content_block_start":
            if hasattr(event.content_block, "type"):
                if event.content_block.type == "thinking":
                    print("[思考開始]")
                elif event.content_block.type == "text":
                    print("[回答開始]")
        elif event.type == "content_block_delta":
            if event.delta.type == "thinking_delta":
                print(f"  (思考中:{event.delta.thinking})", end="")
            elif event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)

importanthropic

client = anthropic.Anthropic()

try:
    with client.messages.stream(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": "長い説明をお願い"}
        ]
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
except anthropic.APIConnectionError:
    print("\n[接続エラー: ネットワークを確認してください]")
except anthropic.RateLimitError:
    print("\n[レート制限: しばらく待ってからリトライしてください]")
except anthropic.APIStatusError as e:
    print(f"\n[APIエラー:{e.status_code}{e.message}]")