Flujo de eventos de la sesión

La comunicación con Claude Managed Agents se basa en eventos. Abres un flujo para seguir una sesión, envías eventos para dirigirla y le respondes cuando se pausa en espera de una entrada.

Tipos de eventos

Los eventos fluyen en dos direcciones: tú envías eventos al agente y la sesión te devuelve eventos.

• Eventos que envías: los eventos user.* inician una sesión y la dirigen a medida que avanza. Los eventos system.message añaden contexto a nivel de sistema.
• Eventos que recibes: los eventos de sesión, los eventos de span y los eventos del agente informan sobre el estado de la sesión y el progreso del agente.

Estas cadenas de tipo de evento siguen una convención de nomenclatura {domain}.{action}. Consulta Tipos de eventos en la referencia para ver el catálogo completo.

Los tipos de eventos de webhook son independientes, y algunos de sus nombres difieren de los del flujo. Por ejemplo, los webhooks usan session.status_idled en lugar de session.status_idle.

Transmitir eventos

Transmite eventos desde la sesión mediante "streaming" (streaming) para recibir actualizaciones en tiempo real mientras el agente trabaja. Un flujo entrega solo los eventos emitidos después de abrirse, así que abre el flujo antes de enviar eventos para evitar una condición de carrera.

# Abre primero el stream y luego envía el mensaje del usuario
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

La sesión informa de los errores mediante el evento session.error. Consulta Tipos de eventos para ver sus campos.

El texto de respuesta del agente llega como eventos agent.message almacenados en búfer, cada uno emitido solo después de que finaliza la solicitud al modelo que lo produjo. Para mostrar el texto mientras el modelo aún lo está generando, consulta Previsualizar respuestas con deltas de eventos.

Reconectar sin perder eventos

Para reconectarte a una sesión existente sin perder eventos, combina un nuevo flujo con el historial de eventos:

1. Abre un nuevo flujo.
2. Lista el historial completo de eventos para inicializar un conjunto de IDs de eventos ya vistos.
3. Sigue el flujo en vivo, omitiendo cualquier evento que ya haya devuelto la lista del historial.

with client.beta.sessions.events.stream(session.id) as stream:
    # El stream está abierto y almacenando en búfer. Lista el historial antes de seguir en vivo.
    history = client.beta.sessions.events.list(session.id)
    seen_event_ids = {past_event.id for past_event in history}

    # Sigue los eventos en vivo, omitiendo lo que ya se haya visto
    for event in stream:
        if event.type == "event_start" or event.type == "event_delta":
            # Las vistas previas de deltas no están habilitadas en esta conexión.
            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

Envía un evento user.message para iniciar o continuar el trabajo del 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",
                },
            ],
        },
    ],
)

Cada evento del historial de la sesión incluye una marca de tiempo processed_at, que se establece cuando el evento termina de procesarse. En los eventos que envías, processed_at es None mientras el evento sigue en cola detrás de eventos anteriores. Las excepciones son user.define_outcome, user.custom_tool_result y user.tool_result, que se procesan al recibirse y se devuelven con processed_at ya completado.

Responder cuando la sesión queda inactiva

Un evento session.status_idle significa que el agente se ha detenido y está esperando una entrada. Su stop_reason.type indica el motivo:

Ningún evento que envíes reanuda una sesión pausada por su presupuesto. El trabajo que el presupuesto pausó se reanuda automáticamente cuando cambias el presupuesto a un valor superior al costo de lista consumido, o lo eliminas. Cuando el trabajo se reanuda, la sesión emite un evento workflow_run.status_running por cada ejecución de flujo de trabajo que el presupuesto pausó. Una ejecución que pausó una interrupción permanece pausada; consulta Reanudar una sesión en su presupuesto. Consulta Cuando una sesión alcanza su presupuesto para ver los eventos que marcan la pausa y los eventos que la sesión sigue aceptando.

Responder a llamadas a herramientas que pausan la sesión

Una sesión se pausa cuando el agente invoca una herramienta personalizada, y cuando una llamada a herramienta necesita tu confirmación según una política de permisos. Ambas pausas siguen la misma secuencia:

1. La sesión emite la llamada a herramienta como un evento agent.custom_tool_use, agent.tool_use o agent.mcp_tool_use.
2. La sesión se pausa con un evento session.status_idle cuyo stop_reason.type es requires_action. Los IDs de los eventos bloqueantes están en el arreglo stop_reason.event_ids.
3. Para cada ID de evento bloqueante, envía un evento user.custom_tool_result o user.tool_confirmation. Un resultado de herramienta personalizada no tiene que esperar al paso 2.
4. Una vez resueltos todos los eventos bloqueantes, la sesión vuelve a running.

En una sesión multiagente, los eventos bloqueantes de un subagente se publican también en el hilo principal. Consulta Permisos de herramientas y herramientas personalizadas.

Devolver un resultado de herramienta personalizada

El evento agent.custom_tool_use contiene el nombre y la entrada de la herramienta. Ejecuta la herramienta en tu sistema. Luego envía un evento user.custom_tool_result, pasando el ID del evento en el parámetro custom_tool_use_id junto con el contenido del resultado.

Puedes enviar el resultado en cuanto llegue el evento agent.custom_tool_use, sin esperar a session.status_idle. La sesión sigue emitiendo session.status_idle con un motivo de detención requires_action para la llamada, y tu cliente puede ignorarlo. Un segundo resultado para la misma llamada se acepta y no tiene efecto.

