Key-Value Storage Functions

Key-Value (KV) storage provides fast, simple data storage with key-based access. Perfect for configuration, caching, state management, and session storage.

What is Key-Value Storage? KV storage is a simple database where you store data with a unique key and retrieve it later using that key. Think of it like a dictionary or hash map that persists across flow executions.

Function List

kv-bucket-create

Create a new KV bucket

kv-key-put

Add or update an item

kv-key-get

Retrieve an item by key

kv-bucket-list

List all KV buckets

kv-key-list

List items in a bucket

kv-bucket-create

Create a new Key-Value storage bucket. Buckets are containers for related key-value pairs with configurable storage options, replication, and TTL settings.

Parameters

bucket

string

required

Name of the bucket to create. Must be unique and can only contain alphanumeric characters, dashes, and underscores.Naming conventions:

Use lowercase with hyphens: user-preferences, api-cache
Be descriptive: feature-flags not bucket1
Include purpose: session-data, product-config

description

string

Optional description for the KeyValue store.

ttl

integer

Time-to-live in nanoseconds. Keys expire after this duration. By default, keys do not expire.Common TTL values (in nanoseconds):

3600000000000 - 1 hour
86400000000000 - 1 day
604800000000000 - 7 days
2592000000000000 - 30 days
No value - No expiration (permanent storage)

history

integer

Number of historical values to keep per key. Default is 1, maximum is 64.

max_bytes

integer

Maximum size in bytes of the KeyValue store. Default is -1 (unlimited).

max_value_size

integer

Maximum size of a value in bytes. Default is -1 (unlimited).

storage

string

Type of storage backend to use. Default is file.Options:

file - Persistent file storage
memory - In-memory storage (faster but not persistent)

num_replicas

integer

Number of replicas to keep in clustered bstream. Default is 1, maximum is 5.

compression

boolean

Enable underlying stream compression to reduce storage size.

metadata

object

Optional bucket-specific metadata (custom key-value pairs).

placement

object

Configure where the stream should be placed in a cluster.Properties:

cluster (string) - Target cluster name
tags (array of strings) - Placement tags

mirror

object

Configuration for mirroring another KeyValue store.Properties:

name (string, required) - Name of the stream to mirror
domain (string) - Domain for cross-domain mirroring
filter_subject (string) - Subject filter for selective mirroring
opt_start_seq (integer) - Starting sequence number
opt_start_time (integer) - Starting timestamp
subject_transforms (array) - Subject transformation rules

republish

object

Configure immediate republishing of messages after storage.Properties:

dest (string) - Destination subject pattern
source (string) - Source subject pattern to match
headers_only (boolean) - Only republish headers

sources

array

Configure sources for the KeyValue store (for aggregating from multiple streams).Each source has:

name (string) - Name of source stream
domain (string) - Source domain
filter_subject (string) - Subject filter
opt_start_seq (integer) - Starting sequence
opt_start_time (integer) - Starting timestamp
subject_transforms (array) - Transformation rules

Response

{
  "status_code": 200,
  "body": {
    "message": "Bucket created successfully"
  }
}

Error Response

{
  "status_code": 400,
  "body": {
    "error": "Bucket already exists"
  }
}

Example Usage

{
  "function": "kv-bucket-create",
  "params": {
    "bucket": "user-preferences",
    "description": "Store user preferences and settings",
    "ttl": 2592000000000000
  }
}

{
  "function": "kv-bucket-create",
  "params": {
    "bucket": "api-cache",
    "description": "High-speed API response cache",
    "storage": "memory",
    "ttl": 3600000000000,
    "compression": true,
    "max_bytes": 1073741824
  }
}

{
  "function": "kv-bucket-create",
  "params": {
    "bucket": "critical-config",
    "description": "Replicated configuration storage",
    "num_replicas": 3,
    "storage": "file",
    "history": 5
  }
}

Common Use Cases

User Preferences

Store user settings and preferences with reasonable TTL

{
  "bucket": "user-preferences",
  "description": "User settings and preferences",
  "ttl": 2592000000000000,
  "storage": "file"
}

High-Speed Cache

In-memory cache for frequently accessed data

{
  "bucket": "api-cache",
  "storage": "memory",
  "ttl": 3600000000000,
  "compression": true
}

Feature Flags

Store feature toggles with history tracking

{
  "bucket": "feature-flags",
  "history": 10,
  "storage": "file"
}

Session Data

Temporary session storage with automatic expiration

{
  "bucket": "user-sessions",
  "ttl": 86400000000000,
  "storage": "memory"
}

kv-key-put

Add a new item or update an existing item in a Key-Value bucket.

