Einen Streaming-Chat mit WebSocket-Connectors in connic bauen

Ein WebSocket-Connector in connic verbindet eine Chat-App über eine dauerhafte Verbindung mit einem bereitgestellten Agenten. Der Browser baut diese Verbindung selbst auf, jede Nachricht startet einen Agenten-Run, und die Antwort kommt als Folge von Ereignissen zurück, die die ID der Nachricht tragen. Das Tutorial baut Client und Agentenkonfiguration auf: Authentifizierung pro Nutzer, gestreamte Antworten und Fortschrittsereignisse, ein Gesprächsverlauf, der einen Verbindungsabbruch übersteht, und die Fälle, in denen Output Guardrails, Freigaben oder Fehler verändern, wie eine Antwort ankommt.

Den Connector anlegen

Der Connector gehört zu dem Agenten, der im Chat antwortet, hier support-chat. Im Dashboard öffnet man den Agenten, klickt im Connector Flow auf Add inbound connector, dann auf Create New Connector und wählt WebSocket. connic erzeugt die URL und zeigt sie im Detail-Drawer des Connectors. Der Connector lässt sich auch im Code deklarieren, damit seine Einstellungen zusammen mit dem Agenten bereitgestellt werden:

type: websocket
config:
  require_auth: false            # stattdessen trägt jede Nachricht ein Nutzer-Token
  progress_events: true          # Ereignisse zu Modell, Tools und Freigaben
  session_timeout: 1800          # Verbindung nach 30 Minuten ohne Nachricht schließen
  max_messages_per_session: 500
agents:
  - support-chat

Das nächste Deployment legt den Connector an. Streaming ist standardmäßig aktiv. require_auth: false schaltet Require Authentication aus, sodass sich der Browser ohne das Secret des Connectors verbinden kann, und progress_events ergänzt die Statusereignisse, die weiter unten folgen. session_timeout schließt eine Verbindung nach so vielen Sekunden ohne Nachricht (60 bis 86.400, Standard 3.600), und max_messages_per_session begrenzt die Nachrichten pro Verbindung (1 bis 10.000, Standard 100). Beide Grenzen schließen die Verbindung, der Verlauf des Gesprächs bleibt erhalten. Die Dokumentation zu Connectors im Code beschreibt Abweichungen pro Umgebung.

Jede Nachricht authentifizieren

Ohne Require Authentication kann jede Person, die die URL kennt, eine Verbindung öffnen. Der Agent prüft deshalb bei jeder Nachricht, wer sie geschickt hat. Der Browser fügt jeder Nachricht das Token der angemeldeten Person hinzu, und die Middleware des Agenten prüft es, bevor der Run weiterläuft. Das Token muss in jeder Nachricht stehen, weil jede Nachricht ein eigener Run ist. Das Secret des Connectors bleibt so aus der Seite heraus, wo jede Person es lesen könnte, und jeder Run kennt die Person, für die er arbeitet:

from typing import Any

from connic import StopProcessing


def user_from_token(token: str | None) -> str | None:
    """Prüft das Token beim Anmeldedienst und gibt die Nutzer-ID zurück."""
    raise NotImplementedError


