Поток событий сессии

Взаимодействие с Claude Managed Agents основано на событиях. Вы отправляете агенту пользовательские события и получаете обратно события агента и сессии для отслеживания статуса.



Типы событий

События передаются в двух направлениях.

• Пользовательские события и системные события — это то, что вы отправляете агенту: события user.* запускают сессию и управляют ею по мере выполнения; system.message обновляет системную подсказку агента между ходами.
• События сессии, события span и события агента отправляются вам для наблюдения за состоянием сессии и прогрессом агента. Потоковые соединения, которые явно включили эту опцию, также получают дельты событий.

Строки типов событий сессии, span, агента, пользователя и системы следуют соглашению об именовании {domain}.{action}. Исключение составляют события предварительного просмотра дельт, доступные только в потоке (event_start, event_delta). Полный каталог см. в разделе Типы событий справочника.

Каждое сохранённое событие включает временную метку processed_at, указывающую, когда событие было записано на стороне сервера. Если processed_at равно null, это означает, что событие было поставлено в очередь оболочкой и обрабатывается после завершения обработки предшествующих событий.



Интеграция событий



Дельты событий

По умолчанию текст ответа агента поступает в поток в виде буферизованных событий agent.message, каждое из которых генерируется только после завершения запроса к модели, который его породил. Дельты событий позволяют отображать этот текст инкрементально, в виде живого предварительного просмотра, пока модель ещё генерирует его. Предварительный просмотр — это не сам ответ: предварительные просмотры являются вспомогательным средством отображения, предоставляемым по мере возможности, а буферизованное событие agent.message всегда остаётся авторитетной записью. Клиент, игнорирующий предварительные просмотры, всё равно получает полный и корректный поток.



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

Предварительные просмотры включаются отдельно для каждого потокового соединения. Добавьте параметр запроса event_deltas[] к GET /v1/sessions/{session_id}/events/stream, повторяя его по одному разу для каждого типа события, для которого вы хотите получать предварительный просмотр. Допустимые значения — agent.message и agent.thinking; любое другое значение возвращает ошибку 400. Этот параметр поддерживается только потоком событий уровня сессии. Потоки событий потоков сессии его отклоняют.

Когда начинается событие с предварительным просмотром, поток генерирует event_start, содержащий тип и id предстоящего события:

{
  "type": "event_start",
  "event": {
    "type": "agent.message",
    "id": "sevt_01abc..."
  }
}

Для agent.message за стартовым событием следуют события event_delta, содержащие инкрементальный текст. Каждая дельта указывает событие, которое она дополняет, в event_id, и блок контента, который она дополняет, в delta.index:

{
  "type": "event_delta",
  "event_id": "sevt_01abc...",
  "delta": {
    "type": "content_delta",
    "index": 0,
    "content": {
      "type": "text",
      "text": "Here is the summary"
    }
  }
}

Когда предварительно просматривается событие agent.thinking, генерируется только event_start. События event_delta за ним не следуют, а контент поступает в буферизованном событии agent.thinking как обычно.

В отличие от сохранённых событий, event_start и event_delta не имеют собственных id или processed_at. Единственный идентификатор, который они несут, — это id события, которое они предварительно просматривают.



Накопление и согласование

SDK для Python, TypeScript и Go включают вспомогательный аккумулятор, который индексирует предварительный просмотр по id события и берёт на себя учёт index. Ручной подход работает на любом языке: в других SDK применяйте его к сгенерированным типам событий.

При ручном подходе рассматривайте предварительный просмотр как черновой буфер, а буферизованное событие — как окончательную запись. Индексируйте буфер по ключу (event_id, index). Выполняйте согласование для каждого запроса к модели: ход открывается одним событием session.status_running, затем в ходе, который завершается нормально, каждый запрос к модели порождает по порядку span.model_request_start, event_start, события event_delta, буферизованное agent.message и, наконец, span.model_request_end (на вкладке «События span»). Обрабатывайте каждое событие по мере его поступления:

1. При получении event_start зафиксируйте объявленный id. Идентификаторы всегда совпадают: event_start.event.id, каждый event_delta.event_id и id буферизованного agent.message — это одно и то же значение.
2. При получении каждого event_delta добавляйте delta.content.text к записи по ключу (event_id, delta.index) и отображайте накопленный текст. Первая дельта для данного index создаёт эту запись.
3. Когда поступает буферизованное agent.message, сопоставьте его по id, отбросьте накопленный предварительный просмотр и отобразите вместо него содержимое сообщения.
4. При получении span.model_request_end закройте любой предварительный просмотр, который не был согласован со своим буферизованным событием. Больше дельт для него не поступит. Если ход завершается с ошибкой или прерывается, буферизованное событие может так и не поступить; span.model_request_end всё равно поступает.

