Langsung ke konten utama
Secara default, Agent SDK menghasilkan objek AssistantMessage lengkap setelah Claude selesai menghasilkan setiap respons. Untuk menerima pembaruan inkremental saat teks dan tool calls dihasilkan, aktifkan partial message streaming dengan mengatur include_partial_messages (Python) atau includePartialMessages (TypeScript) ke true dalam opsi Anda.
Halaman ini mencakup output streaming (menerima token secara real-time). Untuk input modes (cara Anda mengirim pesan), lihat Send messages to agents. Anda juga dapat stream responses using the Agent SDK via the CLI.

​
Enable streaming output

Untuk mengaktifkan streaming, atur include_partial_messages (Python) atau includePartialMessages (TypeScript) ke true dalam opsi Anda. Ini menyebabkan SDK menghasilkan pesan StreamEvent yang berisi raw API events saat tiba, selain AssistantMessage dan ResultMessage yang biasa. Kode Anda kemudian perlu:
  1. Memeriksa tipe setiap pesan untuk membedakan StreamEvent dari tipe pesan lainnya
  2. Untuk StreamEvent, ekstrak field event dan periksa type-nya
  3. Cari event content_block_delta di mana delta.type adalah text_delta, yang berisi chunk teks aktual
Contoh di bawah mengaktifkan streaming dan mencetak chunk teks saat tiba. Perhatikan pemeriksaan tipe bersarang: pertama untuk StreamEvent, kemudian untuk content_block_delta, kemudian untuk 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())

​
StreamEvent reference

Ketika partial messages diaktifkan, Anda menerima raw Claude API streaming events yang dibungkus dalam objek. Tipe memiliki nama berbeda di setiap SDK:
  • Python: StreamEvent (import dari claude_agent_sdk.types)
  • TypeScript: SDKPartialAssistantMessage dengan type: 'stream_event'
Keduanya berisi raw Claude API events, bukan teks terakumulasi. Anda perlu mengekstrak dan mengakumulasi text deltas sendiri. Berikut adalah struktur setiap tipe: 
@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
Field event berisi raw streaming event dari Claude API. Tipe event umum meliputi:
Event TypeDescription
message_startAwal pesan baru
content_block_startAwal blok konten baru (teks atau tool use)
content_block_deltaPembaruan inkremental ke konten
content_block_stopAkhir blok konten
message_deltaPembaruan tingkat pesan (stop reason, usage)
message_stopAkhir pesan

​
Message flow

Dengan partial messages diaktifkan, Anda menerima pesan dalam urutan ini: 
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
Tanpa partial messages diaktifkan (include_partial_messages di Python, includePartialMessages di TypeScript), Anda menerima semua tipe pesan kecuali StreamEvent. Tipe umum meliputi SystemMessage (inisialisasi sesi), AssistantMessage (respons lengkap), ResultMessage (hasil akhir), dan pesan batas kompak yang menunjukkan kapan riwayat percakapan dikompres (SDKCompactBoundaryMessage di TypeScript; SystemMessage dengan subtype "compact_boundary" di Python).

​
Stream text responses

Untuk menampilkan teks saat dihasilkan, cari event content_block_delta di mana delta.type adalah text_delta. Ini berisi chunk teks inkremental. Contoh di bawah mencetak setiap chunk saat tiba: 
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())

​
Stream tool calls

Tool calls juga streaming secara inkremental. Anda dapat melacak kapan tools dimulai, menerima input mereka saat dihasilkan, dan melihat kapan mereka selesai. Contoh di bawah melacak tool yang sedang dipanggil dan mengakumulasi input JSON saat streaming masuk. Ini menggunakan tiga tipe event:
  • content_block_start: tool dimulai
  • content_block_delta dengan input_json_delta: chunk input tiba
  • content_block_stop: tool call selesai
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())

​
Build a streaming UI

Contoh ini menggabungkan text dan tool streaming menjadi UI yang kohesif. Ini melacak apakah agent saat ini mengeksekusi tool (menggunakan flag in_tool) untuk menampilkan indikator status seperti [Using Read...] saat tools berjalan. Teks streaming secara normal ketika tidak dalam tool, dan penyelesaian tool memicu pesan “done”. Pola ini berguna untuk antarmuka chat yang perlu menampilkan progress selama tugas agent multi-langkah. 
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())

​
Keterbatasan yang Diketahui

  • Structured output: hasil JSON muncul hanya di ResultMessage.structured_output akhir, bukan sebagai streaming deltas. Lihat structured outputs untuk detail.

​
Next steps

Sekarang bahwa Anda dapat stream teks dan tool calls secara real-time, jelajahi topik terkait ini: