WebSocket

So funktioniert WebSocket

WebSocket-Verbindungen ermöglichen Full-Duplex-Kommunikation für Chats in Echtzeit. Anders als HTTP-Webhooks halten sie dauerhafte Verbindungen für bidirektionale Nachrichten. Das eignet sich für Chat-Anwendungen, Streaming-Antworten, Gespräche mit mehreren Nachrichten und interaktive Agenten.

// Mit WebSocket verbinden
const ws = new WebSocket('<websocket-url>');

// Authentifizieren (standardmäßig erforderlich)
ws.onopen = () => {
  ws.send(JSON.stringify({ secret: '<your-secret-key>' }));
};

// Verbindungsbestätigung verarbeiten
ws.onmessage = (event) => {
  const data = JSON.parse(event.data);
  if (data.type === 'connected') {
    console.log('Connector-Run-ID:', data.connector_run_id);
    console.log('Agents:', data.agents);
  }
};

Die beim Verbindungsaufbau zurückgegebene connector_run_id identifiziert die gespeicherte Session der Verbindung.

// Nachricht senden
ws.send(JSON.stringify({
  type: 'message',
  id: 'msg-001',
  payload: {
    message: 'Wie ist das Wetter heute?',
    context: { location: 'New York' }
  }
}));

Das Feld id ist optional, hilft aber bei der Korrelation von Antworten.

Füge der Payload ein files-Array hinzu, um Dokumente, Bilder oder Audio an ein multimodales Modell zu senden. Die Bytes jeder Datei sind Base64-kodiert:

// Nachricht mit Base64-kodierten Dateien senden
ws.send(JSON.stringify({
  type: 'message',
  id: 'msg-002',
  payload: {
    message: 'Fasse dieses Dokument zusammen',
    files: [
      {
        name: 'invoice.pdf',
        mime_type: 'application/pdf',
        data: '<base64-bytes>',
        size: 12456
      }
    ]
  }
}));

Dabei gilt dieselbe Konvention wie für Multipart-Webhook-Uploads: Jeder Eintrag unter files wird zu einem Binärteil des LLM-Inputs; der führende Textteil ist die Payload, aus der files entfernt wurde. Die Middleware kann die rohe Payload über context["payload"] prüfen und Binärteile in before() hinzufügen oder entfernen.

// Ereignisse einer Streaming-Antwort
{ "type": "ack", "id": "msg-001", "message_number": 1 }

{ "type": "stream_start", "id": "msg-001", "agent": "assistant" }

{ "type": "stream_chunk", "id": "msg-001", "agent": "assistant", "chunk": "Das Wetter in " }
{ "type": "stream_chunk", "id": "msg-001", "agent": "assistant", "chunk": "New York ist sonnig." }

{
  "type": "stream_end",
  "id": "msg-001",
  "agent": "assistant",
  "full_response": "Das Wetter in New York ist sonnig.",
  "token_usage": { "input_tokens": 45, "output_tokens": 15, "total_tokens": 60 }
}

Ereignistypen: ack (empfangen), stream_start, stream_chunk (Textfragment), stream_end (vollständig, einschließlich Token-Nutzung)

Agenten mit Output Guardrails antworten nach Abschluss des Runs immer mit einem einzigen stream_chunk: Es wird kein Textfragment gesendet, bevor die Guardrails die vollständige Antwort geprüft haben. Implementiere den Client anhand des dokumentierten Ereignisablaufs. Er darf keine feste Anzahl von Chunks voraussetzen.

Modus ohne Streaming

Wenn Streaming deaktiviert ist, wird die vollständige Antwort in einer einzigen Nachricht gesendet:

// Nicht gestreamte Antwort (bei deaktiviertem Streaming)
{ "type": "ack", "id": "msg-001", "message_number": 1 }

{
  "type": "response",
  "id": "msg-001",
  "agent": "assistant",
  "response": "Das Wetter in New York ist sonnig.",
  "token_usage": { "input_tokens": 45, "output_tokens": 15, "total_tokens": 60 }
}

Jede Verbindung erstellt eine eigene Session für den Gesprächsverlauf. Bei mehreren Nachrichten in derselben Session behält der Agent den Kontext. Mit dem Schließen der Verbindung endet die Session.

Python-Beispiel

import asyncio
import websockets
import json

async def chat():
    uri = "<websocket-url>"

    async with websockets.connect(uri) as ws:
        # Authentifizieren
        await ws.send(json.dumps({"secret": "<your-secret-key>"}))
        response = await ws.recv()
        print(f"Verbunden: {json.loads(response)}")

        # Nachricht senden
        await ws.send(json.dumps({"message": "Hallo!"}))

        # Streaming-Antwort empfangen
        while True:
            msg = await ws.recv()
            data = json.loads(msg)
            if data["type"] == "stream_chunk":
                print(data["chunk"], end="", flush=True)
            elif data["type"] == "stream_end":
                print(f"\nFertig! Tokens: {data['token_usage']}")
                break

asyncio.run(chat())

Installation mit pip install websockets

• Ping/Pong-Keepalive, ordnungsgemäßes Schließen, Session-Timeout und Nachrichtenlimits