# Снимки предпросмотра, индексированные по id события. accumulate_managed_agents_event сворачивает каждое
# event_start / event_delta в снимок agent.message; буферизованное
# agent.message заменяет его.
previews: dict[str, BetaManagedAgentsAgentMessageEvent] = {}

# Включаем предпросмотры agent.message для этого соединения
with client.beta.sessions.events.stream(
    session.id, event_deltas=["agent.message"]
) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Describe the repo in one sentence."}],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "event_start":
                snapshot = accumulate_managed_agents_event(None, event)
                if snapshot is not None:
                    previews[event.event.id] = snapshot
                print(f"event_start             {event.event.type} {event.event.id}")
            case "event_delta":
                preview = accumulate_managed_agents_event(previews.get(event.event_id), event)
                if preview is not None:
                    previews[event.event_id] = preview
                    text = "".join(block.text for block in preview.content)
                    print(f"event_delta             preview: {text!r}")
            case "agent.message":
                # Буферизованное событие — это итоговая запись: оно заменяет и закрывает предпросмотр
                preview = accumulate_managed_agents_event(previews.pop(event.id, None), event)
                text = "".join(block.text for block in preview.content)
                print(f"agent.message           {event.id} {text!r}")
            case "span.model_request_end":
                # Дельт больше не будет. Закрываем все предпросмотры, чьё
                # буферизованное событие так и не пришло.
                for event_id in previews:
                    print(f"span.model_request_end  closing preview for {event_id}")
                previews.clear()
            case "session.status_idle":
                break



Ограничения

Предварительные просмотры оптимизированы для отзывчивости. Учитывайте следующие ограничения при разработке:

• По мере возможности: При высокой нагрузке сервер может отбрасывать дельты для события. В этом случае вы получаете непрерывный префикс текста, после чего дельты для этого события больше не поступают. Буферизованное agent.message всё равно поступает полностью. Никогда не рассматривайте накопленный предварительный просмотр как окончательный.
• Нет повторного воспроизведения при переподключении: Дельты доставляются только тому соединению, которое их включило, и только пока оно открыто. Если поток обрывается, следуйте процедуре переподключения на вкладке «Потоковая передача событий»: заново откройте поток и получите историю событий. История включает все буферизованные события, сгенерированные, пока вы были отключены, в том числе agent.message, которого ожидал ваш предварительный просмотр. Повторно запросить пропущенные дельты невозможно.
• Только основной поток, только текст: Предварительные просмотры охватывают текст ассистента в основном потоке сессии. Использование инструментов, результаты инструментов, результаты MCP и активность в других потоках сессии никогда не предварительно просматриваются.
• Только старт для agent.thinking: Предварительный просмотр agent.thinking генерирует только event_start как сигнал о том, что блок мышления начался; события event_delta за ним не следуют.
• Никогда не сохраняются: event_start и event_delta существуют только в живом потоке. Они не появляются в истории событий сессии (GET /v1/sessions/{session_id}/events).



Дополнительные сценарии



Обработка вызовов пользовательских инструментов

Когда агент вызывает пользовательский инструмент:

1. Сессия генерирует событие agent.custom_tool_use, содержащее имя инструмента и входные данные.
2. Сессия приостанавливается с событием session.status_idle, содержащим stop_reason: requires_action. Идентификаторы блокирующих событий находятся в массиве stop_reason.event_ids.
3. Выполните инструмент в вашей системе и отправьте событие user.custom_tool_result для каждого из них, передав идентификатор события в параметре custom_tool_use_id вместе с содержимым результата.
4. После разрешения всех блокирующих событий сессия возвращается в состояние running.

with client.beta.sessions.events.stream(session.id) as stream:
    for event in stream:
        if event.type == "session.status_idle" and (stop_reason := event.stop_reason):
            match stop_reason.type:
                case "requires_action":
                    for event_id in stop_reason.event_ids:
                        # Находим событие использования пользовательского инструмента и выполняем его
                        tool_event = events_by_id[event_id]
                        result = call_tool(tool_event.name, tool_event.input)

                        # Отправляем результат обратно
                        client.beta.sessions.events.send(
                            session.id,
                            events=[
                                {
                                    "type": "user.custom_tool_result",
                                    "custom_tool_use_id": event_id,
                                    "content": [{"type": "text", "text": result}],
                                },
                            ],
                        )
                case "end_turn":
                    break