Parameters

bucket

string

required

Name of the bucket to store the item in. Bucket must exist.

key

string

required

Unique key for the item. If key exists, the value will be updated.Key strategies:

User data: user_${userId} - e.g., user_123
Session data: session_${sessionId} - e.g., session_abc
Cache keys: cache_${resource}_${id} - e.g., cache_product_456
Config: Descriptive names - e.g., max_retries, api_timeout

value

string | object

required

The value to store. Can be a string or an object that will be JSON-serialized.Value types:

String: "Hello World"
Object: {"name": "John", "age": 30}
Array: [1, 2, 3, 4] (as object)
Number/Boolean: Must be wrapped in object or converted to string

Response

{
  "status_code": 200,
  "body": {
    "message": "Key written successfully"
  }
}

Error Response

{
  "status_code": 404,
  "body": {
    "error": "Bucket not found"
  }
}

Example Usage

{
  "function": "kv-key-put",
  "params": {
    "bucket": "user-preferences",
    "key": "user_${$.trigger.user_id}",
    "value": {
      "theme": "${$.trigger.theme}",
      "language": "${$.trigger.language}",
      "email_notifications": "${$.trigger.email_notifications}"
    }
  }
}

{
  "function": "kv-key-put",
  "params": {
    "bucket": "api-cache",
    "key": "weather_${$.trigger.city}",
    "value": "${$.http_request.response}"
  }
}

{
  "function": "kv-key-put",
  "params": {
    "bucket": "app-config",
    "key": "api_endpoint",
    "value": "https://api.example.com/v1"
  }
}

Common Patterns

Configuration Storage

Store application configuration

Flow:
→ Agent decides configuration value
→ Store in KV: kv-key-put
  Bucket: "app-config"
  Key: "max_upload_size"
  Value: {"bytes": 10485760, "mb": 10}
→ Other flows read this config

Rate Limiting

Track API usage per user

Flow:
→ Get current count: kv-key-get
→ Increment count
→ Store updated count: kv-key-put
  Key: "rate_limit_${user_id}"
  Value: {"count": 45, "reset_at": "..."}
→ Check if over limit

User State Management

Maintain user state across sessions

Flow:
→ User performs action
→ Update state: kv-key-put
  Key: "user_state_${user_id}"
  Value: {"onboarding_step": 3, "completed_tutorial": true}
→ Next visit: Retrieve state, continue where left off

Deduplication

Prevent duplicate processing

Flow:
→ Receive event with ID
→ Check if processed: kv-key-get
  Key: "processed_${event_id}"
→ If not found:
  → Process event
  → Mark as processed: kv-key-put
    Key: "processed_${event_id}"
    Value: {"timestamp": "...", "status": "processed"}

kv-key-get

Retrieve an item from a Key-Value bucket by its key.

Parameters

bucket

string

required

Name of the bucket to retrieve from.

key

string

required

The key of the item to retrieve.

json

boolean

required

Whether to parse the value as JSON. Set to true to parse objects, false to get raw string.When to use:

true - When you stored an object and want it parsed
false - When you stored a string or want the raw value

Response

{
  "status_code": 200,
  "body": {
    "body": {
      "key": "user_123",
      "value": {
        "theme": "dark",
        "language": "en"
      }
    },
    "metadata": {
      "Bucket": "user-preferences"
    }
  }
}

Not Found Response

{
  "status_code": 404,
  "body": {
    "error": "Key not found"
  }
}

Example Usage

{
  "function": "kv-key-get",
  "params": {
    "bucket": "user-preferences",
    "key": "user_${$.trigger.user_id}",
    "json": true
  }
}

{
  "function": "kv-key-get",
  "params": {
    "bucket": "api-cache",
    "key": "weather_${$.trigger.city}",
    "json": false
  }
}

Functions: Get user preferences
  Function: kv-key-get
  Bucket: "user-preferences"
  Key: user_${user_id}
  JSON: true
↓
Condition: Check status_code
  → 200: Use value from body.body.value
  → 404: Use defaults

Common Patterns

Cache-Aside Pattern

Check cache before making expensive call

Flow:
→ Get from cache: kv-key-get (json: true)
→ Check status_code
→ If 200: Use cached value
→ If 404:
  → Make API call
  → Store in cache: kv-key-put
  → Return value

Load User Context

Load user data for personalization

Trigger: User request
↓
Functions: Get user preferences
  Function: kv-key-get
  Bucket: "user-preferences"
  Key: user_${user_id}
  JSON: true
↓
Agent: Respond with personalized content
  Uses: body.body.value

Feature Flag Check

Check if feature is enabled

