Stream phản hồi theo thời gian thực

Mặc định, Agent SDK trả về các object AssistantMessage hoàn chỉnh sau khi Claude sinh xong mỗi phản hồi. Để nhận cập nhật theo từng phần khi text và tool call đang được sinh ra, bật partial message streaming bằng cách đặt include_partial_messages (Python) hay includePartialMessages (TypeScript) thành true trong tùy chọn của bạn.

Mẹo

Trang này nói về output streaming (nhận token theo thời gian thực). Với input mode (cách bạn gửi message), xem Gửi message tới agent. Bạn cũng có thể stream phản hồi qua Agent SDK bằng CLI.

Bật streaming output

Để bật streaming, đặt include_partial_messages (Python) hay includePartialMessages (TypeScript) thành true trong tùy chọn của bạn. Việc này khiến SDK trả về các message StreamEvent chứa raw API event khi chúng đến, thêm vào bên cạnh AssistantMessage và ResultMessage thông thường.

Code của bạn sau đó cần:

1. Kiểm tra type của mỗi message để phân biệt StreamEvent với các loại message khác
2. Với StreamEvent, trích trường event và kiểm tra type của nó
3. Tìm event content_block_delta nơi delta.type là text_delta, chứa các chunk text thực sự

Ví dụ dưới đây bật streaming và in các chunk text khi chúng đến. Chú ý các bước kiểm tra type lồng nhau: đầu tiên cho StreamEvent, rồi content_block_delta, rồi 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);
      }
    }
  }
}

Tài liệu tham khảo StreamEvent

Khi partial messages được bật, bạn nhận các raw streaming event của Claude API bọc trong một object. Kiểu này có tên khác nhau ở mỗi SDK:

• Python: StreamEvent (import từ claude_agent_sdk.types)
• TypeScript: SDKPartialAssistantMessage với type: 'stream_event'

Cả hai đều chứa raw Claude API event, không phải text đã tích luỹ. Bạn cần tự trích và tích luỹ text delta. Đây là cấu trúc của mỗi kiểu:

@dataclass
class StreamEvent:
    uuid: str  # Định danh duy nhất cho event này
    session_id: str  # Định danh session
    event: dict[str, Any]  # Raw stream event của Claude API
    parent_tool_use_id: str | None  # Luôn là None

type SDKPartialAssistantMessage = {
  type: "stream_event";
  event: BetaRawMessageStreamEvent; // Từ Anthropic SDK
  parent_tool_use_id: string | null;
  uuid: UUID;
  session_id: string;
  ttft_ms?: number; // Thời gian tới token đầu tiên (ms), chỉ có ở message_start event
};

Trường parent_tool_use_id luôn là None trong Python và null trong TypeScript. Stream event chỉ được phát cho session chính; các delta ở cấp token từ subagent không được chuyển tiếp. Để gán output cho một subagent, dùng message hoàn chỉnh, thứ mang parent_tool_use_id. Xem Phát hiện lệnh gọi subagent.

Trường event chứa raw streaming event từ Claude API. Các loại event phổ biến gồm:

Loại event	Mô tả
message_start	Bắt đầu một message mới
content_block_start	Bắt đầu một content block mới (text hoặc tool use)
content_block_delta	Cập nhật theo từng phần cho content
content_block_stop	Kết thúc một content block
message_delta	Cập nhật ở cấp message (stop reason, usage)
message_stop	Kết thúc message

Luồng message

Khi bật partial messages, bạn nhận message theo thứ tự này:

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 - message hoàn chỉnh với toàn bộ content
... tool thực thi ...
... thêm streaming event cho lượt tiếp theo ...
ResultMessage - kết quả cuối cùng

Khi không bật partial messages (include_partial_messages trong Python, includePartialMessages trong TypeScript), bạn nhận mọi loại message trừ StreamEvent. Các loại phổ biến gồm SystemMessage (khởi tạo session), AssistantMessage (phản hồi hoàn chỉnh), ResultMessage (kết quả cuối), và một message ranh giới compact báo khi lịch sử hội thoại được nén (SDKCompactBoundaryMessage trong TypeScript; SystemMessage với subtype "compact_boundary" trong Python).

