Stream de eventos da sessão

A comunicação com o Claude Managed Agents é baseada em eventos. Você abre um stream para acompanhar uma sessão, envia eventos para direcioná-la e responde a ela quando ela pausa aguardando entrada.

Tipos de eventos

Os eventos fluem em duas direções: você envia eventos para o agente, e a sessão envia eventos de volta para você.

• Eventos que você envia: eventos user.* iniciam uma sessão e a conduzem à medida que ela progride. Eventos system.message acrescentam contexto em nível de sistema.
• Eventos que você recebe: eventos de sessão, eventos de span e eventos do agente informam o estado da sessão e o progresso do agente.

Essas strings de tipo de evento seguem uma convenção de nomenclatura {domain}.{action}. Consulte Tipos de eventos na referência para o catálogo completo.

Os tipos de eventos de webhook são separados, e alguns de seus nomes diferem dos do stream. Por exemplo, os webhooks usam session.status_idled em vez de session.status_idle.

Fazer streaming de eventos

Faça "streaming" (streaming) dos eventos da sessão para receber atualizações em tempo real enquanto o agente trabalha. Um stream entrega apenas os eventos emitidos depois que ele é aberto, então abra o stream antes de enviar eventos para evitar uma "race condition" (condição de corrida).

# Abra o stream primeiro e depois envie a mensagem do usuário
with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Summarize the repo README"}],
            },
        ],
    )

    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.status_idle":
                break
            case "session.error":
                error_message = event.error.message if event.error else "unknown"
                print(f"\n[Error: {error_message}]")
                break

A sessão informa erros por meio do evento session.error. Consulte Tipos de eventos para ver seus campos.

O texto de resposta do agente chega como eventos agent.message armazenados em buffer, cada um emitido somente depois que a requisição ao modelo que o produziu termina. Para renderizar o texto enquanto o modelo ainda o está gerando, consulte Pré-visualizar respostas com deltas de eventos.

Reconectar sem perder eventos

Para se reconectar a uma sessão existente sem perder eventos, combine um novo stream com o histórico de eventos:

1. Abra um novo stream.
2. Liste o histórico completo de eventos para preencher um conjunto de IDs de eventos já vistos.
3. Acompanhe o stream ao vivo, ignorando quaisquer eventos já retornados pela lista do histórico.

with client.beta.sessions.events.stream(session.id) as stream:
    # O stream está aberto e armazenando em buffer. Liste o histórico antes de acompanhar ao vivo.
    history = client.beta.sessions.events.list(session.id)
    seen_event_ids = {past_event.id for past_event in history}

    # Acompanhe eventos ao vivo, ignorando o que já foi visto
    for event in stream:
        if event.type == "event_start" or event.type == "event_delta":
            # As prévias de delta não estão habilitadas nesta conexão.
            continue
        if event.id in seen_event_ids:
            continue
        seen_event_ids.add(event.id)
        match event.type:
            case "agent.message":
                for block in event.content:
                    if block.type == "text":
                        print(block.text, end="")
            case "session.status_idle":
                break

Enviar eventos

Envie um evento user.message para iniciar ou continuar o trabalho do agente:

client.beta.sessions.events.send(
    session.id,
    events=[
        {
            "type": "user.message",
            "content": [
                {
                    "type": "text",
                    "text": "Analyze the performance of the sort function in utils.py",
                },
            ],
        },
    ],
)

Todo evento no histórico da sessão inclui um timestamp processed_at, definido quando o evento termina de ser processado. Nos eventos que você envia, processed_at é None enquanto o evento ainda está na fila atrás de eventos anteriores. As exceções são user.define_outcome, user.custom_tool_result e user.tool_result, que são processados no recebimento e ecoados de volta com processed_at já preenchido.

Responder quando a sessão fica ociosa

Um evento session.status_idle significa que o agente parou e está aguardando entrada. Seu stop_reason.type informa o motivo:

Nenhum evento que você envia retoma uma sessão pausada em seu orçamento. O trabalho que o orçamento pausou é retomado automaticamente quando você altera o orçamento para um valor acima do custo de tabela consumido, ou o remove. Quando o trabalho é retomado, a sessão emite um evento workflow_run.status_running para cada execução de workflow que o orçamento pausou. Uma execução que uma interrupção pausou permanece pausada; consulte Retomar uma sessão em seu orçamento. Consulte Quando uma sessão atinge seu orçamento para ver os eventos que marcam a pausa e os eventos que a sessão ainda aceita.