Cuando te reconectas a una sesión pausada, stop_reason.event_ids en el evento session.status_idle más reciente enumera las llamadas que debes responder.

El siguiente ejemplo responde a cada llamada a medida que llega:

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

                # Envía el resultado de vuelta
                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

Si el agente tiene ejecuciones de flujo de trabajo abiertas, una llamada a herramienta personalizada puede llegar mientras la sesión permanece en running. El evento de inactividad requires_action llega solo cuando ninguno de los hilos de la sesión está trabajando. No lo esperes: responde a cada llamada cuando llegue su evento agent.custom_tool_use. El ejemplo de Seguir una ejecución muestra cómo hacerlo.

Confirmar una llamada a herramienta

Envía un evento user.tool_confirmation, pasando el ID del evento en el parámetro tool_use_id. Establece result en "allow" o "deny". Consulta Responder a solicitudes de confirmación para saber qué llamadas esperan una confirmación y cómo explicar una denegación.

El siguiente ejemplo aprueba todas las llamadas pendientes:

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:
                        # Aprueba la llamada a herramienta pendiente
                        client.beta.sessions.events.send(
                            session.id,
                            events=[
                                {
                                    "type": "user.tool_confirmation",
                                    "tool_use_id": event_id,
                                    "result": "allow",
                                },
                            ],
                        )
                case "end_turn":
                    break

El ejemplo anterior aprueba cada llamada después de que la sesión queda inactiva. Si el agente tiene ejecuciones de flujo de trabajo abiertas, una llamada a herramienta puede esperar tu confirmación mientras la sesión permanece en running. No esperes al evento de inactividad: cuando llegue un evento agent.tool_use o agent.mcp_tool_use cuyo evaluated_permission sea ask, respóndelo. El ejemplo de Seguir una ejecución responde a las llamadas a herramientas personalizadas de esta manera, y su introducción explica cómo añadir confirmaciones.

Reanudar una sesión inactiva

Las sesiones persisten entre interacciones. Para reanudar una sesión, envíale un evento user.message como de costumbre:

# Reanuda una sesión creada previamente enviándole un nuevo evento user.message.
# En producción, pasa el ID almacenado de la sesión que quieres reanudar.
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.",
                },
            ],
        },
    ],
)

El historial de la conversación se conserva a menos que la sesión se elimine explícitamente. Cuando una sesión queda inactiva, se crea un punto de control de su sandbox. El punto de control conserva el estado completo del sandbox, incluido el sistema de archivos, los paquetes instalados y cualquier archivo que haya creado el agente.

Interrumpir al agente

Envía un evento user.interrupt para detener al agente a mitad de la ejecución y, a continuación, envía un evento user.message para redirigirlo:

# El agente está analizando un archivo...
# Interrumpe con una nueva dirección:
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.",
                },
            ],
        },
    ],
)

La llamada regresa en cuanto los eventos quedan en cola. Luego, la interrupción surte efecto en este orden:

1. El agente aplica la interrupción. Hasta entonces, el processed_at de la interrupción permanece en None y la sesión permanece en running. Una respuesta del modelo en curso se detiene de inmediato, pero la interrupción puede tardar más en aplicarse mientras se ejecutan llamadas a herramientas.
2. El evento user.interrupt aparece en el flujo, y el turno interrumpido termina con un evento session.status_idle.
3. El agente comienza su siguiente turno con el user.message que enviaste después de la interrupción.

El stop_reason.type del evento de inactividad es end_turn, el mismo valor que el de un turno que termina por sí solo. Si hay ejecuciones de flujo de trabajo abiertas, la interrupción no finaliza ninguna de ellas. Una ejecución en curso puede mantener la sesión en running, por lo que es posible que el evento session.status_idle del paso 2 no llegue. Consulta Interrumpir una sesión con ejecuciones abiertas.

Listar eventos anteriores

Recupera el historial completo de eventos de una sesión:

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

Pasa un filtro types para devolver solo 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 mensajes del sistema

En un modelo compatible, envía un evento system.message para dar al agente contexto privilegiado a nivel de sistema. El contexto se aplica al turno que lo acompaña y a todos los turnos posteriores. Úsalo cuando el agente necesite orientación actualizada a nivel de sistema a mitad de la sesión, por ejemplo, una persona diferente, restricciones revisadas o contexto obtenido en tiempo de ejecución.

El content del evento acepta de 1 a 1000 elementos de texto. El contenido se añade al contexto del sistema de la sesión como un turno role: "system". No reemplaza la "system prompt" (indicación del sistema) de nivel superior, que establece el campo system en la definición del 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 compatibles

system.message es compatible con estos 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

Si el modelo principal del agente no admite la inyección de mensajes del sistema a mitad de la conversación, el evento se rechaza con un error de validación model_does_not_support_mid_conversation_system. Los modelos de los subagentes no se verifican, porque system.message llega solo al hilo principal.

Mensajes del sistema mientras hay una llamada a herramienta pendiente

Mientras la sesión está inactiva con un motivo de detención requires_action, un system.message debe ir después de un evento de resultado de herramienta en la misma solicitud. Si se envía solo o con un user.message, se rechaza hasta que se resuelvan los eventos de herramienta pendientes.

Próximos pasos