Подтверждение инструментов

Когда политика разрешений требует подтверждения перед выполнением инструмента:

1. Сессия генерирует событие agent.tool_use или agent.mcp_tool_use.
2. Сессия приостанавливается с событием session.status_idle, содержащим stop_reason: requires_action. Идентификаторы блокирующих событий находятся в массиве stop_reason.event_ids.
3. Отправьте событие user.tool_confirmation для каждого из них, передав идентификатор события в параметре tool_use_id. Установите result в значение "allow" или "deny". Используйте deny_message, чтобы объяснить причину отказа.
4. После разрешения всех блокирующих событий сессия возвращается в состояние running.

with client.beta.sessions.events.stream(session.id) as stream:
    for event in stream:
        if event.type == "session.status_idle" and (stop_reason := event.stop_reason):
            match stop_reason.type:
                case "requires_action":
                    for event_id in stop_reason.event_ids:
                        # Одобряем ожидающий вызов инструмента
                        client.beta.sessions.events.send(
                            session.id,
                            events=[
                                {
                                    "type": "user.tool_confirmation",
                                    "tool_use_id": event_id,
                                    "result": "allow",
                                },
                            ],
                        )
                case "end_turn":
                    break



Возобновление простаивающей сессии

Сессии сохраняются между взаимодействиями. История разговора сохраняется, если сессия не удалена явно. Когда сессия переходит в состояние простоя, для её песочницы создаётся контрольная точка, сохраняющая полное состояние песочницы, включая файловую систему, установленные пакеты и любые файлы, созданные агентом. Это позволяет корректно возобновить работу после периода неактивности.

Чтобы возобновить сессию, отправьте ей событие user.message как обычно:

# В продакшене передайте сохранённый ID сессии, которую хотите возобновить.
ant beta:sessions:events send --session-id "$SESSION_ID" <<'YAML'
events:
  - type: user.message
    content:
      - type: text
        text: Now run the tests against the changes you made earlier.
YAML



Отправка системных сообщений

Отправьте событие system.message, чтобы обновить системную подсказку агента между ходами. В отличие от поля system в определении агента (которое фиксируется при создании сессии), system.message позволяет изменять системную подсказку по мере выполнения сессии. Используйте его, когда агенту требуются обновлённые указания системного уровня в середине сессии: другая персона, пересмотренные ограничения или контекст, полученный во время выполнения, который должен влиять на поведение модели в дальнейшем.

ant beta:sessions:events send --session-id "$SESSION_ID" <<'YAML'
events:
  - type: system.message
    content:
      - type: text
        text: "The user's current timezone is America/New_York."
YAML

system.message нельзя отправить, пока сессия простаивает с stop_reason: requires_action. content принимает от 1 до 1000 текстовых элементов.



Отслеживание использования

Объект сессии включает поле usage с накопительной статистикой токенов. Получите сессию после её перехода в состояние простоя, чтобы прочитать актуальные итоговые значения, и используйте их для отслеживания затрат, контроля бюджетов или мониторинга потребления.

{
  "id": "sesn_01...",
  "status": "idle",
  "usage": {
    "input_tokens": 5000,
    "output_tokens": 3200,
    "cache_creation_input_tokens": 2000,
    "cache_read_input_tokens": 20000
  }
}

input_tokens отражает некэшированные входные токены, а output_tokens — общее количество выходных токенов по всем вызовам модели в сессии. Поля cache_creation_input_tokens и cache_read_input_tokens отражают активность кэширования подсказок. Записи кэша имеют TTL 5 минут, поэтому последовательные ходы в пределах этого окна выигрывают от чтения из кэша, что снижает стоимость за токен.



Наблюдаемость в Console

Console предоставляет визуальное представление временной шкалы ваших сессий агента. Перейдите в раздел Claude Managed Agents в Console, чтобы увидеть:

• Список сессий: Все сессии с их статусом, временем создания и моделью
• Представление трассировки: Хронологическое представление событий (содержимое, временные метки, использование токенов) внутри сессии. Представления трассировки доступны только разработчикам и администраторам.
• Выполнение инструментов: Подробности каждого вызова инструмента и его результата



Советы по отладке

• Проверяйте события сессии: Ошибки сессии передаются через событие session.error
• Просматривайте результаты инструментов: Сбои выполнения инструментов часто объясняют неожиданное поведение агента
• Отслеживайте использование токенов: Контролируйте потребление токенов, чтобы оптимизировать подсказки и снизить затраты
• Используйте системные подсказки: Добавьте инструкции по логированию в системную подсказку, чтобы агент объяснял свои рассуждения