Потоковая передача ответов в реальном времени

По умолчанию Agent SDK выдает полные объекты AssistantMessage после того, как Claude завершит генерирование каждого ответа. Чтобы получать дополнительные обновления по мере генерирования текста и вызовов инструментов, включите потоковую передачу частичных сообщений, установив include_partial_messages (Python) или includePartialMessages (TypeScript) в значение true в ваших параметрах.

На этой странице рассматривается потоковая передача выходных данных (получение токенов в реальном времени). Для режимов ввода (как вы отправляете сообщения), см. Отправка сообщений агентам. Вы также можете потоковую передачу ответов с использованием Agent SDK через CLI.

Включение потоковой передачи выходных данных

Чтобы включить потоковую передачу, установите include_partial_messages (Python) или includePartialMessages (TypeScript) в значение true в ваших параметрах. Это заставляет SDK выдавать сообщения StreamEvent, содержащие необработанные события API по мере их поступления, в дополнение к обычным AssistantMessage и ResultMessage.

Ваш код затем должен:

Проверить тип каждого сообщения, чтобы отличить StreamEvent от других типов сообщений
Для StreamEvent извлечь поле event и проверить его type
Искать события content_block_delta, где delta.type — это text_delta, которые содержат фактические текстовые фрагменты

Пример ниже включает потоковую передачу и выводит текстовые фрагменты по мере их поступления. Обратите внимание на вложенные проверки типов: сначала для StreamEvent, затем для content_block_delta, затем для text_delta:

Справочник 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

Без включенных частичных сообщений (include_partial_messages в Python, includePartialMessages в TypeScript) вы получаете все типы сообщений, кроме StreamEvent. Распространенные типы включают SystemMessage (инициализация сеанса), AssistantMessage (полные ответы), ResultMessage (финальный результат) и CompactBoundaryMessage (указывает, когда история разговора была сжата).

Потоковая передача текстовых ответов

Чтобы отобразить текст по мере его генерирования, ищите события content_block_delta, где delta.type — это text_delta. Они содержат дополнительные текстовые фрагменты. Пример ниже выводит каждый фрагмент по мере его поступления:

Потоковая передача вызовов инструментов

Вызовы инструментов также потоковой передачей постепенно. Вы можете отслеживать, когда инструменты начинают работу, получать их входные данные по мере их генерирования и видеть, когда они завершаются. Пример ниже отслеживает текущий вызываемый инструмент и накапливает входные данные JSON по мере их потоковой передачи. Он использует три типа событий:

content_block_start: инструмент начинает работу
content_block_delta с input_json_delta: поступают фрагменты входных данных
content_block_stop: вызов инструмента завершен

Создание потокового пользовательского интерфейса

Этот пример объединяет потоковую передачу текста и инструментов в единый пользовательский интерфейс. Он отслеживает, выполняет ли агент в настоящее время инструмент (используя флаг in_tool), чтобы показать индикаторы статуса, такие как [Using Read...], пока работают инструменты. Текст потоком передается нормально, когда не используется инструмент, и завершение инструмента запускает сообщение "done". Этот паттерн полезен для интерфейсов чата, которым нужно показывать прогресс во время многошаговых задач агента.

Известные ограничения

Некоторые функции SDK несовместимы с потоковой передачей:

Расширенное мышление: когда вы явно устанавливаете max_thinking_tokens (Python) или maxThinkingTokens (TypeScript), сообщения StreamEvent не выдаются. Вы будете получать только полные сообщения после каждого хода. Обратите внимание, что мышление отключено по умолчанию в SDK, поэтому потоковая передача работает, если вы его не включите.
Структурированный вывод: результат JSON появляется только в финальном ResultMessage.structured_output, а не как потоковые дельты. Подробнее см. в разделе структурированные выходные данные.

Следующие шаги

Теперь, когда вы можете потоком передавать текст и вызовы инструментов в реальном времени, изучите эти связанные темы:

Интерактивные и одноразовые запросы: выберите режим ввода для вашего случая использования
Структурированные выходные данные: получайте типизированные ответы JSON от агента
Разрешения: контролируйте, какие инструменты может использовать агент

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

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

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

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