Get started

This guide walks you through the process of creating a K2 stream, writing records to it, and consuming through a subscription.

By the end of this guide, you will have:

• Created a stream with the Cloudflare API.
• Produced a batch of records to the stream over HTTP.
• Created a subscription that tracks your read position.
• Consumed a batch of records and acknowledged it.

1. Sign up for a Cloudflare account ↗︎ and subscribe to the Workers Paid plan. K2 is not available on the Workers Free plan.
2. Find your account ID.
3. Install curl ↗︎ and jq ↗︎.

A K2 stream is a durable, append-only log of records. Each record contains a binary content payload and optional string headers.

Producers append records to a stream in batches. Each batch is written atomically: either every record in the batch is stored, or none are.

Consumers read from a stream through a subscription. A subscription tracks a position in the stream. Multiple subscriptions on the same stream read independently of each other.

When a consumer reads from a subscription, K2 leases a batch of records to that consumer. The consumer acknowledges the batch after it processes the records. If the consumer does not acknowledge the batch before the lease expires, K2 delivers the records again. This gives you at-least-once delivery.

You need an API token to create streams and to read from them.

Export your account ID and API token as shell variables. The commands in this guide use these variables.

export ACCOUNT_ID=<YOUR_ACCOUNT_ID>
export CLOUDFLARE_API_TOKEN=<YOUR_API_TOKEN>

Create a stream named orders. This request enables the HTTP endpoint and requires an API token to produce records to it.

curl "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/k2/streams" \
  --request POST \
  --header "Authorization: Bearer $CLOUDFLARE_API_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{
    "name": "orders",
    "http": {
      "enabled": true,
      "authentication": true
    }
  }'

{
	"success": true,
	"errors": [],
	"messages": [],
	"result": {
		"id": "241fa65b438a4d539a19371f58bfdae0",
		"name": "orders",
		"retention_seconds": 604800,
		"endpoint": "https://241fa65b438a4d539a19371f58bfdae0.k2.cloudflarestorage.com",
		"http": {
			"enabled": true,
			"authentication": true
		},
		"worker_binding": {
			"enabled": true
		},
		"created_at": "2026-09-24T21:19:19.246Z",
		"modified_at": "2026-09-24T21:19:19.246Z"
	}
}

The response includes:

• id: The stream ID. You use it to build the stream endpoint.
• endpoint: The base URL for producing and consuming records.
• retention_seconds: How long K2 keeps records. The default is seven days (604800 seconds).

Export the stream ID and endpoint as shell variables:

export STREAM_ID=<STREAM_ID>
export K2_ENDPOINT=https://$STREAM_ID.k2.cloudflarestorage.com

Send a batch of two records to the /produce endpoint of your stream. Record content must be standard base64. In this example, each record contains a base64-encoded JSON order event.

curl "$K2_ENDPOINT/produce" \
  --request POST \
  --header "Authorization: Bearer $CLOUDFLARE_API_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{
    "records": [
      {
        "content": "eyJvcmRlcl9pZCI6MTAwMSwic3RhdHVzIjoiY3JlYXRlZCJ9",
        "headers": { "event-type": "order.created" }
      },
      {
        "content": "eyJvcmRlcl9pZCI6MTAwMiwic3RhdHVzIjoiY3JlYXRlZCJ9",
        "headers": { "event-type": "order.created" }
      }
    ]
  }'

{ "success": true }

A success: true response means K2 stored every record in the batch.

To encode your own payloads, pipe them through base64:

printf '%s' '{"order_id":1001,"status":"created"}' | base64

A subscription tracks which records a consumer has processed. Create a subscription named orders-processor that starts from the earliest record in the stream.

curl "$K2_ENDPOINT/subscriptions" \
  --request POST \
  --header "Authorization: Bearer $CLOUDFLARE_API_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{
    "name": "orders-processor",
    "start_at": { "type": "earliest" }
  }'