Functions: Check feature flag
  Function: kv-key-get
  Bucket: "feature-flags"
  Key: "new_checkout_flow"
  JSON: true
↓
Condition: status_code === 200 AND body.body.value.enabled
  → Yes: Use new flow
  → No: Use old flow

kv-bucket-list

List all Key-Value storage buckets in your account with their metadata.

Parameters

This function takes no input parameters. It returns all buckets in your account.

Response

{
  "status_code": 200,
  "body": {
    "body": {
      "results_total": 2,
      "results": [
        {
          "name": "user-preferences",
          "description": "Store user preferences and settings",
          "entry_total": 1523,
          "created": 1693574400000,
          "metadata": {}
        },
        {
          "name": "api-cache",
          "description": "Cache external API responses",
          "entry_total": 342,
          "created": 1696240200000,
          "metadata": {}
        }
      ]
    },
    "metadata": {}
  }
}

Example Usage

{
  "function": "kv-bucket-list",
  "params": {}
}

Functions: List all KV buckets
  Function: kv-bucket-list
↓
Map: Format bucket list
  Input: ${$.functions.body.body.results}
↓
Agent: Show storage overview to admin

Common Use Cases

Storage Monitoring

Monitor storage usage and health

List all buckets
→ Check entry_total counts
→ Alert if approaching limits
→ Identify unused buckets

Admin Dashboard

Build storage management interface

List buckets → Display in admin panel
→ Show: Name, description, entry count
→ Allow: View/manage buckets

Audit and Compliance

Track all storage buckets for compliance

List buckets
→ Document all data stores
→ Verify naming conventions
→ Export for audit trail

kv-key-list

List all items (keys) contained within a Key-Value bucket.

Parameters

bucket

string

required

Name of the bucket to list items from.

last_sequence

integer

Sequence number to start listing from (for pagination). Use the sequence from the last item of the previous page.

limit

integer

Maximum number of items to return per request.

Response

{
  "status_code": 200,
  "body": {
    "body": {
      "results_total": 1523,
      "results": [
        {
          "key": "user_123",
          "value": "{\"theme\":\"dark\",\"language\":\"en\"}"
        },
        {
          "key": "user_456",
          "value": "{\"theme\":\"light\",\"language\":\"es\"}"
        }
      ]
    },
    "metadata": {}
  }
}

Note: The value field in list results is always returned as a string. You’ll need to parse JSON values manually if needed.

Example Usage

{
  "function": "kv-key-list",
  "params": {
    "bucket": "user-preferences",
    "limit": 100
  }
}

{
  "function": "kv-key-list",
  "params": {
    "bucket": "user-sessions",
    "last_sequence": 100,
    "limit": 50
  }
}

Functions: List all keys
  Function: kv-key-list
  Bucket: "user-sessions"
  Limit: 100
↓
Map: Process each key
  Parse JSON values if needed
↓
Agent: Generate activity report

Pagination Pattern

First Request:
  kv-key-list(bucket: "data", limit: 100)
  → Returns items 0-99
  → Last item has sequence: 99

Next Request:
  kv-key-list(bucket: "data", last_sequence: 99, limit: 100)
  → Returns items 100-199
  → Continue until results_total reached

Common Patterns

Bulk Export

Export all items for backup or migration

Page 1: List items (limit: 1000)
→ Process each page
→ Use last_sequence for next page
→ Export to file or external system
→ Continue until all items processed

Data Cleanup

Find and remove old or unused items

List all items (paginated)
→ Parse each value
→ Check timestamps or usage
→ Delete old items

Analytics

Analyze stored data patterns

List all items
→ Parse JSON values
→ Aggregate by patterns
→ Generate insights
→ Optimize storage

Search and Filter

Find items matching criteria

List items (paginated)
→ Parse values
→ Filter by key pattern or value content
→ Return matching items

Best Practices

Use Descriptive Keys

Include entity type in key: user_123, session_abc, cache_product_456

Set Appropriate TTLs

Configure TTL at bucket creation. Remember: TTL is in nanoseconds!

Choose Right Storage Type

Use memory for high-speed cache, file for persistent data

Handle Status Codes

Always check status_code in responses (200 = success, 404 = not found)

Parse JSON Carefully

Use json: true in kv-key-get for objects. List results are always strings.

Plan for Replication

Use num_replicas for critical data to ensure high availability

Performance Tips

Choose Right Storage Backend

Match storage type to use case

✅ Memory storage: High-speed cache, temporary data
✅ File storage: Persistent data, long-term storage
❌ Memory storage: Critical data that must survive restarts

Enable Compression for Large Values

Reduce storage size and network transfer

