Потоки сессии

В мультиагентной сессии каждый агент работает в собственном session thread (потоке сессии). На этой странице описано, как получать список потоков, прерывать и архивировать их, какие события они отправляют и как работают разрешения инструментов между ними. «Workflow run» (запуск рабочего процесса) тоже создаёт потоки сессии.

Основной поток и потоки сессии

Поток событий уровня сессии (/v1/sessions/{session_id}/events/stream) считается primary thread (основным потоком) и содержит сжатое представление всей активности во всех потоках. Вы не видите полную активность субагентов, но видите начало и конец их работы, а также блокирующие события, такие как запросы разрешений на использование инструментов.

Потоки сессии позволяют детально изучить активность конкретного агента.

status сессии отражает совокупную активность всех агентов: если хотя бы один поток находится в состоянии running, то и общий статус сессии — running. Выполняющийся запуск рабочего процесса тоже может удерживать сессию в состоянии running, даже если ни один из его потоков не работает. Когда ни один поток не работает и какой-либо поток ожидает вашего клиента, сессия находится в состоянии idle; см. Как узнать, что работа выполнена.

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

Получение списка потоков

Получите список всех потоков сессии следующим образом:

for thread in client.beta.sessions.threads.list(session.id):
    agent = thread.agent
    label = agent.type if agent.type == "advisor" else agent.name
    print(f"[{label}] {thread.status}")

Полный список включает основной поток. Для основного потока parent_thread_id равен null. Все остальные потоки являются дочерними. workflow_run_id равен null, кроме как у потоков запуска.

Чтобы получить только потоки с определёнными статусами, добавьте в запрос statuses[] и повторите его, чтобы указать более одного статуса, например ?statuses[]=running&statuses[]=idle. Если параметр не указан, возвращаются потоки с любым статусом.

Прерывание потока сессии

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

client.beta.sessions.events.send(
    session.id,
    events=[{"type": "user.interrupt", "session_thread_id": thread.id}],
)

Для потока субагента, заблокированного в состоянии requires_action, прерывание закрывает каждый ожидающий вызов инструмента результатом инструмента с ошибкой («Tool execution was interrupted before completion. Please retry.») и напрямую повторно отправляет session.thread_status_idle с stop_reason: end_turn; модель при этом не вызывается. Для дочернего потока, простаивающего с end_turn или budget_reached, прерывание ничего не делает. Прерывание, указывающее завершённый поток, возвращает ошибку 400. Прерванный дочерний поток не отправляет агенту основного потока отчёт, который он отправляет по завершении хода. Пока этот агент ожидает дочерний поток, он не начинает новый ход, пока до него не дойдёт что-то другое, например user.message или отчёт другого потока.

Архивирование потока сессии

При желании заархивируйте поток сессии, когда он завершит свою работу. Архивирование потока освобождает его место в рамках ограничения в 25 дочерних потоков. Сервер сам архивирует потоки запуска рабочего процесса. Вам не нужно их архивировать, и вы не можете сделать это, пока запуск открыт.

archived = client.beta.sessions.threads.archive(thread.id, session_id=session.id)
print(archived.status, archived.archived_at)

Архивирование выполняется успешно, только если поток находится в состоянии idle. Поток, остановленный в состоянии requires_action, считается простаивающим и может быть заархивирован напрямую; прервать сначала нужно только выполняющийся поток:

client.beta.sessions.events.send(
    session.id,
    events=[{"type": "user.interrupt", "session_thread_id": thread.id}],
)
archived = client.beta.sessions.threads.archive(thread.id, session_id=session.id)
print(archived.status, archived.archived_at)

События основного потока

Эти события отражают мультиагентную активность в основном потоке по адресу /v1/sessions/{session_id}/events/stream. События, связанные с направлением сообщений, названы относительно потока, в котором они появляются: agent.thread_message_received означает, что в этот поток пришло сообщение из другого потока, а agent.thread_message_sent означает, что этот поток отправил сообщение. Например, задача, которую делегирует агент основного потока, поступает в собственный поток событий дочернего потока как событие agent.thread_message_received.

Консультации советника отправляют те же события потоков под зарезервированным именем anthropic.advisor (как agent_name в событиях жизненного цикла потока и как from_agent_name при доставке совета); последовательность описана в разделе Предоставление сессии советника.

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

• События жизненного цикла: каждый поток запуска отправляет session.thread_created с workflow_run_id запуска, а также свои события session.thread_status_running, session.thread_status_idle и session.thread_status_terminated.
• События сообщений: подсказка потока запуска, событие agent.thread_message_received, остаётся в его собственном потоке событий.
• События запуска: события workflow_run.* также поступают в этот поток событий; см. События запуска.
• Вызовы инструментов, ожидающие вас: вызовы инструментов потока запуска, которым нужен ваш клиент, дублируются в этот поток событий, как и для любого дочернего потока. См. Разрешения инструментов и пользовательские инструменты.