async def before(content: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    payload = context["payload"]
    user_id = user_from_token(payload.get("token"))
    if user_id is None:
        raise StopProcessing("Bitte melde dich erneut an.")
    context["user_id"] = user_id
    context["conversation_key"] = f"{user_id}:{payload['conversation_id']}"
    files = [part for part in content["parts"] if "text" not in part]
    content["parts"] = [{"text": payload["message"]}, *files]
    return content

user_from_token steht für die Prüfung, die der Anmeldedienst der App anbietet. Schlägt sie fehl, schließt StopProcessing den Run mit seiner Meldung als Antwort ab; wer eine abgelaufene Sitzung hat, liest im Chat „Bitte melde dich erneut an.“ Die Middleware läuft auch, wenn ein Run nach einer Freigabe weiterläuft oder nach einer Wartezeit in der Warteschlange startet, und zwar mit dem Token der ursprünglichen Nachricht. Tokens sollten deshalb länger gültig bleiben als das Timeout der Freigabe. Außerdem ersetzt die Middleware die Eingabe des Modells durch den Nachrichtentext, sodass Token und Gesprächs-ID das Modell des Agenten nie erreichen. Die aufgezeichnete Eingabe, der Run-Kontext und der Session-State enthalten den Payload trotzdem. Neue Umgebungen maskieren Felder namens token in diesen Aufzeichnungen bereits. Ältere Umgebungen können *.token in ihrer Einstellung zum Maskieren sensibler Daten ergänzen. Ein Backend, das den Connector ohne Browser aufruft, kann Require Authentication eingeschaltet lassen und stattdessen das Secret des Connectors senden. Ist Require Authentication noch eingeschaltet, wenn sich ein Browser verbindet, wartet connic auf dieses Secret und schließt die Verbindung nach 30 Sekunden mit dem Fehler AUTH_TIMEOUT.

Nachrichten senden und Antwortereignisse lesen

Sobald die Verbindung steht, sendet connic ein connected-Ereignis, das die verknüpften Agenten und die Grenzen der Verbindung nennt. Jede Chat-Nachricht ist ein message-Ereignis mit einem Payload-Objekt. Die id ist optional, aber connic übernimmt sie in die Bestätigung, die Stream-Ereignisse und die meisten Fehler zu dieser Nachricht, und der Client ordnet Antworten darüber zu. Fehler zur Verbindung und einige andere wie MESSAGE_LIMIT tragen keine ID. Der Payload enthält den Text, das Token und eine conversation_id für den Verlauf, den der nächste Abschnitt einrichtet:

{
  "type": "message",
  "id": "m-1",
  "payload": {
    "conversation_id": "conv-blender",
    "message": "Mein Mixer kam mit einem gesprungenen Behälter an.",
    "token": "<Session-Token>"
  }
}

connic bestätigt die Nachricht und streamt die Antwort:

{ "type": "ack", "id": "m-1", "message_number": 1 }
{ "type": "stream_start", "id": "m-1", "agent": "support-chat" }
{ "type": "thinking", "id": "m-1", "agent": "support-chat" }
{ "type": "stream_chunk", "id": "m-1", "agent": "support-chat", "chunk": "Das tut mir leid. " }
{ "type": "stream_chunk", "id": "m-1", "agent": "support-chat", "chunk": "Möchtest du einen Ersatzbehälter oder eine Erstattung?" }
{
  "type": "stream_end",
  "id": "m-1",
  "agent": "support-chat",
  "full_response": "Das tut mir leid. Möchtest du einen Ersatzbehälter oder eine Erstattung?",
  "token_usage": { "input_tokens": 214, "output_tokens": 21, "total_tokens": 235 }
}

stream_start eröffnet die Antwort, das ist der Moment für eine Tipp-Anzeige, und thinking folgt, sobald das Modell beginnt. stream_chunk-Ereignisse liefern den Text, während das Modell ihn schreibt. stream_end schließt die Antwort mit full_response, der endgültigen Ausgabe des Runs, und token_usage ab. Zu speichern ist full_response. Middleware, die die Antwort umschreibt, kann sie von den aneinandergehängten Chunks abweichen lassen. Die Chunks sind daher eine Vorschau, die stream_end ersetzt. Ist Streaming am Connector ausgeschaltet, kommt statt der Stream-Ereignisse ein einzelnes response-Ereignis, und connic sendet keine Fortschrittsereignisse.

Mit Streaming melden Fortschrittsereignisse, was der Agent vor und zwischen den Textfragmenten tut. thinking markiert den Beginn jedes Modellaufrufs, tool_call und tool_result nennen ein Tool, wenn der Agent es aufruft und wenn es zurückkehrt, und approval_pending und approval_resolved rahmen eine Pause für eine Freigabe ein. Eine Antwort, die eine Bestellung nachschlägt:

{ "type": "thinking", "id": "m-5", "agent": "support-chat" }
{ "type": "tool_call", "id": "m-5", "agent": "support-chat", "tool": "lookup_order" }
{ "type": "tool_result", "id": "m-5", "agent": "support-chat", "tool": "lookup_order" }
{ "type": "thinking", "id": "m-5", "agent": "support-chat" }
{ "type": "stream_chunk", "id": "m-5", "agent": "support-chat", "chunk": "Deine Bestellung wurde am Montag verschickt." }

Die Ereignisse tragen nur den Namen des Tools, nie seine Argumente oder sein Ergebnis, und der Client wählt die Beschriftung, etwa „Bestellung wird gesucht“ für lookup_order. Weil der Browser sie direkt empfängt, sind die Tool-Namen in seinen Entwicklertools sichtbar.

Ein minimaler Browser-Client. getToken liefert das aktuelle Token der angemeldeten Person aus der Anmeldebibliothek der App, sodass ein erneuertes Token mit der nächsten Nachricht mitgeht:

const MAX_RETRY_DELAY_MS = 30_000;

export function createChatConnection({ url, getToken, onPreview, onReply, onFailure, onStatus }) {
  let socket;
  let ready = false;
  let stopped = false;
  let attempt = 0;
  const inFlight = new Map(); // Nachrichten-ID -> Vorschautext

  function connect() {
    if (stopped) return;
    socket = new WebSocket(url);
    socket.onmessage = (event) => handle(JSON.parse(event.data));
    socket.onclose = () => {
      ready = false;
      for (const id of inFlight.keys()) onFailure(id, "interrupted");
      inFlight.clear();
      if (stopped) return;
      setTimeout(connect, Math.min(1000 * 2 ** attempt++, MAX_RETRY_DELAY_MS));
    };
  }

  function handle(event) {
    switch (event.type) {
      case "connected":
        ready = true;
        attempt = 0;
        break;
      case "stream_chunk": {
        const preview = (inFlight.get(event.id) ?? "") + event.chunk;
        inFlight.set(event.id, preview);
        onPreview(event.id, preview);
        break;
      }
      case "stream_end":
        inFlight.delete(event.id);
        if (event.error) onFailure(event.id, event.error);
        else onReply(event.id, event.full_response);
        break;
      case "thinking":
      case "tool_call":
      case "tool_result":
      case "approval_pending":
      case "approval_resolved":
        onStatus(event.id, event); // die UI macht aus event.type und event.tool eine Beschriftung
        break;
      case "error":
        if (!event.id) break; // Verbindungsfehler haben keine ID; onclose meldet offene Nachrichten
        inFlight.delete(event.id);
        onFailure(event.id, event.code);
        break;
      // ack, stream_start und ping brauchen hier keine Behandlung
    }
  }

  connect();

  return {
    close() {
      stopped = true;
      socket.close();
    },
    async send(conversationId, text) {
      const token = await getToken();
      if (!ready) return null; // die UI behält den Text und zeigt "Verbinde neu"
      const id = crypto.randomUUID();
      inFlight.set(id, "");
      socket.send(JSON.stringify({
        type: "message",
        id,
        payload: { conversation_id: conversationId, message: text, token },
      }));
      return id;
    },
  };
}

Jeder Browser-Tab öffnet seine eigene Verbindung. Innerhalb einer Verbindung werden Nachrichten nacheinander verarbeitet, in der Reihenfolge des Eingangs, und eine Nachricht, die während einer laufenden Antwort eintrifft, wartet auf deren Ende. Die UI ruft close() auf, wenn der Chat geschlossen wird. Sonst baut der Client die Verbindung nach jedem Schließen neu auf, auch nach dem Timeout für Inaktivität. Runs aus allen Verbindungen teilen sich das max_concurrent_runs des Agenten, das standardmäßig 1 beträgt; mit diesem Standardwert wartet die Antwort einer Person auf die einer anderen. Die Agentendatei im nächsten Abschnitt erhöht den Wert auf 10, was mindestens den Developer-Tarif voraussetzt. Der Tarif des Projekts begrenzt ihn, und ein Deployment über dieser Grenze schlägt fehl. Sind mehrere Agenten mit dem Connector verknüpft, antworten sie nacheinander, jeder mit eigener Folge von stream_start bis stream_end und seinem Namen in agent. Der Client oben geht von einem Agenten aus.

Das Gespräch über neue Verbindungen hinweg fortsetzen

Jede Nachricht startet einen eigenen Agenten-Run. Frühere Nachrichten sieht der Agent nur über eine persistente Session, und session.key legt fest, welche Nachrichten sich eine Session teilen. Hier liest der Key conversation_key, den die Middleware aus der geprüften Nutzer-ID und der conversation_id bildet:

version: "1.0"

name: support-chat
type: llm
model: connic/gpt-5.6-terra
description: "Beantwortet Kundenfragen im Chat der App"
system_prompt: |
  Du bist der Support-Assistent in der Acme-App. Antworte in wenigen Sätzen.

max_concurrent_runs: 10

session:
  key: context.conversation_key
  ttl: 604800   # Verlauf läuft nach 7 Tagen ohne Nachricht ab

concurrency:
  key: conversation_id
  on_conflict: queue

Weil der Key die Nutzer-ID enthält, bekommt eine Person, die die conversation_id einer anderen sendet, eine eigene, leere Session statt des fremden Verlaufs. Eine Verbindung kann mehrere Gespräche tragen, und eine neue Verbindung setzt ein Gespräch fort, sobald sie dieselbe conversation_id mit einem gültigen Token sendet. Ein Run ohne Key schlägt fehl. Im Beispiel hat eine Person zwei Gespräche auf derselben Verbindung offen, als diese abbricht. Jede Antwort zeigt ihre full_response:

Nach dem Neuaufbau weiß der Agent weiterhin, dass es im Gespräch zum Mixer um den gesprungenen Behälter geht, und das Gespräch zum Versand enthält nur seine eigenen Nachrichten. Lange Gespräche können irgendwann das Kontextfenster des Modells überschreiten. Ist am Agenten die Kontextkomprimierung aktiviert, kürzt connic den Verlauf in diesem Fall.

Grenzen der Verbindung und Lebensdauer des Verlaufs

Die Einstellungen am Connector begrenzen die Verbindung. Die ttl des Agenten begrenzt das Gespräch:

Nach den ersten beiden baut der Client die Verbindung neu auf und setzt dasselbe Gespräch fort. Ohne ttl läuft der Verlauf nie ab.

Neue Verbindungen und unterbrochene Antworten

connic sendet frühere Nachrichten nicht an eine neue Verbindung. Die Chat-UI führt ihr eigenes Transkript für die Anzeige, die Session hält den Verlauf, den das Modell liest. Auch eine Antwort, die beim Abbruch der Verbindung noch läuft, erreicht die neue Verbindung nicht. Ihr Run läuft trotzdem zu Ende, und die Antwort wird Teil des Verlaufs. Der Client oben meldet solche Antworten als interrupted, damit die UI darauf hinweisen und die Person erneut fragen kann.

Für diesen Fall ist der concurrency-Block in der Agentendatei gedacht. Runs aus verschiedenen Verbindungen werden nicht nacheinander ausgeführt, und überschneiden sich zwei Runs in einer Session, schlägt der Run fehl, der als zweiter schreibt. Mit der Concurrency-Regel auf conversation_id wartet eine Nachricht direkt nach dem Neuaufbau der Verbindung oder aus einem zweiten Browser-Tab, bis der frühere Run in diesem Gespräch fertig ist.

Guardrails, Freigaben und Fehler behandeln

Jede Antwort folgt derselben Ereignisfolge. Wann ihr Text ankommt und ob sie mit full_response oder einer Fehlermeldung endet, hängt von Guardrails, Freigaben und Fehlern ab.

Output Guardrails liefern die Antwort in einem Stück

Output Guardrails prüfen die vollständige Antwort, bevor der Client etwas davon sieht. Bei einem Agenten mit Output-Regeln streamt connic deshalb keinen Teiltext. Diese Regel blockiert Antworten mit Kartennummern, US-Sozialversicherungsnummern oder API-Keys und schickt stattdessen einen festen Satz:

guardrails:
  output:
    - type: pii_leakage
      mode: block
      config:
        rejection_message: "Das kann ich hier nicht teilen. Bitte wende dich per E-Mail an den Support."

Dieselbe Frage mit und ohne die Regel:

Bis zum ersten Text vergeht mehr Zeit, das Format bleibt gleich. Ein Client, der keine bestimmte Zahl von Chunks erwartet, deckt beide Fälle ohne Änderung ab. Fortschrittsereignisse kommen während der Wartezeit weiter an, weil sie keinen Antworttext enthalten. Blockiert die Regel, ersetzt die rejection_message die Antwort, und der Run wird abgeschlossen; der Client zeigt sie wie jede andere Antwort. Mit fail_run: true in der Konfiguration der Regel schlägt der Run stattdessen fehl, und stream_end enthält einen Fehler. Input Guardrails laufen vor dem Agenten und ändern am Streaming nichts.

Freigaben pausieren die Antwort

Ein freigabepflichtiges Tool hält den Run an, bis ein Teammitglied den Aufruf freigibt oder ablehnt. Die Verbindung bleibt so lange mit dieser Antwort beschäftigt, und spätere Nachrichten auf der Verbindung reihen sich dahinter ein. approval_pending nennt das Tool, sobald der Run pausiert, und gibt der UI damit ihren Wartezustand; approval_resolved meldet die Entscheidung (approved, rejected oder timeout), sobald sie gefallen ist. Ohne Fortschrittsereignisse kommt während der Wartezeit nichts an. Die Wartezeit zählt nicht zum Run-Timeout, und approval.timeout steht standardmäßig auf 3.600 Sekunden, eine lange Zeit für jemanden, der auf ein Chatfenster schaut. Ein kurzes Timeout (mindestens 30 Sekunden) und on_rejection: continue halten das Gespräch in Bewegung:

tools:
  - orders.lookup_order
  - orders.cancel_order

approval:
  tools:
    - orders.cancel_order
  timeout: 120
  on_rejection: continue
  message: "Eine Kundin oder ein Kunde möchte im Chat eine Bestellung stornieren."

Mit on_rejection: continue bekommt das Modell bei einer Ablehnung oder einem abgelaufenen Timeout eine Ablehnungsmeldung als Ergebnis des Tools, und der Agent erklärt der Kundin oder dem Kunden, was passiert ist. Mit dem Standardwert fail schlägt der Run fehl, und die Antwort endet mit einem Fehler. Nach der Entscheidung wird der Rest der Antwort gestreamt, und full_response enthält den Text von vor und nach der Pause. Der Artikel zu Human-in-the-Loop-Freigaben behandelt bedingte Freigaben und Benachrichtigungskanäle.

Fehler beenden die Antwort mit einer Fehlermeldung

Jede Nachricht endet mit stream_end (ohne Streaming mit response) oder mit einem error-Ereignis, solange die Verbindung nicht vorher abbricht. Ein Run, der fehlschlägt, abgebrochen wird oder sein Zeitlimit überschreitet, erzeugt stream_end mit einem Feld error statt full_response. Dieser Text richtet sich an das Team. Die UI zeigt der Kundschaft deshalb eine eigene Meldung, und das Team findet die Details im Trace des Runs. Ein abgelehntes Token ist kein Fehler: Seine StopProcessing-Meldung kommt als normale Antwort an. error-Ereignisse melden die anderen häufigen Probleme mit einem Code:

Auf einer inaktiven Verbindung sendet connic außerdem ping-Ereignisse, die der Client ignorieren kann. Die Referenz zum WebSocket-Connector beschreibt Einrichtung, Einstellungen und Nachrichtenformat.