{
	"result": { "id": "e2f747f8bccf453eb2767496779cd135" },
	"success": true,
	"errors": [],
	"messages": []
}

The request body contains:

• name: 1 to 128 letters, numbers, underscores, or hyphens. Must be unique within the stream.
• start_at.type: earliest reads from the oldest retained record. latest skips records that exist when you create the subscription.

Creating a subscription with the same name and settings returns the existing subscription ID, so the request is safe to retry.

Export the subscription ID as a shell variable:

export SUBSCRIPTION_ID=<SUBSCRIPTION_ID>

Request a batch of up to 100 records. The worker_id identifies your consumer. Each worker can hold one lease at a time.

curl "$K2_ENDPOINT/subscriptions/$SUBSCRIPTION_ID/consume" \
  --request POST \
  --header "Authorization: Bearer $CLOUDFLARE_API_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{
    "worker_id": "worker-1",
    "max_records": 100
  }'

{
	"result": {
		"batch_id": "4f1c2a9e8b7d4c6f9a0e1d2c3b4a5968",
		"leased_until_ms": 1790165100000,
		"records": [
			{
				"timestamp_ms": 1790164800000,
				"content": "eyJvcmRlcl9pZCI6MTAwMSwic3RhdHVzIjoiY3JlYXRlZCJ9",
				"headers": { "event-type": "order.created" }
			},
			{
				"timestamp_ms": 1790164800000,
				"content": "eyJvcmRlcl9pZCI6MTAwMiwic3RhdHVzIjoiY3JlYXRlZCJ9",
				"headers": { "event-type": "order.created" }
			}
		]
	},
	"success": true,
	"errors": [],
	"messages": []
}

The response contains:

• batch_id: The ID of the leased batch. You need it to acknowledge the batch.
• leased_until_ms: When the lease expires, in milliseconds since the Unix epoch. The lease lasts five minutes.
• records: The records in the batch. timestamp_ms is the time K2 received the record, in milliseconds since the Unix epoch. content is base64.

To decode the record contents, pipe the response through jq:

curl --silent "$K2_ENDPOINT/subscriptions/$SUBSCRIPTION_ID/consume" \
  --request POST \
  --header "Authorization: Bearer $CLOUDFLARE_API_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{"worker_id": "worker-1", "max_records": 100}' \
  | jq -r '.result.records[].content | @base64d'

{"order_id":1001,"status":"created"}
{"order_id":1002,"status":"created"}

Because worker-1 still holds its lease, this second request returns the same batch and refreshes the lease. Use this behavior to recover a batch if your consumer loses a response.

If there are no records to read, the response contains an empty records array and batch_id is null. Wait before you poll again.

Export the batch ID from the response as a shell variable:

export BATCH_ID=<BATCH_ID>

After you process the records, acknowledge the batch. This advances the subscription past these records and releases the lease so the worker can read the next batch.

curl "$K2_ENDPOINT/subscriptions/$SUBSCRIPTION_ID/batches/$BATCH_ID/ack" \
  --request POST \
  --header "Authorization: Bearer $CLOUDFLARE_API_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{ "worker_id": "worker-1" }'

{ "result": {}, "success": true, "errors": [], "messages": [] }

If processing fails, send the same request to /nack instead of /ack. K2 releases the lease without advancing the subscription, and delivers the records again.

Run the consume request from step 5 again. Because you acknowledged the first batch, the response contains no records.

Delete the subscription:

curl "$K2_ENDPOINT/subscriptions/$SUBSCRIPTION_ID" \
  --request DELETE \
  --header "Authorization: Bearer $CLOUDFLARE_API_TOKEN"

{
	"result": { "id": "e2f747f8bccf453eb2767496779cd135" },
	"success": true,
	"errors": [],
	"messages": []
}

Delete the stream and all of its records:

curl "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/k2/streams/$STREAM_ID" \
  --request DELETE \
  --header "Authorization: Bearer $CLOUDFLARE_API_TOKEN"

{ "success": true, "errors": [], "messages": [], "result": {} }