Recevoir les réponses en SSE

Utilisez ce guide lorsque vous souhaitez afficher la réponse IA de manière incrémentale au fur et à mesure de sa génération, plutôt que d'attendre la réponse complète.

Résultat attendu : votre client lit le flux SSE et traite chaque delta de texte à sa réception.

Quand utiliser le streaming

Ajoutez ?stream=true à POST /public/v1/api/execute lorsque :

• Vous construisez une interface de chat qui affiche le texte au fur et à mesure de son apparition.
• Votre flow produit des réponses longues et vous souhaitez réduire la latence perçue.
• Vous avez besoin de run_id ou session_id avant la fin de la réponse (ils arrivent dans le premier événement data-run).

Pour les automatisations courtes et synchrones (traitement en lot, webhooks vers d'autres systèmes), omettez le paramètre de requête ou utilisez ?stream=false et lisez la réponse JSON. Le paramètre de requête fait autorité si le champ historique du corps est également présent.

Effectuer une requête en streaming

POST /public/v1/api/execute?stream=true
Authorization: Bearer <clé>
Content-Type: application/json

{"payload":{"user_message":"Expliquez l'impact de ce changement de code."}}

Le serveur répond avec Content-Type: text/event-stream; charset=utf-8 et x-vercel-ai-ui-message-stream: v1, puis commence à émettre des événements.

Séquence des événements

Chaque réponse en streaming suit cet ordre :

data: {"type":"data-run","data":{"run_id":"9c1d…","session_id":"b1f2…"}}
data: {"type":"text-start","id":"pub_a1b2…"}
data: {"type":"text-delta","id":"pub_a1b2…","delta":"Le changement "}
data: {"type":"text-delta","id":"pub_a1b2…","delta":"introduit une rupture d'API."}
data: {"type":"text-end","id":"pub_a1b2…"}
data: {"type":"finish","finishReason":"stop"}
data: [DONE]

Si l'exécution échoue après le début du streaming :

data: {"type":"error","errorText":"Echec d'exécution : timeout"}

Aucun événement finish ne suit un error. Le statut HTTP est déjà 200 à ce stade ; l'événement error est le signal d'échec.

Gérer une attente livechat

Lorsqu'un flow passe à WAITING_HUMAN pour un transfert livechat, le serveur émet immédiatement data-status :

data: {"type":"data-status","data":{"status":"pending","output":{"messages":[],"metadata":{"livechat_mode":"livechat_waiting"}}}}

Continuez à lire la même connexion. Le serveur envoie des commentaires keepalive SSE pendant l'attente, puis les publications ultérieures de l'automatisation sur cette connexion lorsque le flow reprend. Ne traitez pas data-status comme une fin ; seuls finish, puis data: [DONE], terminent un flux réussi.

Si la conversation était déjà en livechat avant la requête, le serveur conserve le transport SSE mais ne crée pas d'exécution : data-run contient run_id: null, suivi de data-status, finish et data: [DONE].

Un nœud Question produit la même séquence (question publiée, puis data-status), mais avec livechat_mode: null. L'exécution ne reprend qu'à réception de la réponse, envoyée par un nouvel appel /execute avec le même session_id : voir Répondre à un nœud Question.

Préserver les limites de publication

Un flow peut publier plusieurs messages assistant avant de se terminer. Chaque publication logique possède ses propres événements text-start, zéro ou plusieurs text-delta, puis text-end, tous associés au même id stable de publication. Le même identifiant apparaît en JSON dans output.messages[].id et dans l'historique d'origine automation sous publication_id.

Regroupez le texte par id. N'attendez pas de delta synthétique "\n\n" entre les publications et ne dédupliquez pas les publications selon leur contenu. La ligne vide entre les enregistrements data: relève uniquement du framing SSE standard, pas d'une limite de message.

Analyser le flux en Node.js

async function streamApiChannel(
  payload: string,
  apiKey: string,
  onDelta: (publicationId: string, delta: string) => void
): Promise<{ runId: string | null; sessionId: string }> {
  const body = JSON.stringify({ payload: { user_message: payload } });

  const response = await fetch('https://api.example.com/public/v1/api/execute?stream=true', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${apiKey}`,
      'Content-Type': 'application/json',
    },
    body,
  });

  if (!response.ok) {
    const err = await response.json();
    throw new Error(`${err.error?.code}: ${err.error?.message}`);
  }

  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = '';
  let runId: string | null = null;
  let sessionId = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    const chunks = buffer.split('\n\n');
    buffer = chunks.pop() ?? '';

    for (const chunk of chunks) {
      const dataLine = chunk.split('\n').find((line) => line.startsWith('data: '));
      if (!dataLine) continue; // Inclut les commentaires keepalive.
      const data = dataLine.slice(6);
      if (data === '[DONE]') return { runId, sessionId };
      const event = JSON.parse(data);

      switch (event.type) {
        case 'data-run':
          runId = event.data.run_id;
          sessionId = event.data.session_id;
          break;
        case 'text-delta':
          onDelta(event.id, event.delta);
          break;
        case 'data-status':
          // Continuer la lecture : l'automatisation reprise arrive sur cette connexion.
          break;
        case 'error':
          throw new Error(`Erreur de flux : ${event.errorText}`);
      }
    }
  }

  return { runId, sessionId };
}

Analyser le flux dans un navigateur

const response = await fetch('/public/v1/api/execute?stream=true', {
  method: 'POST',
  headers: {
    Authorization: 'Bearer ' + publicKey,
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({ payload: { user_message: userMessage } }),
});

const reader = response.body!.getReader();
const decoder = new TextDecoder();
let buffer = '';
const publications = new Map<string, string>();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  buffer += decoder.decode(value, { stream: true });

  const chunks = buffer.split('\n\n');
  buffer = chunks.pop() ?? '';
  for (const chunk of chunks) {
    const dataLine = chunk.split('\n').find((line) => line.startsWith('data: '));
    if (!dataLine || dataLine === 'data: [DONE]') continue;
    const event = JSON.parse(dataLine.slice(6));
    if (event.type === 'text-delta') {
      publications.set(event.id, (publications.get(event.id) ?? '') + event.delta);
      renderPublications(publications);
    }
  }
}

Les appelants navigateur doivent utiliser une clé nx_pub_ et le canal doit avoir l'origine de la page dans sa liste d'origines autorisées. Le navigateur envoie lui-même l'en-tête Origin : ne le définissez pas dans le code.

Gestion de la déconnexion client

Lorsque le client ferme la connexion, le serveur arrête d'observer cette requête mais n'annule pas l'exécution sous-jacente. Récupérez ensuite l'état assistant avec GET /public/v1/api/conversations/{session_id}/messages.

Pour un retry ambigu, envoyez la même valeur Idempotency-Key et la même requête sémantique. Une fois l'exécution terminée, le serveur rejoue en SSE la séquence de publications enregistrée au lieu de créer une nouvelle exécution. Le mode de streaming fait partie de l'identité de la requête : conservez ?stream=true lors du retry.

Mise en mémoire tampon par proxy

Si vous routez les requêtes via nginx ou un autre proxy inverse, assurez-vous que la mise en mémoire tampon des réponses est désactivée pour le point d'accès de streaming :

location /public/v1/api/ {
    proxy_buffering off;
    proxy_pass http://core_api;
}

Le serveur définit X-Accel-Buffering: no automatiquement, mais tous les proxys ne respectent pas cet en-tête.

Référence

Pour les définitions complètes des types d'événements et la sémantique des erreurs, voir la référence d'API.