AI Agent รู้อยู่แล้วว่าทำอะไร — Streaming (Claude Agent SDK Part 3)

Rick Hightower — Part 3 ของซีรีส์ 14 part. Non-streaming agent = silent pause 15 วินาที (UX failure). Streaming output = 3 nested check + 30 บรรทัด → live narration. Streaming input = architectural choice ก่อน ship. 2 setting ที่ปิด streaming เงียบ ๆ

Your user stares at a frozen cursor for fifteen seconds, wondering if the agent crashed. It did not. It just never told anyone what it was doing.

2 Stream ต่างกัน — อย่าสับสน

Streaming output = รับ token ตอน generate. SDK ให้ text เป็น fragment ตอน Claude produce — render Claude reasoning + tool call realtime

Streaming input = วิธีส่ง message. เปิด session ต่อเนื่อง feed message + image + interrupt over time

2 อันอิสระ. Stream output จาก one-shot query โดยไม่ต้องแตะ streaming input ได้ — บทความนี้ทำงี้ก่อน (ง่ายกว่า)

เปิด Streaming Output

Default — SDK ให้ AssistantMessage complete หลัง Claude generate จบ. 1 option เปลี่ยน: include_partial_messages (Python) / includePartialMessages (TS) — SDK yield StreamEvent เพิ่ม + complete message ปกติ

Raw = catch: StreamEvent ไม่ให้ text ที่ accumulated เรียบร้อย — wrap raw API streaming event — งานเลือก piece ที่ต้องการ = ของคุณ

Event Sequence

Turn เดียวไม่มาเป็น message เดียว — มาเป็น sequence ของ fine-grained event: message_start → ต่อ content chunk: content_block_start → run ของ content_block_delta → content_block_stop → message_delta → message_stop. หลังนั้นยังได้ complete AssistantMessage แบบ non-streaming → tool execute → turn ถัดไป event เริ่ม → สุดท้าย ResultMessage ปิด run

2 event type เดียวที่ต้องใช้ build UI:

• content_block_start — block ใหม่เริ่ม. ถ้า tool_use = cue ว่า tool กำลังจะรัน อ่าน name ที่นี่
• content_block_delta — incremental update. delta.type = text_delta = chunk ของ Claude text. input_json_delta = chunk ของ tool input argument

ที่เหลือ (message_start, stop, message_delta) = scaffolding ignore ได้

Streaming Text — Version ต่ำสุด

Pattern = 3 nested check. Message = StreamEvent มั้ย? Event = content_block_delta มั้ย? Delta = text_delta มั้ย? Yes ทั้ง 3 = print chunk

# Python
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 buggy-shop", 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())

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

for await (const message of query({
  prompt: "List the files in buggy-shop",
  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);
      }
    }
  }
}

Detail สำคัญ: end="", flush=True (Python) + process.stdout.write (TS) — เขียนไม่มี trailing newline + push terminal ทันที. ไม่มี flush = buffer + dump ตอนเต็ม. มี flush = text ออกทีละ character

Terminal UI จริง

Print raw text = demo. Interface ใช้งาน = ต้องบอกว่า agent หยุดพูดเริ่มทำ. Bash รัน 10 วินาทีที่ print อะไรไม่ออก = เหมือน hang. UI เงียบตอน tool call = rebuild silent pause ใหม่

Track state 1 อัน — อยู่ใน tool call มั้ย. ใช้ block event สลับ 2 mode: “narrate text” + “show tool status line”

Logic เล็ก. content_block_start ประกาศ tool_use = print [Using Read...] + set in_tool = True. ตอน flag on — suppress text delta (delta ที่มา = tool input argument ไม่ใช่สิ่งที่ user อ่าน). content_block_stop fire ตอน in_tool = print “done” + flip flag off. Text stream ปกติเมื่อไม่อยู่ใน tool

# Python
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"],
    )
    in_tool = False  # are we currently inside a tool call?
    async for message in query(
        prompt="Find the failing test in buggy-shop and explain the bug.",
        options=options,
    ):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")
            if event_type == "content_block_start":
                block = event.get("content_block", {})
                if block.get("type") == "tool_use":
                    print(f"\n[Using {block.get('name')}...]", end="", flush=True)
                    in_tool = True
            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                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:
                    print(" done", flush=True)
                    in_tool = False
        elif isinstance(message, ResultMessage):
            print("\n\n--- Complete ---")

asyncio.run(streaming_ui())

Run แล้ว = live narration. Claude reasoning stream เป็น text. Tool call ประกาศ [Using Bash...] ปิดด้วย done. Run จบ --- Complete ---. 30 บรรทัด + state = 1 boolean

ทำไม Thinking ปิด + อะไรจะไม่ Stream

