AWS Kinesis Data Streams ExecModule

Overview

AwsKinesisModule brings governed real-time event streaming into ValkyrAI workflows. It discovers streams and shards, publishes single or batch records, creates short-lived shard iterators, reads bounded record pages, and performs explicitly confirmed stream creation, scaling, and deletion. It uses ValkyrAI's native map ABI and a verified IntegrationAccount; credentials never belong in workflow input.

The connector implements ten operations:

• list_streams, describe_stream, and list_shards discover bounded stream topology and status.
• put_record publishes one UTF-8, base64, or canonical JSON record.
• put_records publishes up to 500 records in one provider request and reports every partial failure without retrying the batch.
• get_shard_iterator obtains an expiring iterator at a reviewed stream position.
• get_records reads one bounded record page and returns record bytes as base64.
• create_stream, update_shard_count, and delete_stream require explicit confirmation and make one provider call.

n8n's AWS nodes and shared AWS credential model demonstrate useful workflow patterns: one reusable credential, region-aware resources, bounded collection operations, pagination, and composable node output. Its current built-in AWS credential documentation does not list a Kinesis node, so ValkyrAI adds a native streaming connector with strict input allowlists, record-size bounds, explicit iterator semantics, partial-batch receipts, read-only retry, and lifecycle confirmations.

Usage

1. Identify the exact streams and operations the workflow needs.
2. Create a least-privilege IAM principal restricted to those kinesis:* actions and stream ARNs.
3. Store its access key ID and secret access key in an AWS IntegrationAccount; verify it and keep it in READY status.
4. Add AwsKinesisModule and bind that account through ExecModuleConfig.authConfig.integrationAccount.
5. Select one operation and supply only its documented fields. Unexpected fields fail before provider access.
6. Put a stable event ID in each record body when downstream consumers need deduplication.
7. Reconcile the stream, consumer state, or downstream event store before replaying a timed-out write or lifecycle mutation.

Never place an AWS access key, secret key, session token, or credential object in module input. Those fields are rejected before any network request. Shard iterators are short-lived bearer capabilities; keep them only in ACL-scoped workflow state and never log them.

Inputs

Name	Type	Required for	Default	Constraints
operation	string	Every execution	None	One of the ten documented operations.
region	string	Every execution	None	AWS region such as us-west-2; custom endpoints are not accepted.
streamName	string	Every operation except list_streams and get_records	None	1-128 letters, digits, underscore, hyphen, or period.
nextToken	string	Optional list field	None	Opaque provider token, at most 4,096 characters.
limit	integer	Optional list/read field	100	List total: 1-10,000. get_records: 1-1,000.
returnAll	boolean	Optional list field	false	Follow list pages until limit, exhaustion, or the 100-page guard.
partitionKey	string	put_record; each batch record	None	1-256 safe UTF-8 bytes.
data	object, array, or string	put_record; each batch record	None	1 byte through 1 MiB after decoding or serialization.
dataEncoding	string	Optional record field	json	json, utf8, or base64.
records	array	put_records	None	1-500 record objects; total decoded data and partition keys capped at 5 MiB.
explicitHashKey	string	Optional record field	None	Unsigned 128-bit decimal value.
sequenceNumberForOrdering	string	Optional put_record field	None	Prior decimal sequence number for same-shard ordering.
shardId	string	get_shard_iterator	None	AWS shard identifier such as shardId-000000000000.
iteratorType	string	get_shard_iterator	None	LATEST, TRIM_HORIZON, AT_SEQUENCE_NUMBER, AFTER_SEQUENCE_NUMBER, or AT_TIMESTAMP.
startingSequenceNumber	string	Sequence iterator types	None	Required only for AT_SEQUENCE_NUMBER and AFTER_SEQUENCE_NUMBER.
timestamp	string	AT_TIMESTAMP	None	ISO-8601 UTC instant, required only for AT_TIMESTAMP.
shardIterator	string	get_records	None	Nonblank opaque iterator, at most 512 characters.
capacityMode	string	Optional create_stream field	ON_DEMAND	ON_DEMAND or PROVISIONED.
shardCount	integer	Provisioned create_stream	1	1-1,000; rejected for on-demand creation.
targetShardCount	integer	update_shard_count	None	1-10,000; AWS account and resharding limits still apply.
confirmCreate	boolean	create_stream	false	Must be exactly true.
confirmScale	boolean	update_shard_count	false	Must be exactly true.
confirmDelete	boolean	delete_stream	false	Must be exactly true.
enforceConsumerDeletion	boolean	Optional delete_stream field	false	Allows deletion while registered enhanced fan-out consumers exist.

json encoding accepts an object, array, or string containing a JSON object/array and serializes it deterministically. utf8 and base64 require string input. The connector intentionally caps each record at 1 MiB even where an account may have newer large-record support; this portable bound works across standard streams and protects workflow memory.

Outputs

Every call returns stable status, operation, and attempts fields.

