Stream responses แบบ real-time

รับ responses แบบ real-time จาก Agent SDK ขณะที่ text และ tool calls กำลัง stream เข้ามา

โดยค่าเริ่มต้น Agent SDK จะ yield AssistantMessage objects ที่สมบูรณ์หลังจาก Claude สร้าง response แต่ละครั้งเสร็จแล้ว หากต้องการรับการอัปเดตแบบ incremental ขณะที่ text และ tool calls กำลังถูกสร้าง ให้เปิดใช้งาน partial message streaming โดยตั้งค่า include_partial_messages (Python) หรือ includePartialMessages (TypeScript) เป็น true ใน options ของคุณ

tip

หน้านี้ครอบคลุม output streaming (รับ tokens แบบ real-time) สำหรับ input modes (วิธีส่งข้อความ) ดู ส่งข้อความไปยัง agents

เปิดใช้งาน Streaming Output

หากต้องการเปิดใช้งาน streaming ให้ตั้งค่า include_partial_messages (Python) หรือ includePartialMessages (TypeScript) เป็น true ใน options ของคุณ สิ่งนี้ทำให้ SDK yield messages StreamEvent ที่มี raw API events ขณะที่มาถึง นอกเหนือจาก AssistantMessage และ ResultMessage ปกติ

โค้ดของคุณจำเป็นต้อง:

1. ตรวจสอบ type ของแต่ละ message เพื่อแยก StreamEvent จาก message types อื่น
2. สำหรับ StreamEvent ให้ดึง field event และตรวจสอบ type ของมัน
3. มองหา content_block_delta events ที่ delta.type เป็น text_delta ซึ่งมี text chunks จริง

ตัวอย่างด้านล่างเปิดใช้งาน streaming และพิมพ์ text chunks ขณะที่มาถึง สังเกต type checks ที่ซ้อนกัน: ก่อนสำหรับ StreamEvent จากนั้น content_block_delta แล้วก็ text_delta:

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio


async def stream_response():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Bash", "Read"],
    )

    async for message in query(prompt="List the files in my project", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    print(delta.get("text", ""), end="", flush=True)


asyncio.run(stream_response())

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "List the files in my project",
  options: {
    includePartialMessages: true,
    allowedTools: ["Bash", "Read"]
  }
})) {
  if (message.type === "stream_event") {
    const event = message.event;
    if (event.type === "content_block_delta") {
      if (event.delta.type === "text_delta") {
        process.stdout.write(event.delta.text);
      }
    }
  }
}

StreamEvent reference

เมื่อเปิดใช้งาน partial messages คุณจะได้รับ raw Claude API streaming events ที่ถูก wrap ใน object ชนิดนี้มีชื่อต่างกันในแต่ละ SDK:

• Python: StreamEvent (import จาก claude_agent_sdk.types)
• TypeScript: SDKPartialAssistantMessage ที่มี type: 'stream_event'

ทั้งคู่มี raw Claude API events ไม่ใช่ข้อความที่สะสม คุณต้องดึงและสะสม text deltas เอง นี่คือโครงสร้างของแต่ละชนิด:

@dataclass
class StreamEvent:
    uuid: str  # Unique identifier for this event
    session_id: str  # Session identifier
    event: dict[str, Any]  # The raw Claude API stream event
    parent_tool_use_id: str | None  # Parent tool ID if from a subagent

type SDKPartialAssistantMessage = {
  type: "stream_event";
  event: BetaRawMessageStreamEvent; // From Anthropic SDK
  parent_tool_use_id: string | null;
  uuid: UUID;
  session_id: string;
  ttft_ms?: number; // Time to first token in ms, present only on message_start events
};

Field event มี raw streaming event จาก Claude API ชนิด event ที่พบบ่อยได้แก่:

Event Type	คำอธิบาย
message_start	เริ่มต้น message ใหม่
content_block_start	เริ่มต้น content block ใหม่ (text หรือ tool use)
content_block_delta	การอัปเดตแบบ incremental ของเนื้อหา
content_block_stop	สิ้นสุด content block
message_delta	การอัปเดตระดับ message (stop reason, usage)
message_stop	สิ้นสุด message

Message flow

เมื่อเปิดใช้งาน partial messages คุณจะได้รับ messages ในลำดับนี้:

StreamEvent (message_start)
StreamEvent (content_block_start) - text block
StreamEvent (content_block_delta) - text chunks...
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - tool_use block
StreamEvent (content_block_delta) - tool input chunks...
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
AssistantMessage - complete message with all content
... tool executes ...
... more streaming events for next turn ...
ResultMessage - final result

Stream text responses

หากต้องการแสดง text ขณะที่ถูกสร้าง ให้มองหา content_block_delta events ที่ delta.type เป็น text_delta ซึ่งมี text chunks แบบ incremental ตัวอย่างด้านล่างพิมพ์แต่ละ chunk ขณะที่มาถึง:

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio


async def stream_text():
    options = ClaudeAgentOptions(include_partial_messages=True)

    async for message in query(prompt="Explain how databases work", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    # Print each text chunk as it arrives
                    print(delta.get("text", ""), end="", flush=True)

    print()  # Final newline


asyncio.run(stream_text())

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "Explain how databases work",
  options: { includePartialMessages: true }
})) {
  if (message.type === "stream_event") {
    const event = message.event;
    if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
      process.stdout.write(event.delta.text);
    }
  }
}

console.log(); // Final newline

Stream tool calls

Tool calls ยัง stream แบบ incremental ด้วย คุณสามารถติดตามเมื่อ tools เริ่มต้น รับ input ขณะที่ถูกสร้าง และดูเมื่อเสร็จสิ้น ตัวอย่างด้านล่างติดตาม tool ปัจจุบันที่กำลังถูกเรียกใช้และสะสม JSON input ขณะที่ stream เข้ามา โดยใช้สาม event types:

