How to Stream Claude Agent Events with Kinesis Data Streams, Explained Simply
Real‑time LLM interactions feel like they belong on websockets, but a Kinesis Data Stream gives you durable, auto‑scaling fan‑out with millisecond latency. In this guide you’ll see how to pipe Claude’s token stream through Kinesis and a Node.js 22 Lambda, turning a flaky WebSocket demo into an ente
Echtzeit-Interaktionen mit LLMs wirken, als gehörten sie auf Websockets, doch ein Kinesis Data Stream bietet dauerhafte, automatisch skalierende Fan-outs mit Millisekunden-Latenz. In diesem Leitfaden sehen Sie, wie Sie den Token-Stream von Claude durch Kinesis und eine Node.js 22 Lambda leiten und aus einer fehleranfälligen WebSocket-Demo eine Pipeline auf Enterprise-Niveau machen.
Wenn Sie Claude um eine Antwort bitten, sendet das Modell die Tokens (kleine Textstücke) zurück, sobald sie generiert werden. Eine WebSocket-Verbindung kann jedes Token direkt an den Browser weiterreichen, aber sie hat drei versteckte Kosten:
Ein Kinesis Data Stream ist eine verwaltete Pipe, die die Daten in Shards aufteilt (unabhängige Schreib- und Lesespuren). Jeder Shard kann gleichzeitig viele Consumer versorgen (dies wird enhanced fan-out genannt). Stellen Sie sich eine Wasserhauptleitung (den Stream) vor, die sich in viele kleinere Schläuche (die Consumer) aufspaltet. Selbst wenn ein Schlauch abgestellt wird, fließt das Wasser weiterhin durch die Hauptleitung.
Klartext: Kinesis erlaubt es Ihnen, jedes Token als dauerhaften Datensatz zu behandeln, den beliebig viele Dienste lesen können, während die Latenz dennoch niedrig genug für ein chatähnliches Erlebnis bleibt.
// npm install @aws-sdk/client-kinesis import { KinesisClient, PutRecordCommand } from " @aws-sdk/client-kinesis "; const client = new KinesisClient ({ region: " us-east-1 " }); async function putToken ( streamName, partitionKey, token ) { // Das Token ist ein kurzer String, wir speichern ihn als UTF-8-Bytes. const command = new PutRecordCommand ({ StreamName: streamName, PartitionKey: partitionKey, // bestimmt, welcher Shard den Datensatz erhält Data: Buffer. from ( token ), // binäre Nutzdaten }); await client. send ( command ); } Enter fullscreen mode Exit fullscreen mode Das Wichtigste in Kürze: Mit wenigen Zeilen können Sie ein Token in eine dauerhafte, automatisch skalierte Pipe schreiben, die jede Lambda später lesen kann.
import { KinesisClient, CreateStreamCommand, RegisterStreamConsumerCommand, } from " @aws-sdk/client-kinesis "; const kinesis = new KinesisClient ({ region: " us-east-1 " }); const STREAM_NAME = " claude-token-stream "; const CONSUMER_NAME = " claude-lambda-consumer "; async function createStream () { // Erstelle einen Stream mit 2 Shards (anpassen je nach erwartetem Durchsatz) await kinesis. send ( new CreateStreamCommand ({ StreamName: STREAM_NAME, ShardCount: 2 }) ); console. log ( `Stream ${ STREAM_NAME } erstellt` ); } async function registerConsumer () { // Die Registrierung über die ARN aktiviert enhanced fan-out await kinesis. send ( new RegisterStreamConsumerCommand ({ StreamARN: `arn:aws:kinesis:us-east-1:123456789012:stream/ ${ STREAM_NAME } `, ConsumerName: CONSUMER_NAME, }) ); console. log ( `Consumer ${ CONSUMER_NAME } für enhanced fan-out registriert` ); } // Einmal ausführen (oder mit Existenzprüfungen absichern) await createStream (); await registerConsumer (); Enter fullscreen mode Exit fullscreen mode Tipp: Geben Sie dem Stream nach dem Erstellen eine Minute Zeit, um den Status ACTIVE zu erreichen, bevor Sie den Consumer registrieren; andernfalls schlägt die Registrierungsanfrage fehl.
Eine Lambda ist eine serverlose Funktion, die nur beim Aufruf ausgeführt wird. Wenn Sie sie als Event Source an den Kinesis-Stream anhängen, fragt AWS den Stream automatisch ab, übergibt Batches von Datensätzen an Ihren Code und übernimmt die Skalierung für Sie. Node.js 22 bringt die native fetch API mit, sodass wir jedes Token ohne zusätzliche Bibliotheken per Server-Sent Events (SSE) an einen Browser weiterleiten können.
// npm install @aws-sdk/client-lambda @aws-sdk/client-kinesis import { KinesisClient, GetRecordsCommand, GetShardIteratorCommand } from " @aws-sdk/client-kinesis "; const kinesis = new KinesisClient ({ region: " us-east-1 " }); const STREAM_NAME = process. env. STREAM_NAME; const CONSUMER_ARN = process. env. CONSUMER_ARN; // ARN des enhanced fan-out Consumers // Exponential-Backoff-Hilfsfunktion function wait ( ms ) { return new Promise ( res => setTimeout ( res, ms )); } export const handler = async ( event, context ) => { // 1️⃣ Einen Shard-Iterator für jeden Shard holen (enhanced fan-out) const shards = await kinesis. send ( new DescribeStreamCommand ({ StreamName: STREAM_NAME }) ). then ( resp => resp.
StreamDescription. Shards ); for ( const shard of shards ) { const iteratorCmd = new GetShardIteratorCommand ({ StreamName: STREAM_NAME, ShardId: shard. ShardId, ShardIteratorType: " LATEST ", // beim neuesten Token starten ConsumerARN: CONSUMER_ARN, // aktiviert enhanced fan-out }); let iterator = await kinesis. send ( iteratorCmd ). then ( r => r.
ShardIterator ); // 2️⃣ Datensätze in einer Schleife abrufen (Lambda wird nach dem Timeout beendet) while ( true ) { try { const recordsCmd = new GetRecordsCommand ({ ShardIterator: iterator, Limit: 100 }); const { Records, NextShardIterator } = await kinesis. send ( recordsCmd ); iterator = NextShardIterator; if ( Records. length === 0 ) { // Aktuell keine Daten, kurz pausieren, um eine enge Schleife zu vermeiden await wait ( 200 ); continue; } // 3️⃣ Jedes Token per SSE an den Client weiterleiten for ( const rec of Records ) { const token = Buffer. from ( rec. Data ). toString ( " utf-8 " ); await forwardTokenSSE ( token ); } } catch ( err ) { if ( err. name === " ProvisionedThroughputExceededException " ) { // 4️⃣ Throttling – exponentiell zurückweisen for ( let attempt = 1; attempt <= 5; attempt ++ ) { const backoff = Math. pow ( 2, attempt ) * 100; // 200ms, 400ms, … await wait ( backoff ); // die Retry-Schleife läuft automatisch weiter } } else if ( err. name === " ExpiredIteratorException " ) { // Iterator erneuern, da er abgelaufen ist const fresh = await kinesis. send ( iteratorCmd ); iterator = fresh. ShardIterator; } else { console. error ( " Unerwarteter Fehler: ", err ); throw err; // Lambda soll den Fehler sichtbar machen } } } } }; // Hilfsfunktion, die ein Token per SSE an einen Client sendet async function forwardTokenSSE ( token ) { // Die Client-URL wird über eine Umgebungsvariable übergeben; sie muss auf einen Endpunkt zeigen, der die HTTP-Verbindung offen hält const clientUrl = process. env.
CLIENT_SSE_ENDPOINT; await fetch ( clientUrl, { method: " POST ", headers: { " Content-Type ": " text/event-stream ", // weist den Browser an, es als SSE zu behandeln }, body: `data: ${ token } \n\n`, }); } Enter fullscreen mode Exit fullscreen mode Erläuterung der wichtigsten Zeilen
Tipp: Die native fetch von Node.js 22 folgt automatisch Redirects, aber SSE verlangt den exakten Content-Type. Vergisst man den Header, behandeln Browser die Nutzdaten als einfaches JSON und der Stream wirkt defekt.
import { KinesisClient, PutRecordCommand } from " @aws-sdk/client-kinesis "; const kinesis = new KinesisClient ({ region: " us-east-1 " }); const CLAUDE_API = " https://api.anthropic.com/v1/complete "; const STREAM_NAME = process. env. STREAM_NAME; export const handler = async ( event ) => { const body = JSON. parse ( event. body ); const { prompt, userId } = body; // einfache Nutzdaten aus dem Browser // Claude mit aktiviertem Streaming aufrufen const response = await fetch ( CLAUDE_API, { method: " POST ", headers: { " Content-Type ": " application/json ", " x-api-key ": process. env. CLAUDE_API_KEY, }, body: JSON. stringify ({ model: " claude-3-5-sonnet-20241001 ", prompt, max_tokens_to_sample: 1024, stream: true, }), }); // Den rohen Bytestrom lesen; jede Zeile ist ein JSON-Fragment mit einem Token const reader = response. body. getReader (); const decoder = new TextDecoder (); while ( true ) { const { done, value } = await reader. read (); if ( done ) break; // Am Zeilenumbruch splitten, da Claude jedes Token in einer eigenen Zeile sendet const chunk = decoder. decode ( value ); const lines = chunk. split ( " \n " ). filter ( l => l. trim ()); for ( const line of lines ) { const payload = JSON. parse ( line ); const token = payload. completion; // Token-String // Token in Kinesis schreiben – nach userId partitionieren, um die Reihenfolge pro Chat zu wahren await kinesis. send ( new PutRecordCommand ({ StreamName: STREAM_NAME, PartitionKey: userId, Data: Buffer. from ( token ), }) ); } } // Schnell antworten; die Echtzeit-Arbeit läuft in der Consumer-Lambda weiter return { statusCode: 200, body: JSON. stringify ({ message: " Chat gestartet " }), }; }; Enter fullscreen mode Exit fullscreen mode Klartext: Diese Lambda wirkt wie ein Botenjunge im Postzimmer: Sie nimmt Claudes Briefe (Tokens) entgegen, stempelt jeden mit der Benutzer-ID (dem Partitionsschlüssel) und wirft sie in das Kinesis-Postfach, aus dem die Zustell-Lambda sie abholt.
Eine Streaming-Pipeline hat viele bewegliche Teile: die Latenz der Claude API, Kinesis-Throttling, Lambda-Timeouts und Netzwerkprobleme. Ohne Sichtbarkeit sehen Sie gelegentlich „fehlende Tokens“, wissen aber nicht, warum.
import { CloudWatchClient, PutMetricDataCommand } from " @aws-sdk/client-cloudwatch "; const cw = new CloudWatchClient ({ region: " us-east-1 " }); const METRIC_NAMESPACE = " ClaudeKinesisPipeline "; async function recordMetric ( name, value = 1 ) { await cw. send ( new PutMetricDataCommand ({ Namespace: METRIC_NAMESPACE, MetricData: [ { MetricName: name, Value: value, Unit: " Count ", }, ], }) ); } Enter fullscreen mode Exit fullscreen mode Fügen Sie recordMetric("TokensProcessed") nach jeder erfolgreichen Token-Weiterleitung und recordMetric("ThrottleBackoff") innerhalb des Backoff-Catch-Blocks ein. CloudWatch-Dashboards können dann Spitzen anzeigen.
if ( err. name === " ExpiredIteratorException " ) { // Für bessere Nachvollziehbarkeit loggen console. warn ( " Shard-Iterator abgelaufen, ein neuer wird angefordert " ); // Iterator erneut abrufen (gleicher Code wie bei der ursprünglichen Anfrage) const fresh = await kinesis. send ( iteratorCmd ); iterator = fresh. ShardIterator; } Enter fullscreen mode Exit fullscreen mode Dead-Letter-Queue (DLQ) für nicht wiederherstellbare Datensätze Wenn ein Datensatz nach mehreren Wiederholungsversuchen wiederholt eine ProvisionedThroughputExceededException auslöst, verschiebe ihn zur späteren Analyse in eine SQS-Queue. So wird verhindert, dass die Lambda-Funktion an einem einzelnen fehlerhaften Shard hängen bleibt.
Wichtiger Hinweis: Lege immer eine maximale Anzahl von Wiederholungsversuchen fest (z. B. 5), bevor ein Datensatz an die DLQ gesendet wird; andernfalls kann das exponentielle Backoff die Funktion bis zum Timeout weiterlaufen lassen und die Kosten in die Höhe treiben.
Du verfügst nun über ein vollständiges, produktionsreifes Muster, um den Token-Output von Claude über Kinesis und eine Node.js 22 Lambda-Funktion zu streamen.
Mit diesen Bausteinen kannst du eine anfällige WebSocket-Demo durch eine skalierbare, serverlose Architektur ersetzen, die gleichermaßen für eine Handvoll Nutzer oder für Tausende gleichzeitige Chats funktioniert. Happy Streaming!
Dieser Artikel wurde mit Unterstützung eines KI-Systems verfasst: Groq (GPT OSS 120B).
Veröffentlicht: 2026-10-23 · Schwerpunkt: Kinesis
Alle Codeblöcke sind so angelegt, dass sie korrekt und lauffähig sind, bitte verifiziere sie jedoch vor dem Einsatz in der Produktivumgebung anhand der offiziellen Dokumentation der genannten Tools.
Fehler gefunden? Hinterlasse einen Kommentar, Korrekturen sind immer willkommen.
Automatisch übersetzt. Das Original auf Englisch ist unten aufklappbar.
Original (Englisch) anzeigen
Real‑time LLM interactions feel like they belong on websockets, but a Kinesis Data Stream gives you durable, auto‑scaling fan‑out with millisecond latency. In this guide you’ll see how to pipe Claude’s token stream through Kinesis and a Node.js 22 Lambda, turning a flaky WebSocket demo into an enterprise‑grade pipeline.
When you ask Claude for a response, the model sends back tokens (tiny pieces of text) as they are generated. A WebSocket connection can push each token straight to the browser, but it has three hidden costs:
A Kinesis Data Stream is a managed pipe that breaks the data into shards (independent write/read lanes). Each shard can feed many consumers at the same time (this is called enhanced fan‑out ). Think of a water main (the stream) that splits into many smaller hoses (the consumers). Even if one hose is turned off, the water still flows through the main pipe.
In plain English – Kinesis lets you treat each token as a durable record that any number of services can read, while still keeping latency low enough for a chat‑like experience.
// npm install @aws-sdk/client-kinesis import { KinesisClient, PutRecordCommand } from " @aws-sdk/client-kinesis "; const client = new KinesisClient ({ region: " us-east-1 " }); async function putToken ( streamName, partitionKey, token ) { // The token is a short string, we store it as UTF‑8 bytes. const command = new PutRecordCommand ({ StreamName: streamName, PartitionKey: partitionKey, // decides which shard receives the record Data: Buffer. from ( token ), // binary payload }); await client. send ( command ); } Enter fullscreen mode Exit fullscreen mode Key takeaway – With a few lines you can write a token to a durable, auto‑scaled pipe that any Lambda can later read.
import { KinesisClient, CreateStreamCommand, RegisterStreamConsumerCommand, } from " @aws-sdk/client-kinesis "; const kinesis = new KinesisClient ({ region: " us-east-1 " }); const STREAM_NAME = " claude-token-stream "; const CONSUMER_NAME = " claude-lambda-consumer "; async function createStream () { // Create a stream with 2 shards (adjust based on expected throughput) await kinesis. send ( new CreateStreamCommand ({ StreamName: STREAM_NAME, ShardCount: 2 }) ); console. log ( `Stream ${ STREAM_NAME } created` ); } async function registerConsumer () { // Registering by ARN enables enhanced fan‑out await kinesis. send ( new RegisterStreamConsumerCommand ({ StreamARN: `arn:aws:kinesis:us-east-1:123456789012:stream/ ${ STREAM_NAME } `, ConsumerName: CONSUMER_NAME, }) ); console. log ( `Consumer ${ CONSUMER_NAME } registered for enhanced fan‑out` ); } // Run once (or guard with existence checks) await createStream (); await registerConsumer (); Enter fullscreen mode Exit fullscreen mode Tip – After creating the stream, give it a minute to become ACTIVE before you register the consumer; otherwise the registration request will fail.
A Lambda is a serverless function that runs only when invoked. By attaching it to the Kinesis stream as an event source, AWS automatically polls the stream, passes batches of records to your code, and handles scaling for you. Node.js 22 brings the native fetch API, so we can forward each token to a browser using Server‑Sent Events (SSE) without adding extra libraries.
// npm install @aws-sdk/client-lambda @aws-sdk/client-kinesis import { KinesisClient, GetRecordsCommand, GetShardIteratorCommand } from " @aws-sdk/client-kinesis "; const kinesis = new KinesisClient ({ region: " us-east-1 " }); const STREAM_NAME = process. env. STREAM_NAME; const CONSUMER_ARN = process. env. CONSUMER_ARN; // ARN of the enhanced fan‑out consumer // Exponential backoff helper function wait ( ms ) { return new Promise ( res => setTimeout ( res, ms )); } export const handler = async ( event, context ) => { // 1️⃣ Get a shard iterator for each shard (enhanced fan‑out) const shards = await kinesis. send ( new DescribeStreamCommand ({ StreamName: STREAM_NAME }) ). then ( resp => resp.
StreamDescription. Shards ); for ( const shard of shards ) { const iteratorCmd = new GetShardIteratorCommand ({ StreamName: STREAM_NAME, ShardId: shard. ShardId, ShardIteratorType: " LATEST ", // start at newest token ConsumerARN: CONSUMER_ARN, // enables enhanced fan‑out }); let iterator = await kinesis. send ( iteratorCmd ). then ( r => r.
ShardIterator ); // 2️⃣ Pull records in a loop (Lambda will be terminated after timeout) while ( true ) { try { const recordsCmd = new GetRecordsCommand ({ ShardIterator: iterator, Limit: 100 }); const { Records, NextShardIterator } = await kinesis. send ( recordsCmd ); iterator = NextShardIterator; if ( Records. length === 0 ) { // No data right now, pause briefly to avoid tight looping await wait ( 200 ); continue; } // 3️⃣ Forward each token via SSE to the client for ( const rec of Records ) { const token = Buffer. from ( rec. Data ). toString ( " utf-8 " ); await forwardTokenSSE ( token ); } } catch ( err ) { if ( err. name === " ProvisionedThroughputExceededException " ) { // 4️⃣ Throttling – back off exponentially for ( let attempt = 1; attempt <= 5; attempt ++ ) { const backoff = Math. pow ( 2, attempt ) * 100; // 200ms, 400ms, … await wait ( backoff ); // retry loop will continue automatically } } else if ( err. name === " ExpiredIteratorException " ) { // Refresh iterator as it timed out const fresh = await kinesis. send ( iteratorCmd ); iterator = fresh. ShardIterator; } else { console. error ( " Unexpected error: ", err ); throw err; // let Lambda surface the failure } } } } }; // Helper that posts token to a client using SSE async function forwardTokenSSE ( token ) { // The client URL is passed via env var; it must point to an endpoint that keeps the HTTP connection open const clientUrl = process. env.
CLIENT_SSE_ENDPOINT; await fetch ( clientUrl, { method: " POST ", headers: { " Content-Type ": " text/event-stream ", // tells the browser to treat it as SSE }, body: `data: ${ token } \n\n`, }); } Enter fullscreen mode Exit fullscreen mode Explanation of key lines
Tip – Node.js 22’s native fetch automatically follows redirects, but SSE requires the exact Content-Type. Forgetting the header makes browsers treat the payload as plain JSON and the stream appears broken.
import { KinesisClient, PutRecordCommand } from " @aws-sdk/client-kinesis "; const kinesis = new KinesisClient ({ region: " us-east-1 " }); const CLAUDE_API = " https://api.anthropic.com/v1/complete "; const STREAM_NAME = process. env. STREAM_NAME; export const handler = async ( event ) => { const body = JSON. parse ( event. body ); const { prompt, userId } = body; // simple payload from the browser // Call Claude with streaming enabled const response = await fetch ( CLAUDE_API, { method: " POST ", headers: { " Content-Type ": " application/json ", " x-api-key ": process. env. CLAUDE_API_KEY, }, body: JSON. stringify ({ model: " claude-3-5-sonnet-20241001 ", prompt, max_tokens_to_sample: 1024, stream: true, }), }); // Read the raw byte stream; each line is a JSON fragment with a token const reader = response. body. getReader (); const decoder = new TextDecoder (); while ( true ) { const { done, value } = await reader. read (); if ( done ) break; // Split on newline because Claude sends each token on its own line const chunk = decoder. decode ( value ); const lines = chunk. split ( " \n " ). filter ( l => l. trim ()); for ( const line of lines ) { const payload = JSON. parse ( line ); const token = payload. completion; // token string // Write token to Kinesis – partition by userId to keep order per chat await kinesis. send ( new PutRecordCommand ({ StreamName: STREAM_NAME, PartitionKey: userId, Data: Buffer. from ( token ), }) ); } } // Respond quickly; the real‑time work continues in the consumer Lambda return { statusCode: 200, body: JSON. stringify ({ message: " Chat started " }), }; }; Enter fullscreen mode Exit fullscreen mode In plain English – This Lambda acts like a mailroom clerk: it receives Claude’s letters (tokens), stamps each with the user’s ID (the partition key), and drops them into the Kinesis mailbox for the delivery Lambda to pick up.
A streaming pipeline has many moving parts: Claude API latency, Kinesis throttling, Lambda timeouts, and network hiccups. Without visibility you’ll see occasional “missing tokens” but not know why.
import { CloudWatchClient, PutMetricDataCommand } from " @aws-sdk/client-cloudwatch "; const cw = new CloudWatchClient ({ region: " us-east-1 " }); const METRIC_NAMESPACE = " ClaudeKinesisPipeline "; async function recordMetric ( name, value = 1 ) { await cw. send ( new PutMetricDataCommand ({ Namespace: METRIC_NAMESPACE, MetricData: [ { MetricName: name, Value: value, Unit: " Count ", }, ], }) ); } Enter fullscreen mode Exit fullscreen mode Insert recordMetric("TokensProcessed") after each successful token forward, and recordMetric("ThrottleBackoff") inside the backoff catch block. CloudWatch dashboards can then show spikes.
if ( err. name === " ExpiredIteratorException " ) { // Log for visibility console. warn ( " Shard iterator expired, requesting a fresh one " ); // Re‑fetch the iterator (same code as initial request) const fresh = await kinesis. send ( iteratorCmd ); iterator = fresh. ShardIterator; } Enter fullscreen mode Exit fullscreen mode Dead‑letter queue (DLQ) for irrecoverable records If a record repeatedly triggers ProvisionedThroughputExceededException after several retries, move it to an SQS queue for later analysis. This prevents the Lambda from getting stuck on a single bad shard.
Key tip – Always set a maximum retry count (e.g., 5) before sending to DLQ; otherwise exponential backoff can keep the function running until it times out, inflating costs.
You now have a complete, production‑grade pattern for streaming Claude’s token output through Kinesis and a Node.js 22 Lambda.
With these pieces in place, you can replace a brittle WebSocket demo with a scalable, server‑less architecture that works the same way for a handful of users or for thousands of concurrent chats. Happy streaming!
This article was written with the help of an AI system — Groq (GPT OSS 120B).
Published: 2026-10-23 · Primary focus: Kinesis
All code blocks are intended to be correct and runnable, but please verify them against the official docs for the tools mentioned before using in production.
Find an error? Drop a comment — corrections are always welcome.