استریمِ پاسخ‌ها به‌صورتِ بی‌درنگ

Get real-time responses from the Agent SDK as text and tool calls stream in

به‌صورت پیش‌فرض، Agent SDK پس از این‌که کلود تولیدِ هر پاسخ را تمام کرد، شیءهای کاملِ AssistantMessage را برمی‌گرداند. برای دریافتِ به‌روزرسانی‌های تدریجی هم‌زمان با تولیدِ متن و فراخوانیِ ابزارها، استریمِ پیامِ جزئی را با تنظیمِ include_partial_messages (در Python) یا includePartialMessages (در TypeScript) روی true در گزینه‌هایت فعال کن.

فعال‌سازیِ استریمِ خروجی

برای فعال‌سازیِ استریم، include_partial_messages (در Python) یا includePartialMessages (در TypeScript) را در گزینه‌هایت روی true تنظیم کن. این باعث می‌شود SDK علاوه بر AssistantMessage و ResultMessageِ معمول، پیام‌های StreamEvent را که حاویِ رویدادهای خامِ API به‌محضِ رسیدن‌اند برگرداند.

سپس کدِ تو باید:

نوعِ هر پیام را بررسی کند تا StreamEvent را از سایرِ انواعِ پیام تشخیص دهد
برای StreamEvent، فیلدِ event را استخراج و type آن را بررسی کند
به‌دنبالِ رویدادهای content_block_delta بگردد که در آن‌ها delta.type برابرِ text_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())

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);
      }
    }
  }
}

مرجعِ 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

type SDKPartialAssistantMessage = {
  type: "stream_event";
  event: BetaRawMessageStreamEvent; // From Anthropic SDK
  parent_tool_use_id: string | null;
  uuid: UUID;
  session_id: string;
  ttft_ms?: number; // Time to first token in ms, present only on message_start events
};

فیلدِ 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

بدونِ فعال‌بودنِ پیام‌های جزئی (include_partial_messages در Python، includePartialMessages در TypeScript)، همه‌ی انواعِ پیام را به‌جز StreamEvent دریافت می‌کنی. انواعِ رایج عبارت‌اند از SystemMessage (مقداردهیِ اولیه‌ی نشست)، AssistantMessage (پاسخ‌های کامل)، ResultMessage (نتیجه‌ی نهایی)، و یک پیامِ مرزِ فشرده‌سازی که نشان می‌دهد تاریخچه‌ی گفتگو کِی فشرده شده است (SDKCompactBoundaryMessage در TypeScript؛ SystemMessage با subtypeِ "compact_boundary" در Python).

استریمِ پاسخ‌های متنی

برای نمایشِ متن هم‌زمان با تولیدش، به‌دنبالِ رویدادهای content_block_delta بگرد که در آن‌ها delta.type برابرِ text_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())

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(); // Final newline

استریمِ فراخوانیِ ابزارها

فراخوانیِ ابزارها هم به‌صورتِ تدریجی استریم می‌شود. می‌توانی پیگیری کنی کِی ابزارها شروع می‌شوند، ورودیِ آن‌ها را هم‌زمان با تولید دریافت کنی، و ببینی کِی کامل می‌شوند. مثالِ زیر ابزارِ در حالِ فراخوانیِ فعلی را پیگیری می‌کند و ورودیِ 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())

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

// Track the current tool and accumulate its input JSON
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") {
      // New tool call is starting
      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") {
        // Accumulate JSON input as it streams in
        const chunk = event.delta.partial_json;
        toolInput += chunk;
        console.log(`  Input chunk: ${chunk}`);
      }
    } else if (event.type === "content_block_stop") {
      // Tool call complete - show final input
      if (currentTool) {
        console.log(`Tool ${currentTool} called with: ${toolInput}`);
        currentTool = null;
      }
    }
  }
}

ساختنِ یک رابطِ کاربریِ استریمی

این مثال استریمِ متن و ابزار را در یک رابطِ کاربریِ منسجم ترکیب می‌کند. پیگیری می‌کند که آیا ایجنت در حالِ حاضر در حالِ اجرای یک ابزار است (با استفاده از پرچمِ 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())

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

// Track whether we're currently in a tool call
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 is starting - show status indicator
        process.stdout.write(`\n[Using ${event.content_block.name}...]`);
        inTool = true;
      }
    } else if (event.type === "content_block_delta") {
      // Only stream text when not executing a 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 finished
        console.log(" done");
        inTool = false;
      }
    }
  } else if (message.type === "result") {
    // Agent finished all work
    console.log("\n\n--- Complete ---");
  }
}

محدودیت‌های شناخته‌شده

خروجیِ ساخت‌یافته (Structured output): نتیجه‌ی JSON فقط در ResultMessage.structured_output نهایی ظاهر می‌شود، نه به‌صورتِ دلتاهای استریمی. برای جزئیات به خروجی‌های ساخت‌یافته مراجعه کن.

گام‌های بعدی

حالا که می‌توانی متن و فراخوانیِ ابزارها را به‌صورتِ بی‌درنگ استریم کنی، این موضوعاتِ مرتبط را بررسی کن:

query‌های تعاملی در برابرِ تک‌مرحله‌ای: بینِ حالت‌های ورودی برای موردِ استفاده‌ی خودت انتخاب کن
خروجی‌های ساخت‌یافته: پاسخ‌های JSONِ نوع‌دار از ایجنت بگیر
دسترسی‌ها: کنترل کن ایجنت از کدام ابزارها بتواند استفاده کند