Flux d'événements de session

La communication avec Claude Managed Agents est basée sur les événements. Vous ouvrez un flux pour suivre une session, envoyez des événements pour la diriger, et lui répondez lorsqu'elle se met en pause en attente d'une entrée.

Types d'événements

Les événements circulent dans deux directions : vous envoyez des événements à l'agent, et la session vous renvoie des événements.

• Événements que vous envoyez : les événements user.* démarrent une session et l'orientent au fil de sa progression. Les événements system.message ajoutent du contexte au niveau système.
• Événements que vous recevez : les événements de session, les événements de span et les événements d'agent rendent compte de l'état de la session et de la progression de l'agent.

Ces chaînes de type d'événement suivent une convention de nommage {domain}.{action}. Consultez Types d'événements dans la référence pour le catalogue complet.

Les types d'événements webhook sont distincts, et certains de leurs noms diffèrent de ceux du flux. Par exemple, les webhooks utilisent session.status_idled plutôt que session.status_idle.

Diffuser les événements en streaming

Diffusez en « streaming » (diffusion en flux) les événements de la session pour recevoir des mises à jour en temps réel pendant que l'agent travaille. Un flux ne délivre que les événements émis après son ouverture ; ouvrez donc le flux avant d'envoyer des événements afin d'éviter une « race condition » (situation de concurrence).

# Ouvrir d'abord le stream, puis envoyer le message utilisateur
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 session signale les erreurs via l'événement session.error. Consultez Types d'événements pour ses champs.

Le texte de réponse de l'agent arrive sous forme d'événements agent.message mis en mémoire tampon, chacun n'étant émis qu'une fois terminée la requête au modèle qui l'a produit. Pour afficher le texte pendant que le modèle le génère encore, consultez Prévisualiser les réponses avec les deltas d'événements.

Se reconnecter sans manquer d'événements

Pour vous reconnecter à une session existante sans manquer d'événements, combinez un nouveau flux avec l'historique des événements :

1. Ouvrez un nouveau flux.
2. Listez l'historique complet des événements pour initialiser un ensemble d'identifiants d'événements déjà vus.
3. Suivez le flux en direct, en ignorant tous les événements déjà renvoyés par la liste de l'historique.

with client.beta.sessions.events.stream(session.id) as stream:
    # Le stream est ouvert et met en mémoire tampon. Lister l'historique avant de suivre le direct.
    history = client.beta.sessions.events.list(session.id)
    seen_event_ids = {past_event.id for past_event in history}

    # Suivre les événements en direct, en ignorant ceux déjà vus
    for event in stream:
        if event.type == "event_start" or event.type == "event_delta":
            # Les aperçus de deltas ne sont pas activés sur cette connexion.
            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

Envoyer des événements

Envoyez un événement user.message pour démarrer ou poursuivre le travail de l'agent :

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",
                },
            ],
        },
    ],
)

Chaque événement de l'historique de la session comporte un horodatage processed_at, défini lorsque le traitement de l'événement se termine. Sur les événements que vous envoyez, processed_at vaut None tant que l'événement est encore en file d'attente derrière des événements antérieurs. Les exceptions sont user.define_outcome, user.custom_tool_result et user.tool_result, qui sont traités dès leur réception et renvoyés avec processed_at déjà renseigné.

Répondre lorsque la session devient inactive

Un événement session.status_idle signifie que l'agent s'est arrêté et attend une entrée. Son stop_reason.type indique pourquoi :

Aucun événement que vous envoyez ne reprend une session mise en pause à son budget. Le travail que le budget a mis en pause reprend automatiquement lorsque vous modifiez le budget pour une valeur supérieure au coût catalogue consommé, ou que vous le supprimez. Lorsque le travail reprend, la session émet un événement workflow_run.status_running pour chaque exécution de workflow que le budget a mise en pause. Une exécution mise en pause par une interruption reste en pause ; consultez Reprendre une session à son budget. Consultez Lorsqu'une session atteint son budget pour les événements qui marquent la pause et les événements que la session accepte encore.

Répondre aux appels d'outils qui mettent la session en pause

Une session se met en pause lorsque l'agent invoque un outil personnalisé, et lorsqu'un appel d'outil nécessite votre confirmation en vertu d'une politique d'autorisation. Les deux pauses suivent la même séquence :

1. La session émet l'appel d'outil sous forme d'événement agent.custom_tool_use, agent.tool_use ou agent.mcp_tool_use.
2. La session se met en pause avec un événement session.status_idle dont le stop_reason.type est requires_action. Les identifiants des événements bloquants se trouvent dans le tableau stop_reason.event_ids.
3. Pour chaque identifiant d'événement bloquant, envoyez un événement user.custom_tool_result ou user.tool_confirmation. Un résultat d'outil personnalisé n'a pas besoin d'attendre l'étape 2.
4. Une fois tous les événements bloquants résolus, la session repasse à l'état running.

Dans une session multiagent, les événements bloquants d'un sous-agent sont également publiés sur le fil principal. Consultez Autorisations d'outils et outils personnalisés.

Renvoyer un résultat d'outil personnalisé

L'événement agent.custom_tool_use contient le nom et l'entrée de l'outil. Exécutez l'outil dans votre système. Envoyez ensuite un événement user.custom_tool_result, en passant l'identifiant de l'événement dans le paramètre custom_tool_use_id avec le contenu du résultat.

Vous pouvez envoyer le résultat dès que l'événement agent.custom_tool_use arrive, sans attendre session.status_idle. La session émet tout de même session.status_idle avec une raison d'arrêt requires_action pour l'appel, et votre client peut l'ignorer. Un second résultat pour le même appel est accepté et n'a aucun effet.