2 limitation ต้องรู้ก่อน ship — symptom เหมือนกัน: streaming หยุด work เงียบ code ดู ok

Extended thinking: เปิด thinking = SDK หยุด emit StreamEvent — ได้ complete message เท่านั้น. Thinking = off default → streaming work out of box. เสียตอน turn thinking on

max_thinking_tokens/maxThinkingTokens = วิธี legacy. SDK ปัจจุบันใช้ structured thinking config: thinking={"type": "enabled", "budget_tokens": N}. Turn on = disable streaming

Structured output: ขอ typed JSON = JSON ไม่มาเป็น streaming delta. มา end เดียวใน ResultMessage.structured_output. Half-formed JSON mid-generation = parse ไม่ได้

Gotcha: เปิด thinking + “streaming broke” = ไม่ใช่ bug. Thinking-token option ปิด partial message by design. Drop thinking หรือ accept per-turn complete message. 2 feature ไม่ coexist. SDK ไม่ warn

Stream อีกอัน: Keep Session Open

ทั้งหมดข้างบน = stream output จาก one-shot query. Input side มี 2 mode — distinction shape application ได้

Single message input = one-shot query(). Prompt → agent loop รัน → read result. Perfect สำหรับ stateless env (Lambda, CI job) — ง่ายสุด. ไม่ support: image attachment ตรง, queue message dynamic, real-time interrupt, hook integration

Streaming input mode = persistent interactive session. SDK แนะนำเป็น default สำหรับ rich app. แทน prompt เดียว — provide async generator ที่ yield message over time. Agent รันเป็น long-lived process — รับ input, handle interrupt, surface permission request, keep filesystem + conversation state ข้าม message

Mode นี้ unlock attach image ใน follow-up + queue หลาย message process sequence + interrupt run mid-flight

Python ใช้ ClaudeSDKClient + pass message generator:

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage, TextBlock
import asyncio

async def interactive_session():
    async def message_generator():
        # First message: kick off the work
        yield {
            "type": "user",
            "message": {"role": "user", "content": "Audit buggy-shop for crash bugs."},
        }
        # Later messages can depend on timing, user input, or earlier results
        await asyncio.sleep(2)
        yield {
            "type": "user",
            "message": {"role": "user", "content": "Now fix the first one you found."},
        }
    options = ClaudeAgentOptions(max_turns=10, allowed_tools=["Read", "Edit", "Grep"])
    async with ClaudeSDKClient(options) as client:
        await client.query(message_generator())
        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(interactive_session())

Generator yield user message ตอนไหนก็ได้ที่ตัดสินใจ. Session warm ระหว่าง message. Warmth นั้นทำให้ message ที่ 2 “now fix the first one you found” มีความหมาย — agent จำสิ่งที่เจอได้

Production: streaming input mode = default ที่ถูกสำหรับ interactive surface (chat UI, REPL, agent ที่คน supervise). Reserve single-message สำหรับ genuinely stateless one-shot job. เลือก input mode by accident = ทีมสุดท้าย add image upload หรือ interrupt ไม่ได้ต้อง rewrite. Decision ครั้งเดียว — ตัดสินใจ on purpose

ทำวันนี้

• Add include_partial_messages ให้ agent ง่ายสุด + print text_delta chunk ด้วย flush=True. 3 nested check เปลี่ยน frozen cursor เป็น live text
• Add in_tool boolean + ประกาศ tool call ด้วย [Using ...] — silent Bash run จะไม่ดูเหมือน hang
• Audit thinking-token setting — max_thinking_tokens/maxThinkingTokens set + streaming ไม่ work = สาเหตุนี้ ไม่ใช่ bug
• Decide input mode deliberately — ถ้ามีคนดู, interrupt, attach image = เลือก streaming input mode ตอนนี้ ก่อน architecture harden รอบ one-shot query()

Takeaway

Streaming = ต่างระหว่าง agent ที่รู้สึกเป็น black box vs colleague ที่ทำงาน out loud

Output streaming = ถูก: 3 nested type check + 1 boolean → silent pause กลายเป็น live narration

Input streaming = ไม่ใช่ feature ที่ toggle = architectural choice. Pick persistent session ตอนคน in loop + one-shot ตอนไม่มีใคร

Agent รู้ตลอดว่าทำอะไร — อ่านไฟล์ รัน test form plan ตอน cursor นิ่ง. Streaming ไม่ทำให้ agent ฉลาดขึ้น — เปิดประตูให้คนรอข้างนอกเห็นงาน

แหล่งที่มา: “Claude Agent SDK Streaming: Your AI Agent Already Knows What It Is Doing.” (Part 3 of 14) โดย Rick Hightower ตีพิมพ์ 20 มิ.ย. 2026 บน Towards AI (Medium) · อ่านต้นฉบับ