Streaming Input

Tổng quan

Claude Agent SDK hỗ trợ hai chế độ input riêng biệt để tương tác với agent:

• Streaming Input Mode (Mặc định & khuyến nghị) - một phiên tương tác, sống lâu (persistent)
• Single Message Input - query một lần, dùng session state và resuming

Hướng dẫn này giải thích khác biệt, lợi ích, và use case cho mỗi chế độ để giúp bạn chọn cách tiếp cận đúng cho ứng dụng của mình.

Streaming Input Mode (khuyến nghị)

Streaming input mode là cách được ưu tiên để dùng Claude Agent SDK. Nó cung cấp quyền truy cập đầy đủ vào năng lực của agent và cho phép trải nghiệm tương tác phong phú.

Nó cho phép agent hoạt động như một process sống lâu, nhận input từ người dùng, xử lý gián đoạn, hiển thị yêu cầu phê duyệt, và quản lý session.

Nó hoạt động thế nào

sequenceDiagram
    participant App as Ứng dụng của bạn
    participant Agent as Claude Agent
    participant Tools as Tools/Hooks
    participant FS as Môi trường/<br/>File System

    App->>Agent: Khởi tạo với AsyncGenerator
    activate Agent

    App->>Agent: Yield Message 1
    Agent->>Tools: Thực thi tool
    Tools->>FS: Đọc file
    FS-->>Tools: Nội dung file
    Tools->>FS: Ghi/sửa file
    FS-->>Tools: Success/Error
    Agent-->>App: Stream phản hồi từng phần
    Agent-->>App: Stream thêm nội dung...
    Agent->>App: Hoàn tất Message 1

    App->>Agent: Yield Message 2 + Image
    Agent->>Tools: Xử lý image & thực thi
    Tools->>FS: Truy cập filesystem
    FS-->>Tools: Kết quả thao tác
    Agent-->>App: Stream phản hồi 2

    App->>Agent: Queue Message 3
    App->>Agent: Interrupt/Cancel
    Agent->>App: Xử lý gián đoạn

    Note over App,Agent: Session vẫn sống
    Note over Tools,FS: Trạng thái filesystem<br/>được duy trì liên tục

    deactivate Agent

Lợi ích

• Upload ảnh: đính kèm ảnh trực tiếp vào message để phân tích và hiểu hình ảnh
• Queued messages: gửi nhiều message xử lý tuần tự, có thể ngắt giữa chừng
• Tích hợp tool: truy cập đầy đủ mọi tool và MCP server tuỳ biến trong suốt session
• Phản hồi thời gian thực: thấy phản hồi khi chúng được tạo ra, không chỉ kết quả cuối
• Duy trì context: giữ context hội thoại tự nhiên qua nhiều lượt

Ví dụ implementation

Các ví dụ này đọc một file ảnh tên diagram.png trong thư mục làm việc. Tạo một file ở đó trước, hoặc đổi tên file để trỏ tới ảnh của riêng bạn.

TypeScript:

import { query, type SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
import { readFile } from "fs/promises";

async function* generateMessages(): AsyncGenerator<SDKUserMessage> {
  // Message đầu tiên
  yield {
    type: "user",
    message: {
      role: "user",
      content: "Analyze this codebase for security issues"
    },
    parent_tool_use_id: null
  };

  // Chờ điều kiện hoặc input người dùng
  await new Promise((resolve) => setTimeout(resolve, 2000));

  // Follow-up kèm ảnh
  yield {
    type: "user",
    message: {
      role: "user",
      content: [
        {
          type: "text",
          text: "Review this architecture diagram"
        },
        {
          type: "image",
          source: {
            type: "base64",
            media_type: "image/png",
            data: await readFile("diagram.png", "base64")
          }
        }
      ]
    },
    parent_tool_use_id: null
  };
}

// Xử lý phản hồi streaming
for await (const message of query({
  prompt: generateMessages(),
  options: {
    maxTurns: 10,
    allowedTools: ["Read", "Grep"]
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

Python:

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
)
import asyncio
import base64


async def streaming_analysis():
    async def message_generator():
        # Message đầu tiên
        yield {
            "type": "user",
            "message": {
                "role": "user",
                "content": "Analyze this codebase for security issues",
            },
        }

        # Chờ điều kiện
        await asyncio.sleep(2)

        # Follow-up kèm ảnh
        with open("diagram.png", "rb") as f:
            image_data = base64.b64encode(f.read()).decode()

        yield {
            "type": "user",
            "message": {
                "role": "user",
                "content": [
                    {"type": "text", "text": "Review this architecture diagram"},
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/png",
                            "data": image_data,
                        },
                    },
                ],
            },
        }

    # Dùng ClaudeSDKClient cho streaming input
    options = ClaudeAgentOptions(max_turns=10, allowed_tools=["Read", "Grep"])

    async with ClaudeSDKClient(options) as client:
        # Gửi streaming input
        await client.query(message_generator())

        # Xử lý phản hồi
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(block.text)


asyncio.run(streaming_analysis())

Khi bạn chạy ví dụ, bản TypeScript in ra mỗi phản hồi ngay khi hoàn tất. Vòng lặp receive_response() của bản Python kết thúc ở result message đầu tiên, nên nó in ra phân tích bảo mật; để đọc cả hai phản hồi, dùng một cặp query() và receive_response() cho mỗi message.

Ghi chú

Trong TypeScript SDK, nếu message generator của bạn throw lỗi, ví dụ khi một file nó đọc bị thiếu, stream kết thúc với lỗi ghi Claude Code process aborted by user thay vì lỗi gốc, nên hãy kiểm tra code bên trong generator của bạn trước khi thấy message này. Lỗi cũng có thể đi kèm một dòng dài, minified chứa SDK source đóng gói, nên đọc tới cuối output để thấy nội dung lỗi.

Trong Python SDK, một exception từ generator được log ở mức debug và session bị treo mà không raise, nên nếu một streaming session bị treo không có output, bật debug logging và kiểm tra generator của bạn.

Single Message Input

Single message input đơn giản hơn nhưng hạn chế hơn.

Khi nào dùng Single Message Input

Dùng single message input khi:

• Bạn cần phản hồi một lần
• Bạn không cần đính kèm ảnh hay các phương thức kiểm soát giữa session
• Bạn cần vận hành trong môi trường stateless, như một lambda function

Hạn chế

Chú ý

Single message input mode không hỗ trợ:

• Đính kèm ảnh trực tiếp trong message
• Queue message động
• Ngắt theo thời gian thực
• Hội thoại nhiều lượt tự nhiên

Nếu một query kết thúc với result lỗi, như error_max_turns, một lời gọi query() single message raise lỗi kèm text thất bại sau khi yield result message cuối cùng, nên bọc vòng lặp trong try block nếu code của bạn cần tiếp tục. Xem Xử lý kết quả để biết các result subtype.

Ví dụ implementation

TypeScript:

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

// Query một lần đơn giản
// query() throw lỗi sau một result lỗi, như error_max_turns
try {
  for await (const message of query({
    prompt: "Explain the authentication flow",
    options: {
      maxTurns: 5,
      allowedTools: ["Read", "Grep"]
    }
  })) {
    if (message.type === "result" && message.subtype === "success") {
      console.log(message.result);
    }
  }
} catch (error) {
  console.error(`Query failed: ${error}`);
}

// Tiếp tục hội thoại với quản lý session
try {
  for await (const message of query({
    prompt: "Now explain the authorization process",
    options: {
      continue: true,
      maxTurns: 5
    }
  })) {
    if (message.type === "result" && message.subtype === "success") {
      console.log(message.result);
    }
  }
} catch (error) {
  console.error(`Query failed: ${error}`);
}

Python:

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
import asyncio


async def single_message_example():
    # Query một lần đơn giản dùng hàm query()
    # query() raise lỗi sau một result lỗi, như error_max_turns
    try:
        async for message in query(
            prompt="Explain the authentication flow",
            options=ClaudeAgentOptions(max_turns=5, allowed_tools=["Read", "Grep"]),
        ):
            if isinstance(message, ResultMessage) and message.subtype == "success":
                print(message.result)
    # SDK raise một Exception thuần cho result lỗi, nên bắt Exception ở đây
    except Exception as e:
        print(f"Query failed: {e}")

    # Tiếp tục hội thoại với quản lý session
    try:
        async for message in query(
            prompt="Now explain the authorization process",
            options=ClaudeAgentOptions(continue_conversation=True, max_turns=5),
        ):
            if isinstance(message, ResultMessage) and message.subtype == "success":
                print(message.result)
    except Exception as e:
        print(f"Query failed: {e}")


asyncio.run(single_message_example())

Khi bạn chạy ví dụ, mỗi query in ra text kết quả cuối cùng của nó: đầu tiên là giải thích authentication, rồi giải thích authorization.

Nguồn

Dịch và biên soạn từ Streaming Input.