• content_block_start: tool เริ่มต้น
• content_block_delta ที่มี input_json_delta: input chunks มาถึง
• content_block_stop: tool call เสร็จสมบูรณ์

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio


async def stream_tool_calls():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash"],
    )

    # Track the current tool and accumulate its input JSON
    current_tool = None
    tool_input = ""

    async for message in query(prompt="Read the README.md file", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

            if event_type == "content_block_start":
                # New tool call is starting
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    current_tool = content_block.get("name")
                    tool_input = ""
                    print(f"Starting tool: {current_tool}")

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "input_json_delta":
                    # Accumulate JSON input as it streams in
                    chunk = delta.get("partial_json", "")
                    tool_input += chunk
                    print(f"  Input chunk: {chunk}")

            elif event_type == "content_block_stop":
                # Tool call complete - show final input
                if current_tool:
                    print(f"Tool {current_tool} called with: {tool_input}")
                    current_tool = None


asyncio.run(stream_tool_calls())

import { query } from "@anthropic-ai/claude-agent-sdk";

// Track the current tool and accumulate its input JSON
let currentTool: string | null = null;
let toolInput = "";

for await (const message of query({
  prompt: "Read the README.md file",
  options: {
    includePartialMessages: true,
    allowedTools: ["Read", "Bash"]
  }
})) {
  if (message.type === "stream_event") {
    const event = message.event;

    if (event.type === "content_block_start") {
      // New tool call is starting
      if (event.content_block.type === "tool_use") {
        currentTool = event.content_block.name;
        toolInput = "";
        console.log(`Starting tool: ${currentTool}`);
      }
    } else if (event.type === "content_block_delta") {
      if (event.delta.type === "input_json_delta") {
        // Accumulate JSON input as it streams in
        const chunk = event.delta.partial_json;
        toolInput += chunk;
        console.log(`  Input chunk: ${chunk}`);
      }
    } else if (event.type === "content_block_stop") {
      // Tool call complete - show final input
      if (currentTool) {
        console.log(`Tool ${currentTool} called with: ${toolInput}`);
        currentTool = null;
      }
    }
  }
}

สร้าง Streaming UI

ตัวอย่างนี้รวม text และ tool streaming เข้าด้วยกันเป็น UI ที่สอดคล้องกัน โดยติดตามว่า agent กำลัง execute tool อยู่หรือไม่ (โดยใช้ flag in_tool) เพื่อแสดง status indicators เช่น [Using Read...] ขณะที่ tools รัน Text จะ stream ตามปกติเมื่อไม่อยู่ใน tool และเมื่อ tool เสร็จสิ้นจะแสดงข้อความ "done" รูปแบบนี้มีประโยชน์สำหรับ chat interfaces ที่ต้องแสดงความคืบหน้าระหว่าง multi-step agent tasks:

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
from claude_agent_sdk.types import StreamEvent
import asyncio
import sys


async def streaming_ui():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash", "Grep"],
    )

    # Track whether we're currently in a tool call
    in_tool = False

    async for message in query(
        prompt="Find all TODO comments in the codebase", options=options
    ):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

            if event_type == "content_block_start":
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    # Tool call is starting - show status indicator
                    tool_name = content_block.get("name")
                    print(f"\n[Using {tool_name}...]", end="", flush=True)
                    in_tool = True

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                # Only stream text when not executing a tool
                if delta.get("type") == "text_delta" and not in_tool:
                    sys.stdout.write(delta.get("text", ""))
                    sys.stdout.flush()

            elif event_type == "content_block_stop":
                if in_tool:
                    # Tool call finished
                    print(" done", flush=True)
                    in_tool = False

        elif isinstance(message, ResultMessage):
            # Agent finished all work
            print(f"\n\n--- Complete ---")


asyncio.run(streaming_ui())

import { query } from "@anthropic-ai/claude-agent-sdk";

// Track whether we're currently in a tool call
let inTool = false;

for await (const message of query({
  prompt: "Find all TODO comments in the codebase",
  options: {
    includePartialMessages: true,
    allowedTools: ["Read", "Bash", "Grep"]
  }
})) {
  if (message.type === "stream_event") {
    const event = message.event;

    if (event.type === "content_block_start") {
      if (event.content_block.type === "tool_use") {
        // Tool call is starting - show status indicator
        process.stdout.write(`\n[Using ${event.content_block.name}...]`);
        inTool = true;
      }
    } else if (event.type === "content_block_delta") {
      // Only stream text when not executing a tool
      if (event.delta.type === "text_delta" && !inTool) {
        process.stdout.write(event.delta.text);
      }
    } else if (event.type === "content_block_stop") {
      if (inTool) {
        // Tool call finished
        console.log(" done");
        inTool = false;
      }
    }
  } else if (message.type === "result") {
    // Agent finished all work
    console.log("\n\n--- Complete ---");
  }
}

ข้อจำกัดที่ทราบ

• Structured output: JSON result จะปรากฏเฉพาะใน ResultMessage.structured_output สุดท้าย ไม่ใช่เป็น streaming deltas ดู structured outputs สำหรับรายละเอียด

ขั้นตอนถัดไป

ตอนนี้ที่คุณสามารถ stream text และ tool calls แบบ real-time ได้แล้ว สำรวจหัวข้อที่เกี่ยวข้องเหล่านี้:

• Interactive vs one-shot queries: เลือกระหว่าง input modes สำหรับ use case ของคุณ
• Structured outputs: รับ typed JSON responses จาก agent
• Permissions: ควบคุม tools ที่ agent สามารถใช้ได้