Lorsque vous vous reconnectez à une session en pause, stop_reason.event_ids sur le dernier événement session.status_idle liste les appels auxquels répondre.

L'exemple suivant répond à chaque appel dès son arrivée :

with client.beta.sessions.events.stream(session.id) as stream:
    for event in stream:
        match event.type:
            case "agent.custom_tool_use":
                # Exécuter l'outil
                result = call_tool(event.name, event.input)

                # Renvoyer le résultat
                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 l'agent a des exécutions de workflow ouvertes, un appel d'outil personnalisé peut arriver alors que la session reste running. L'événement d'inactivité requires_action n'arrive que lorsqu'aucun des fils de la session n'est en cours de travail. Ne l'attendez pas : répondez à chaque appel lorsque son événement agent.custom_tool_use arrive. L'exemple de Suivre une exécution montre comment procéder.

Confirmer un appel d'outil

Envoyez un événement user.tool_confirmation, en passant l'identifiant de l'événement dans le paramètre tool_use_id. Définissez result sur "allow" ou "deny". Consultez Répondre aux demandes de confirmation pour savoir quels appels attendent une confirmation et comment expliquer un refus.

L'exemple suivant approuve chaque appel en attente :

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:
                        # Approuver l'appel d'outil en attente
                        client.beta.sessions.events.send(
                            session.id,
                            events=[
                                {
                                    "type": "user.tool_confirmation",
                                    "tool_use_id": event_id,
                                    "result": "allow",
                                },
                            ],
                        )
                case "end_turn":
                    break

L'exemple précédent approuve chaque appel après que la session est devenue inactive. Si l'agent a des exécutions de workflow ouvertes, un appel d'outil peut attendre votre confirmation alors que la session reste running. N'attendez pas l'événement d'inactivité : lorsqu'un événement agent.tool_use ou agent.mcp_tool_use arrive dont le champ evaluated_permission vaut ask, répondez-y. L'exemple de Suivre une exécution répond aux appels d'outils personnalisés de cette manière, et son introduction explique comment ajouter des confirmations.

Reprendre une session inactive

Les sessions persistent entre les interactions. Pour reprendre une session, envoyez-lui un événement user.message comme d'habitude :

# Reprendre une session précédemment créée en lui envoyant un nouvel événement user.message.
# En production, passez l'ID stocké de la session que vous souhaitez reprendre.
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.",
                },
            ],
        },
    ],
)

L'historique de conversation est conservé sauf si la session est explicitement supprimée. Lorsqu'une session devient inactive, son sandbox fait l'objet d'un point de contrôle. Le point de contrôle préserve l'état complet du sandbox, y compris le système de fichiers, les paquets installés et tous les fichiers créés par l'agent.

Interrompre l'agent

Envoyez un événement user.interrupt pour arrêter l'agent en cours d'exécution, puis envoyez un événement user.message pour le rediriger :

# L'agent est en train d'analyser un fichier...
# Interrompre avec une nouvelle direction :
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.",
                },
            ],
        },
    ],
)

L'appel retourne dès que les événements sont mis en file d'attente. L'interruption prend ensuite effet dans cet ordre :

1. L'agent applique l'interruption. Jusque-là, le processed_at de l'interruption reste None et la session reste à l'état running. Une réponse du modèle en cours s'arrête immédiatement, mais l'interruption peut mettre plus de temps à s'appliquer pendant que des appels d'outils sont en cours d'exécution.
2. L'événement user.interrupt apparaît sur le flux, et le tour interrompu se termine par un événement session.status_idle.
3. L'agent commence son tour suivant avec le user.message que vous avez envoyé après l'interruption.

Le stop_reason.type de l'événement d'inactivité est end_turn, la même valeur que pour un tour qui se termine de lui-même. Si des exécutions de workflow sont ouvertes, l'interruption n'en termine aucune. Une exécution en cours peut maintenir la session à l'état running, de sorte que l'événement session.status_idle de l'étape 2 pourrait ne pas arriver. Consultez Interrompre une session avec des exécutions ouvertes.

Lister les événements passés

Récupérez l'historique complet des événements d'une session :

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

Transmettez un filtre types pour ne renvoyer que des types d'événements spécifiques :

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}")

Envoyer des messages système

Sur un modèle pris en charge, envoyez un événement system.message pour fournir à l'agent un contexte privilégié au niveau système. Le contexte s'applique au tour qui l'accompagne et à tous les tours suivants. Utilisez-le lorsque l'agent a besoin de directives système mises à jour en cours de session, par exemple un persona différent, des contraintes révisées ou un contexte récupéré à l'exécution.

Le champ content de l'événement accepte de 1 à 1000 éléments de texte. Le contenu est ajouté au contexte système de la session sous forme de tour role: "system". Il ne remplace pas l'« system prompt » (invite système) de premier niveau, définie par le champ system de la définition de l'agent.

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

Modèles pris en charge

system.message est pris en charge par ces modèles :

• 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 le modèle principal de l'agent ne prend pas en charge l'injection système en cours de conversation, l'événement est rejeté avec une erreur de validation model_does_not_support_mid_conversation_system. Les modèles des sous-agents ne sont pas vérifiés, car system.message n'arrive que sur le fil principal.

Messages système lorsqu'un appel d'outil est en attente

Lorsque la session est inactive avec une raison d'arrêt requires_action, un system.message doit suivre un événement de résultat d'outil dans la même requête. Envoyé seul ou avec un user.message, il est rejeté jusqu'à ce que les événements d'outils en attente soient résolus.

Étapes suivantes