Name	Type	When present	Description
status	string	Always	success or error.
operation	string	Always	Normalized operation.
attempts	integer	Always	Provider calls including pages and retry-safe reads.
resultType	string	Success	streams, stream, shards, record, record_results, shard_iterator, records, or stream_lifecycle.
items	array	List, batch, or record success	Bounded normalized streams, shards, batch receipts, or records.
count	integer	Items are present	Number of returned items.
pages	integer	List success	Provider list pages consumed.
hasMore	boolean	List success	Whether AWS returned another page token.
nextToken	string	Another list page exists	Opaque continuation token.
data	object	Success	Stream metadata, acceptance receipt, batch counts, iterator, lag, or lifecycle state.
requestId	string	AWS supplies one	Bounded request reference for reconciliation.
error	object	Failure	Safe {code, message, httpStatus?, retryable} details.

get_records returns each record's dataBase64 and dataBytes, plus partition key, sequence number, arrival timestamp, and encryption type where present. It never guesses whether bytes are text or JSON. Decode only under the downstream schema and trust policy for that stream.

put_records can return status: success with data.partialFailure: true. Inspect data.failedRecordCount and each ordered item. Successful items contain a sequence number and shard ID; failed items contain only a bounded provider error code. Provider error messages are omitted because they can expose account, stream, or shard details.

IntegrationAccount Requirements

Bind one AWS IntegrationAccount through the normalized ExecModule authentication relationship:

Field	Requirement
Provider	Amazon Web Services / Kinesis Data Streams
status	Must be READY.
verified	Must be true.
apiKey SecureField	AWS access key ID.
password SecureField	AWS secret access key.

Grant only the actions enabled for the workflow:

• Discovery: kinesis:ListStreams, kinesis:DescribeStreamSummary, and kinesis:ListShards.
• Producers: kinesis:PutRecord and kinesis:PutRecords.
• Consumers: kinesis:GetShardIterator and kinesis:GetRecords.
• Lifecycle: kinesis:CreateStream, kinesis:UpdateShardCount, and kinesis:DeleteStream only where explicitly required.
• Encrypted streams may also require the corresponding KMS permissions under the key policy.

Restrict stream ARNs and IAM conditions wherever AWS supports resource-level permissions. The connector currently supports a long-lived access-key pair. STS session credentials, role assumption, workload identity, VPC/custom endpoints, enhanced fan-out subscriptions, and KCL lease coordination are deferred.

Configuration

The awsAccount relationship is the only credential configuration. Operation, region, stream, iterator, and payload fields belong in mapped input or module parameters.

{
  "version": "1.0.0",
  "authConfig": {
    "authStrategy": 1,
    "integrationAccount": "integration-account:aws-kinesis-production"
  },
  "payloadConfig": {
    "parameters": "{\"operation\":\"put_record\",\"region\":\"us-west-2\"}"
  }
}

The relationship value is illustrative. Persisted workflows bind the generated IntegrationAccount relationship, never plaintext credentials.

Operations

Operation	Provider behavior	Side effect and retry behavior
list_streams	Lists bounded stream names.	Read-only; each page retries transient failures up to the configured bound.
describe_stream	Reads status, mode, shard count, retention, encryption, and creation metadata.	Read-only; bounded retry.
list_shards	Lists shard identifiers and hash/sequence ranges.	Read-only; opaque pagination and bounded retry.
put_record	Writes one record and returns its shard and sequence number.	Billable immutable write; one provider attempt. Caller event IDs support downstream deduplication, not provider idempotency.
put_records	Writes one ordered batch of up to 500 records.	One attempt. AWS can partially accept a batch, so every item receipt must be reconciled before replay.
get_shard_iterator	Creates a short-lived iterator at an explicit position.	Read-only and retryable; returned iterator expires after about five minutes.
get_records	Reads one bounded page from an iterator and returns the next iterator and lag.	Read-only; transient retry. Do not call in a tight loop.
create_stream	Starts asynchronous on-demand or provisioned stream creation.	Billable lifecycle mutation; confirmation and one provider attempt. Poll describe_stream for ACTIVE.
update_shard_count	Starts uniform scaling to a reviewed shard target.	Capacity and billing mutation; confirmation and one provider attempt.
delete_stream	Starts asynchronous deletion of the stream and its retained records.	Destructive; confirmation and one provider attempt. Registered-consumer deletion is opt-in.

Errors and Failure Modes

