指南

即時串流回應

從 Agent SDK 取得即時回應，隨著文字和工具呼叫串流傳入

預設情況下，Agent SDK 會在 Claude 完成每個回應的生成後，產出完整的 AssistantMessage 物件。若要在文字和工具呼叫生成時接收增量更新，請透過在選項中將 include_partial_messages（Python）或 includePartialMessages（TypeScript）設定為 true 來啟用部分訊息串流。

本頁涵蓋輸出串流（即時接收 token）。關於輸入模式（如何傳送訊息），請參閱傳送訊息給代理。您也可以透過 CLI 使用 Agent SDK 串流回應。

啟用串流輸出

若要啟用串流，請在選項中將 include_partial_messages（Python）或 includePartialMessages（TypeScript）設定為 true。這會使 SDK 在產出一般的 AssistantMessage 和 ResultMessage 之外，還會產出包含原始 API 事件的 StreamEvent 訊息。

您的程式碼需要：

檢查每個訊息的類型，以區分 StreamEvent 和其他訊息類型
對於 StreamEvent，提取 event 欄位並檢查其 type
尋找 delta.type 為 text_delta 的 content_block_delta 事件，其中包含實際的文字片段

以下範例啟用串流並在文字片段到達時列印它們。請注意巢狀的類型檢查：首先檢查 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())

StreamEvent 參考

啟用部分訊息時，您會收到包裝在物件中的原始 Claude API 串流事件。該類型在每個 SDK 中有不同的名稱：

Python：StreamEvent（從 claude_agent_sdk.types 匯入）
TypeScript：SDKPartialAssistantMessage，具有 type: 'stream_event'

兩者都包含原始 Claude API 事件，而非累積的文字。您需要自行提取和累積文字差異。以下是每種類型的結構：

@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

event 欄位包含來自 Claude API 的原始串流事件。常見的事件類型包括：

事件類型	描述
`message_start`	新訊息的開始
`content_block_start`	新內容區塊的開始（文字或工具使用）
`content_block_delta`	內容的增量更新
`content_block_stop`	內容區塊的結束
`message_delta`	訊息層級的更新（停止原因、使用量）
`message_stop`	訊息的結束

訊息流程

啟用部分訊息後，您會按以下順序接收訊息：

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

未啟用部分訊息（Python 中的 include_partial_messages，TypeScript 中的 includePartialMessages）時，您會收到除 StreamEvent 以外的所有訊息類型。常見類型包括 SystemMessage（工作階段初始化）、AssistantMessage（完整回應）、ResultMessage（最終結果）和 CompactBoundaryMessage（表示對話歷史已被壓縮）。

串流文字回應

若要在文字生成時顯示它，請尋找 delta.type 為 text_delta 的 content_block_delta 事件。這些事件包含增量文字片段。以下範例在每個片段到達時列印它：

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())

串流工具呼叫

工具呼叫也會增量串流。您可以追蹤工具何時開始、在其輸入生成時接收它，以及查看它們何時完成。以下範例追蹤目前正在呼叫的工具，並在 JSON 輸入串流傳入時累積它。它使用三種事件類型：

content_block_start：工具開始
content_block_delta 搭配 input_json_delta：輸入片段到達
content_block_stop：工具呼叫完成

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())

建構串流 UI

此範例將文字和工具串流結合成一個連貫的 UI。它追蹤代理目前是否正在執行工具（使用 in_tool 旗標），以在工具執行時顯示狀態指示器，如 [Using Read...]。不在工具中時文字正常串流，工具完成時觸發「done」訊息。此模式適用於需要在多步驟代理任務期間顯示進度的聊天介面。

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())

已知限制

某些 SDK 功能與串流不相容：

延伸思考：當您明確設定 max_thinking_tokens（Python）或 maxThinkingTokens（TypeScript）時，不會發出 StreamEvent 訊息。您只會在每個回合後收到完整的訊息。請注意，SDK 中預設停用思考功能，因此除非您啟用它，否則串流可以正常運作。
結構化輸出：JSON 結果僅出現在最終的 ResultMessage.structured_output 中，而非作為串流差異。詳情請參閱結構化輸出。

後續步驟

現在您可以即時串流文字和工具呼叫了，請探索以下相關主題：

互動式與一次性查詢：為您的使用案例選擇輸入模式
結構化輸出：從代理取得型別化的 JSON 回應
權限：控制代理可以使用哪些工具

Was this page helpful?

指南

即時串流回應

從 Agent SDK 取得即時回應，隨著文字和工具呼叫串流傳入

本頁涵蓋輸出串流（即時接收 token）。關於輸入模式（如何傳送訊息），請參閱傳送訊息給代理。您也可以透過 CLI 使用 Agent SDK 串流回應。

啟用串流輸出

您的程式碼需要：

檢查每個訊息的類型，以區分 StreamEvent 和其他訊息類型
對於 StreamEvent，提取 event 欄位並檢查其 type
尋找 delta.type 為 text_delta 的 content_block_delta 事件，其中包含實際的文字片段

以下範例啟用串流並在文字片段到達時列印它們。請注意巢狀的類型檢查：首先檢查 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())

StreamEvent 參考

啟用部分訊息時，您會收到包裝在物件中的原始 Claude API 串流事件。該類型在每個 SDK 中有不同的名稱：

Python：StreamEvent（從 claude_agent_sdk.types 匯入）
TypeScript：SDKPartialAssistantMessage，具有 type: 'stream_event'

兩者都包含原始 Claude API 事件，而非累積的文字。您需要自行提取和累積文字差異。以下是每種類型的結構：

@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

event 欄位包含來自 Claude API 的原始串流事件。常見的事件類型包括：

事件類型	描述
`message_start`	新訊息的開始
`content_block_start`	新內容區塊的開始（文字或工具使用）
`content_block_delta`	內容的增量更新
`content_block_stop`	內容區塊的結束
`message_delta`	訊息層級的更新（停止原因、使用量）
`message_stop`	訊息的結束

訊息流程

啟用部分訊息後，您會按以下順序接收訊息：

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

串流文字回應

若要在文字生成時顯示它，請尋找 delta.type 為 text_delta 的 content_block_delta 事件。這些事件包含增量文字片段。以下範例在每個片段到達時列印它：

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())

串流工具呼叫

content_block_start：工具開始
content_block_delta 搭配 input_json_delta：輸入片段到達
content_block_stop：工具呼叫完成

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())

建構串流 UI

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())

已知限制

某些 SDK 功能與串流不相容：

延伸思考：當您明確設定 max_thinking_tokens（Python）或 maxThinkingTokens（TypeScript）時，不會發出 StreamEvent 訊息。您只會在每個回合後收到完整的訊息。請注意，SDK 中預設停用思考功能，因此除非您啟用它，否則串流可以正常運作。
結構化輸出：JSON 結果僅出現在最終的 ResultMessage.structured_output 中，而非作為串流差異。詳情請參閱結構化輸出。

後續步驟

現在您可以即時串流文字和工具呼叫了，請探索以下相關主題：

互動式與一次性查詢：為您的使用案例選擇輸入模式
結構化輸出：從代理取得型別化的 JSON 回應
權限：控制代理可以使用哪些工具

Was this page helpful?