{
  "bucket": "document-cache",
  "compression": true,
  "storage": "file"
}

Use History for Versioning

Track value changes over time

{
  "bucket": "config-history",
  "history": 10
}

Set Max Value Size

Prevent oversized values from consuming resources

{
  "bucket": "user-data",
  "max_value_size": 1048576
}

Example Workflows

User Preference Management

Create Bucket

{
  "function": "kv-bucket-create",
  "params": {
    "bucket": "user-preferences",
    "description": "User settings",
    "ttl": 2592000000000000,
    "storage": "file"
  }
}

Store Preferences

{
  "function": "kv-key-put",
  "params": {
    "bucket": "user-preferences",
    "key": "user_${user_id}",
    "value": {
      "theme": "dark",
      "language": "en",
      "notifications": true
    }
  }
}

Load on Login

{
  "function": "kv-key-get",
  "params": {
    "bucket": "user-preferences",
    "key": "user_${user_id}",
    "json": true
  }
}

API Response Caching

Create Cache Bucket

{
  "function": "kv-bucket-create",
  "params": {
    "bucket": "api-cache",
    "storage": "memory",
    "ttl": 3600000000000,
    "compression": true
  }
}

Check Cache

{
  "function": "kv-key-get",
  "params": {
    "bucket": "api-cache",
    "key": "weather_${city}",
    "json": true
  }
}

# Check status_code
# If 200: Use cached value
# If 404: Fetch from API

Cache Miss - Store Response

{
  "function": "kv-key-put",
  "params": {
    "bucket": "api-cache",
    "key": "weather_${city}",
    "value": "${api_response}"
  }
}

Feature Flag System

Create Flags Bucket

{
  "function": "kv-bucket-create",
  "params": {
    "bucket": "feature-flags",
    "history": 10,
    "storage": "file"
  }
}

Set Feature Flag

{
  "function": "kv-key-put",
  "params": {
    "bucket": "feature-flags",
    "key": "new_checkout",
    "value": {
      "enabled": true,
      "rollout_percentage": 100,
      "updated_at": "${now}"
    }
  }
}

Check in Flow

{
  "function": "kv-key-get",
  "params": {
    "bucket": "feature-flags",
    "key": "new_checkout",
    "json": true
  }
}

# If status_code == 200 and body.body.value.enabled:
#   Use new checkout
# Else:
#   Use old checkout

Important Notes

TTL is in Nanoseconds: When setting TTL, remember that the value must be in nanoseconds, not seconds. 1 hour = 3,600,000,000,000 nanoseconds.

JSON Parsing: The json parameter in kv-key-get is required. Set it to true to parse object values, false for strings.

List Values are Strings: When using kv-key-list, all values are returned as strings. You must parse JSON values manually.

Bucket Names are Permanent: Bucket names cannot be changed after creation. Choose descriptive, meaningful names from the start.

Next Steps

Stream Functions

Real-time event processing

Object Storage

Store large files and documents

Data Transformation

Transform and encode data

Functions Step

Using Functions in flows

​Key-Value Storage Functions

​Function List

kv-bucket-create

kv-key-put

kv-key-get

kv-bucket-list

kv-key-list

​kv-bucket-create

​Parameters

​Response

​Error Response

​Example Usage

​Common Use Cases

​kv-key-put

​Parameters

​Response

​Error Response

​Example Usage

​Common Patterns

​kv-key-get

​Parameters

​Response

​Not Found Response

​Example Usage

​Common Patterns

​kv-bucket-list

​Parameters

​Response

​Example Usage

​Common Use Cases

​kv-key-list

​Parameters

​Response

​Example Usage

​Pagination Pattern

​Common Patterns

​Best Practices

Use Descriptive Keys

Set Appropriate TTLs

Choose Right Storage Type

Handle Status Codes

Parse JSON Carefully

Plan for Replication

​Performance Tips

​Example Workflows

​User Preference Management

​API Response Caching

​Feature Flag System

​Important Notes

​Next Steps

Stream Functions

Object Storage

Data Transformation

Functions Step

Key-Value Storage Functions

Function List

kv-bucket-create

Parameters

Response

Error Response

Example Usage

Common Use Cases

kv-key-put

Parameters

Response

Error Response

Example Usage

Common Patterns

kv-key-get

Parameters

Response

Not Found Response

Example Usage

Common Patterns

kv-bucket-list

Parameters

Response

Example Usage

Common Use Cases

kv-key-list

Parameters

Response

Example Usage

Pagination Pattern

Common Patterns

Best Practices

Performance Tips

Example Workflows

User Preference Management

API Response Caching

Feature Flag System

Important Notes

Next Steps