Stream phản hồi text

Để hiển thị text khi nó được sinh ra, tìm event content_block_delta nơi delta.type là text_delta. Chúng chứa các chunk text theo từng phần. Ví dụ dưới đây in mỗi chunk khi nó đến:

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":
                    # In mỗi chunk text khi nó đến
                    print(delta.get("text", ""), end="", flush=True)

    print()  # Xuống dòng cuối cùng


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(); // Xuống dòng cuối cùng

Stream tool call

Tool call cũng được stream theo từng phần. Bạn có thể theo dõi khi tool bắt đầu, nhận input của nó khi được sinh ra, và thấy khi nó hoàn tất. Ví dụ dưới đây theo dõi tool hiện tại đang được gọi và tích luỹ JSON input khi nó stream về. Nó dùng ba loại event:

• content_block_start: tool bắt đầu
• content_block_delta với input_json_delta: các chunk input đến
• content_block_stop: tool call hoàn tất

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"],
    )

    # Theo dõi tool hiện tại và tích luỹ JSON input của nó
    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":
                # Một tool call mới đang bắt đầu
                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":
                    # Tích luỹ JSON input khi nó stream về
                    chunk = delta.get("partial_json", "")
                    tool_input += chunk
                    print(f"  Input chunk: {chunk}")

            elif event_type == "content_block_stop":
                # Tool call hoàn tất - hiển thị input cuối cùng
                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";

// Theo dõi tool hiện tại và tích luỹ JSON input của nó
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") {
      // Một tool call mới đang bắt đầu
      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") {
        // Tích luỹ JSON input khi nó stream về
        const chunk = event.delta.partial_json;
        toolInput += chunk;
        console.log(`  Input chunk: ${chunk}`);
      }
    } else if (event.type === "content_block_stop") {
      // Tool call hoàn tất - hiển thị input cuối cùng
      if (currentTool) {
        console.log(`Tool ${currentTool} called with: ${toolInput}`);
        currentTool = null;
      }
    }
  }
}

Xây một streaming UI

Ví dụ này kết hợp streaming text và tool thành một UI mạch lạc. Nó theo dõi agent có đang thực thi một tool hay không (dùng cờ in_tool) để hiển thị chỉ báo trạng thái như [Using Read...] khi tool đang chạy. Text stream bình thường khi không trong tool, và tool hoàn tất kích hoạt message “done”. Pattern này hữu ích cho chat interface cần hiển thị tiến trình trong các tác vụ agent nhiều bước.

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"],
    )

    # Theo dõi có đang trong một tool call hay không
    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 đang bắt đầu - hiển thị chỉ báo trạng thái
                    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", {})
                # Chỉ stream text khi không đang thực thi 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 đã xong
                    print(" done", flush=True)
                    in_tool = False

        elif isinstance(message, ResultMessage):
            # Agent đã hoàn tất mọi việc
            print(f"\n\n--- Complete ---")


asyncio.run(streaming_ui())

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

// Theo dõi có đang trong một tool call hay không
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 đang bắt đầu - hiển thị chỉ báo trạng thái
        process.stdout.write(`\n[Using ${event.content_block.name}...]`);
        inTool = true;
      }
    } else if (event.type === "content_block_delta") {
      // Chỉ stream text khi không đang thực thi 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 đã xong
        console.log(" done");
        inTool = false;
      }
    }
  } else if (message.type === "result") {
    // Agent đã hoàn tất mọi việc
    console.log("\n\n--- Complete ---");
  }
}

Giới hạn đã biết

• Structured output: kết quả JSON chỉ xuất hiện trong ResultMessage.structured_output cuối cùng, không phải dưới dạng streaming delta. Xem structured outputs để biết chi tiết.

Bước tiếp theo

Giờ bạn đã có thể stream text và tool call theo thời gian thực, khám phá các chủ đề liên quan sau:

• Interactive vs one-shot queries: chọn giữa các input mode cho use case của bạn
• Structured outputs: nhận phản hồi JSON có kiểu từ agent
• Permissions: kiểm soát tool nào agent được dùng

Nguồn

Dịch và biên soạn từ Stream responses in real-time.