События потока сессии

Критические события передаются в основной поток. Однако вам всё равно может понадобиться изучить рассуждения и вызовы инструментов конкретного агента. Для этого получайте события связанного потока сессии через потоковую передачу или в виде списка.

У каждого потока сессии есть собственный поток событий по адресу /v1/sessions/{session_id}/threads/{thread_id}/stream, и он принимает тот же параметр event_deltas[], что и поток событий уровня сессии, поэтому вы можете предварительно просматривать текст субагента по мере его генерации моделью. Соединение показывает предварительный просмотр только того потока, который оно читает: предварительные просмотры дочернего потока никогда не появляются в потоке событий уровня сессии, поэтому, чтобы наблюдать за субагентом в реальном времени, откройте его собственный поток событий. Сведения о включении, накоплении и согласовании предварительных просмотров см. в разделе Предварительный просмотр событий потоков выполнения сессии.

В запуске рабочего процесса сервер выполняет рабочий процесс — программу, которую пишет агент основного потока. В каждом из потоков запуска первое событие agent.thread_message_received — это подсказка, которую написал рабочий процесс. Его from_session_thread_id — это ID основного потока, а from_agent_name в этом событии нет. API не гарантирует текст подсказки, поэтому не разбирайте его. Событие session.thread_status_terminated потока в потоке событий основного потока сообщает вам, что поток завершён. Ни одно событие не фиксирует результат, который он вернул рабочему процессу.

Поток событий потока не воспроизводит более ранние события. Сразу после session.thread_created список событий потока запуска может быть пустым, поскольку сервер записывает первое событие потока после него. Поэтому сначала откройте поток событий потока, затем получите список событий потока и пропускайте каждое событие из потоковой передачи, id которого вернул список.

with client.beta.sessions.threads.events.stream(
    thread.id,
    session_id=session.id,
) as stream:
    for event in stream:
        match event.type:
            case "agent.message":
                for block in event.content:
                    if block.type == "text":
                        print(block.text, end="")
            case "session.thread_status_idle":
                break

for event in client.beta.sessions.threads.events.list(
    thread.id,
    session_id=session.id,
):
    print(f"[{event.type}] {event.processed_at}")

Разрешения инструментов и пользовательские инструменты

Если субагенту что-то нужно от вашего клиента, например разрешение на выполнение вызова инструмента или результат пользовательского инструмента, событие дублируется в основной поток с session_thread_id, идентифицирующим исходный поток сессии. Вызову инструмента требуется ваше разрешение при always_ask или при auto, когда сервер не приходит к решению.

{
  "type": "session.thread_status_idle",
  "id": "sevt_01ABC...",
  "session_thread_id": "sthr_01DEF...",
  "agent_name": "code-reviewer",
  "stop_reason": {
    "type": "requires_action",
    "event_ids": ["sevt_01XYZ..."]
  }
}

Отправьте user.tool_confirmation (с tool_use_id) или user.custom_tool_result (с custom_tool_use_id); сервер автоматически направит ответ в нужный поток. Ответ может появиться в основном потоке и в потоке субагента с разными значениями id. Чтобы сопоставить две копии, сравнивайте type и tool_use_id (или custom_tool_use_id), а не id.

Сессия переходит в состояние idle, только когда ни один поток не находится в состоянии running, поэтому session.status_idle может прийти намного позже вызова субагента. Вам не нужно его ждать: отправьте user.custom_tool_result, как только придёт продублированное событие agent.custom_tool_use.

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

Следующий пример размещается внутри цикла событий обработчика подтверждения инструментов. Для каждого ID в stop_reason.event_ids он отправляет user.tool_confirmation, разрешающий вызов. Тот же шаблон применим к user.custom_tool_result.

for event_id in stop.event_ids:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.tool_confirmation",
                "tool_use_id": event_id,
                "result": "allow",
            }
        ],
    )

Приведённый выше шаблон отвечает на вызовы, перечисленные в событии простоя. В основном потоке событий событие session.thread_status_idle субагента может прийти раньше событий agent.tool_use или agent.mcp_tool_use, перечисленных в его stop_reason.event_ids. user.tool_confirmation для вызова, событие которого ещё не пришло, может вернуть ошибку 400. Чтобы избежать этого, отвечайте на каждый вызов, у которого evaluated_permission равно ask, когда его собственное событие приходит в основной поток событий.