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 enterprise‑grade pipeline.

Why Kinesis Beats WebSockets for LLM Streams

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:

1. Ephemeral connection – if the client disconnects, the in‑flight tokens are lost.
2. Single consumer – only the WebSocket server receives the data; you can’t add a logger, analytics, or a backup consumer without duplicating work.
3. Scaling limits – a single server must handle every concurrent chat, which becomes fragile under load.

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.

Quick code sketch: putting a token into Kinesis

// 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);
}

Key takeaway – With a few lines you can write a token to a durable, auto‑scaled pipe that any Lambda can later read.

Setting Up a Kinesis Data Stream for Claude Tokens

What you need to know first

• Shard – a unit of capacity; each shard can ingest up to 1 MiB / second (≈ 1000 records) and serve up to 2 MiB / second to consumers.
• Enhanced fan‑out – a feature that gives each consumer its own read throughput, avoiding the 2 MiB shared limit. It only works when the consumer is registered by ARN (Amazon Resource Name). Forgetting this registration falls back to the shared 2 MiB pool and adds ~200 ms latency.
• Iterator – a pointer that tells Kinesis where to start reading. It expires after 5 minutes, so a consumer must refresh it regularly.

Step‑by‑step creation with the SDK

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();

Tip – After creating the stream, give it a minute to become ACTIVE before you register the consumer; otherwise the registration request will fail.

Gotchas to keep in mind

Lambda Consumer with Node.js 22: Decoding and Forwarding

Why a Lambda is a good fit

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.

Minimal Lambda handler

// 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`,
  });
}

Explanation of key lines

• ShardIteratorType: "LATEST" starts reading from the newest token, ideal for live chat.
• ConsumerARN registers the consumer for enhanced fan‑out; without it you’d see the 200 ms latency penalty.
• The while (true) loop keeps the Lambda busy until AWS times it out (default 15 min). In production you’d configure a timeout that matches your latency budget.
• ProvisionedThroughputExceededException signals that the read rate is too high; the exponential backoff reduces pressure without crashing.

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.

Putting It Together: Real‑Time Claude Chat Proxy

End‑to‑end flow (text diagram)

1. Browser opens an SSE connection to /chat/events.
2. API Gateway routes the request to a thin Lambda (proxy‑handler).
3. proxy‑handler calls Claude’s completion endpoint with stream=true.
4. Claude streams token chunks back to the Lambda; each chunk is written to the Kinesis stream using PutRecord.
5. The consumer Lambda (code from the previous section) reads the tokens from Kinesis and forwards them to the browser’s SSE endpoint.
6. The browser renders tokens as they arrive, giving a live‑typing feel.

Minimal proxy Lambda that talks to Claude and pushes into Kinesis

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

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.

Common pitfalls for this proxy

• Missing stream:true – Claude will wait until the whole answer is ready, defeating the real‑time goal.
• Incorrect partition key – If you use a constant key, all chats share a single shard, quickly hitting the 1 MiB/s write ceiling.
• Forgot to set CONTENT_TYPE: text/event-stream on the SSE endpoint – browsers will buffer the response, appearing as if nothing is happening.

Handling Errors and Observability

Why observability matters

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.

Simple CloudWatch logging and metrics

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

Insert recordMetric("TokensProcessed") after each successful token forward, and recordMetric("ThrottleBackoff") inside the backoff catch block. CloudWatch dashboards can then show spikes.

Dealing with iterator expiration

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;
}

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.

The Takeaway

You now have a complete, production‑grade pattern for streaming Claude’s token output through Kinesis and a Node.js 22 Lambda.

• Kinesis provides durability and multi‑consumer fan‑out, eliminating the fragility of a single WebSocket server.
• Enhanced fan‑out only works when you register the consumer by ARN; forgetting this leaves you with higher latency and extra read costs.
• Shards are the unit of capacity; start with multiple shards and monitor throughput to avoid throttling.
• Lambda can read directly from Kinesis, decode token bytes, and forward them with SSE using the built‑in fetch API.
• Handle ProvisionedThroughputExceededException with exponential backoff and refresh stale shard iterators to keep the pipeline alive.
• Add lightweight CloudWatch metrics and a DLQ to gain visibility and prevent silent data loss.

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!

Transparency notice

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.