Code	Typical cause	Retryable	Resolution
VALIDATION_ERROR	Missing/malformed region, stream, shard, iterator, record, encoding, capacity, pagination value, confirmation, or unexpected field.	No	Correct the named input; no provider request was sent.
UNSUPPORTED_OPERATION	Unknown operation.	No	Select a documented operation.
INTEGRATION_ACCOUNT_ERROR	Account missing, unverified, not READY, or missing key material.	No	Repair and bind the AWS account.
AWS_KINESIS_HTTP_400	Invalid state, iterator, capacity, record, KMS access, throughput pressure, or provider limit.	Reads may retry selected transient provider codes	Inspect the exact operation, stream state, and request ID.
AWS_KINESIS_HTTP_403	IAM or KMS denies the action or resource.	No	Grant only the missing permission under the intended conditions.
AWS_KINESIS_HTTP_404	Stream, shard, or iterator target no longer exists.	No	Reconcile the immutable stream/shard identifier and retention window.
AWS_KINESIS_HTTP_429 / 500 / 503	Throttle or transient provider pressure.	Reads retry; mutations report ambiguity	Inspect stream state, item receipts, downstream event IDs, and request ID before replay.
NETWORK_ERROR	DNS, TLS, timeout, or connectivity failure.	Reads retry; mutations report ambiguity	Treat write acceptance as unknown until exact provider or downstream state is checked.
EXECUTION_ERROR	Unexpected bounded runtime failure.	No automatic replay	Preserve operation and request evidence for investigation.

Provider exception messages are never copied into workflow output. This prevents credentials, iterators, event bodies, and unbounded account or stream detail from leaking through errors.

Example

Publish a JSON order event:

{
  "operation": "put_record",
  "region": "us-west-2",
  "streamName": "approved-order-events",
  "partitionKey": "customer-1042",
  "dataEncoding": "json",
  "data": {
    "eventId": "evt-order-A-1042-approved-v1",
    "orderId": "A-1042",
    "eventType": "approved",
    "occurredAt": "2026-09-06T10:05:00Z"
  }
}

Expected result shape:

{
  "status": "success",
  "operation": "put_record",
  "attempts": 1,
  "resultType": "record",
  "data": {
    "accepted": true,
    "shardId": "shardId-000000000000",
    "sequenceNumber": "49664830819800123456789012345678901234567890123456",
    "encryptionType": "KMS"
  },
  "requestId": "aws-request-reference"
}

The receipt proves that AWS accepted the record. It does not prove any consumer processed it. Use the stable eventId in an idempotent downstream store when end-to-end processing evidence matters.

Notes

• Pagination: list_streams and list_shards return one page unless returnAll=true; the connector follows opaque tokens with unchanged filters and stops at limit, exhaustion, or 100 pages.
• Consumer cadence: shard iterators expire after roughly five minutes. Use the returned nextShardIterator, wait at least the provider-recommended interval between reads, and stop when the shard is closed. This module is for bounded workflow reads, not a replacement for a continuously running Kinesis Client Library worker.
• Rate limits: Kinesis quotas vary by account, region, capacity mode, shard, and action. Read calls use at most five configured attempts with bounded backoff. Record writes and lifecycle mutations never retry automatically.
• API limits: the connector caps records at 1 MiB, batch decoded bytes plus partition keys at 5 MiB, batch items at 500, record reads at 1,000 per call, list results at 10,000, and list pages at 100. AWS can impose tighter or newer account-specific limits.
• Ordering: records with the same partition key map to the same shard. sequenceNumberForOrdering can enforce producer ordering for one record, but put_records can partially fail and does not guarantee ordering across retries.
• Idempotency: Kinesis PutRecord and PutRecords have no caller idempotency token. Embed a stable event ID and deduplicate downstream. Never blindly replay a timed-out or partially accepted write.
• Batch partial failures: put_records is a single provider request. Failed and successful results share the input order. Retry only explicitly failed application events after verifying that the failure receipt is definitive.
• Destructive behavior: delete_stream permanently removes retained records and requires confirmDelete=true. enforceConsumerDeletion=true expands the destructive scope to registered consumers. Scaling can increase cost and is guarded separately.
• Capacity: new streams are on-demand by default. Provisioned mode requires a shard count. Creation, resharding, and deletion are asynchronous; poll describe_stream for terminal state rather than assuming the acceptance response completed the change.
• Security: record bytes and shard iterators are classified confidential. Use explicit ACLs and downstream schema validation; never expose an iterator in logs, email, public content, or URLs.
• Deferred operations: stream retention changes, encryption controls, tagging, enhanced monitoring, enhanced fan-out consumer registration/subscription, resource policies, warm throughput, large-record opt-in, KPL aggregation, KCL leases, STS/role assumption, and custom endpoints are not implemented.
• Provider verification boundary: deterministic tests cover validation, encoding, batch bounds, pagination, operation families, retry boundaries, confirmations, iterator and credential redaction, error normalization, registration, and metadata. Live AWS behavior remains unverified until an authorized AWS account and reviewed stream are supplied.
• Functional reference: n8n's AWS credential model informs reusable credentials, region handling, and workflow ergonomics. AWS Kinesis API reference defines provider operations and limits. ValkyrAI adds the Kinesis-specific lifecycle and governance controls described above.
• Runtime boundary: merged source and published documentation do not update the deployed Workflow Studio catalog until a ValkyrAI backend release exposes AwsKinesisModule through /v1/modules/metadata.

Related Documentation

• ExecModule catalog
• AWS EventBridge ExecModule
• AWS Lambda ExecModule
• AWS Step Functions ExecModule
• Amazon Kinesis Data Streams concepts
• Integration Account Console