Responder a chamadas de ferramenta que pausam a sessão

Uma sessão pausa quando o agente invoca uma ferramenta personalizada, e quando uma chamada de ferramenta precisa da sua confirmação sob uma política de permissões. Ambas as pausas seguem a mesma sequência:

1. A sessão emite a chamada de ferramenta como um evento agent.custom_tool_use, agent.tool_use ou agent.mcp_tool_use.
2. A sessão pausa com um evento session.status_idle cujo stop_reason.type é requires_action. Os IDs dos eventos bloqueantes estão no array stop_reason.event_ids.
3. Para cada ID de evento bloqueante, envie um evento user.custom_tool_result ou user.tool_confirmation. Um resultado de ferramenta personalizada não precisa esperar pela etapa 2.
4. Depois que todos os eventos bloqueantes forem resolvidos, a sessão volta para running.

Em uma sessão multiagente, os eventos bloqueantes de um subagente são replicados na thread principal. Consulte Permissões de ferramentas e ferramentas personalizadas.

Retornar um resultado de ferramenta personalizada

O evento agent.custom_tool_use contém o nome e a entrada da ferramenta. Execute a ferramenta no seu sistema. Em seguida, envie um evento user.custom_tool_result, passando o ID do evento no parâmetro custom_tool_use_id junto com o conteúdo do resultado.

Você pode enviar o resultado assim que o evento agent.custom_tool_use chegar, sem esperar por session.status_idle. A sessão ainda emite session.status_idle com um motivo de parada requires_action para a chamada, e seu cliente pode ignorá-lo. Um segundo resultado para a mesma chamada é aceito e não tem efeito.

Quando você se reconecta a uma sessão pausada, stop_reason.event_ids no evento session.status_idle mais recente lista as chamadas a serem respondidas.

O exemplo a seguir responde a cada chamada à medida que ela chega:

with client.beta.sessions.events.stream(session.id) as stream:
    for event in stream:
        match event.type:
            case "agent.custom_tool_use":
                # Execute a ferramenta
                result = call_tool(event.name, event.input)

                # Envie o resultado de volta
                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 "session.status_idle":
                if event.stop_reason and event.stop_reason.type == "end_turn":
                    break

Se o agente tiver execuções de workflow abertas, uma chamada de ferramenta personalizada pode chegar enquanto a sessão permanece running. O evento de ociosidade requires_action chega somente quando nenhuma das threads da sessão está trabalhando. Não espere por ele: responda a cada chamada quando seu evento agent.custom_tool_use chegar. O exemplo em Acompanhar uma execução mostra como.

Confirmar uma chamada de ferramenta

Envie um evento user.tool_confirmation, passando o ID do evento no parâmetro tool_use_id. Defina result como "allow" ou "deny". Consulte Responder a solicitações de confirmação para saber quais chamadas aguardam uma confirmação e como explicar uma negação.

O exemplo a seguir aprova todas as chamadas pendentes:

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:
                        # Aprove a chamada de ferramenta pendente
                        client.beta.sessions.events.send(
                            session.id,
                            events=[
                                {
                                    "type": "user.tool_confirmation",
                                    "tool_use_id": event_id,
                                    "result": "allow",
                                },
                            ],
                        )
                case "end_turn":
                    break

O exemplo anterior aprova cada chamada depois que a sessão fica ociosa. Se o agente tiver execuções de workflow abertas, uma chamada de ferramenta pode aguardar sua confirmação enquanto a sessão permanece running. Não espere pelo evento de ociosidade: quando chegar um evento agent.tool_use ou agent.mcp_tool_use cujo evaluated_permission seja ask, responda a ele. O exemplo em Acompanhar uma execução responde a chamadas de ferramentas personalizadas dessa forma, e sua introdução explica como adicionar confirmações.

Retomar uma sessão ociosa

As sessões persistem entre interações. Para retomar uma sessão, envie um evento user.message para ela normalmente:

# Retome uma sessão criada anteriormente enviando a ela um novo evento user.message.
# Em produção, passe o ID armazenado da sessão que você deseja retomar.
client.beta.sessions.events.send(
    session.id,
    events=[
        {
            "type": "user.message",
            "content": [
                {
                    "type": "text",
                    "text": "Now run the tests against the changes you made earlier.",
                },
            ],
        },
    ],
)

O histórico da conversa é preservado, a menos que a sessão seja explicitamente excluída. Quando uma sessão fica ociosa, é criado um checkpoint do seu sandbox. O checkpoint preserva o estado completo do sandbox, incluindo o sistema de arquivos, os pacotes instalados e quaisquer arquivos que o agente criou.

Interromper o agente

Envie um evento user.interrupt para parar o agente durante a execução e, em seguida, envie um evento user.message para redirecioná-lo:

# O agente está analisando um arquivo no momento...
# Interrompa com uma nova direção:
client.beta.sessions.events.send(
    session.id,
    events=[
        {"type": "user.interrupt"},
        {
            "type": "user.message",
            "content": [
                {
                    "type": "text",
                    "text": "Instead, focus on fixing the bug in line 42.",
                },
            ],
        },
    ],
)

A chamada retorna assim que os eventos são enfileirados. A interrupção então entra em vigor nesta ordem:

1. O agente aplica a interrupção. Até lá, o processed_at da interrupção permanece None e a sessão permanece running. Uma resposta do modelo em andamento para imediatamente, mas a interrupção pode levar mais tempo para ser aplicada enquanto chamadas de ferramenta estão em execução.
2. O evento user.interrupt aparece no stream, e o turno interrompido termina com um evento session.status_idle.
3. O agente inicia seu próximo turno com o user.message que você enviou após a interrupção.

O stop_reason.type do evento de ociosidade é end_turn, o mesmo valor de um turno que termina por conta própria. Se houver execuções de workflow abertas, a interrupção não encerra nenhuma delas. Uma execução em andamento pode manter a sessão running, então o evento session.status_idle da etapa 2 pode não chegar. Consulte Interromper uma sessão com execuções abertas.

Listar eventos anteriores

Recupere o histórico completo de eventos de uma sessão:

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

Passe um filtro types para retornar apenas tipos de eventos específicos:

events = client.beta.sessions.events.list(
    session.id,
    types=["agent.tool_use", "agent.tool_result"],
)
for event in events.data:
    print(f"[{event.type}] {event.processed_at}")

Enviar mensagens do sistema

Em um modelo compatível, envie um evento system.message para fornecer ao agente contexto privilegiado em nível de sistema. O contexto se aplica ao turno correspondente e a todos os turnos subsequentes. Use-o quando o agente precisar de orientação atualizada em nível de sistema no meio da sessão, por exemplo, uma persona diferente, restrições revisadas ou contexto obtido em tempo de execução.

O content do evento aceita de 1 a 1000 itens de texto. O conteúdo é acrescentado ao contexto de sistema da sessão como um turno role: "system". Ele não substitui o "system prompt" (prompt do sistema) de nível superior, que é definido pelo campo system na definição do agente.

client.beta.sessions.events.send(
    session.id,
    events=[
        {
            "type": "system.message",
            "content": [
                {
                    "type": "text",
                    "text": "The user's current timezone is America/New_York.",
                },
            ],
        },
    ],
)

Modelos compatíveis

system.message é compatível com estes modelos:

• Claude Fable 5.1
• Claude Mythos 5.1
• Claude Fable 5
• Claude Mythos 5
• Claude Opus 5.5
• Claude Opus 5
• Claude Opus 4.8
• Claude Sonnet 5.5
• Claude Haiku 5.5

Se o modelo principal do agente não for compatível com injeção de sistema no meio da conversa, o evento é rejeitado com um erro de validação model_does_not_support_mid_conversation_system. Os modelos de subagentes não são verificados, porque system.message chega apenas à thread principal.

Mensagens do sistema enquanto uma chamada de ferramenta está pendente

Enquanto a sessão está ociosa com um motivo de parada requires_action, um system.message deve vir depois de um evento de resultado de ferramenta na mesma requisição. Enviado sozinho ou com um user.message, ele é rejeitado até que os eventos de ferramenta pendentes sejam resolvidos.

Próximos passos