Cogitator
Deployment

Redis Client

Unified Redis client with standalone and cluster support — the foundation for memory, worker queues, and pub/sub.

Overview

The @cogitator-ai/redis package provides a unified Redis client that works identically in standalone and cluster modes. It wraps ioredis (a peer dependency, loaded on first use) with a consistent interface used by the @cogitator-ai/memory Redis adapter and accepted by the workflow Redis stores and the tool cache.

pnpm add @cogitator-ai/redis ioredis

Quick Start

Standalone

import { createRedisClient } from '@cogitator-ai/redis';

const redis = await createRedisClient({
  url: 'redis://localhost:6379',
  keyPrefix: 'myapp:',
});

await redis.set('key', 'value');
const value = await redis.get('key');
await redis.quit();

Cluster

import { createRedisClient } from '@cogitator-ai/redis';

const redis = await createRedisClient({
  mode: 'cluster',
  nodes: [
    { host: 'redis-1', port: 6379 },
    { host: 'redis-2', port: 6379 },
    { host: 'redis-3', port: 6379 },
  ],
  keyPrefix: '{myapp}:',
});

Options

OptionModeDescription
urlstandaloneConnection URL; otherwise built from host / port
host, portstandaloneDefault localhost:6379
dbstandaloneDatabase number
nodesclusterSeed nodes (at least one, otherwise createRedisClient throws)
scaleReadscluster'master' (default), 'slave' or 'all'
natMapclusterMap internal node addresses to reachable ones
keyPrefixbothPrepended to every key; use a {hashtag} in cluster mode
passwordbothAuthentication password
tlsbothConnect over TLS
maxRetriesPerRequestbothDefault 3
lazyConnectbothDelay connecting until the first command (default false)

A config is treated as cluster when it has mode: 'cluster' or a nodes array (isClusterConfig(config)). Reconnects back off up to 2 seconds.

Environment Configuration

Configure Redis entirely from environment variables:

import { createRedisClient, createConfigFromEnv } from '@cogitator-ai/redis';

const config = createConfigFromEnv(); // or createConfigFromEnv(customEnv)
const redis = await createRedisClient(config);

parseClusterNodesEnv(value) parses a REDIS_CLUSTER_NODES-style string on its own and returns null for invalid JSON.

VariableDescription
REDIS_URLRedis connection URL (takes precedence over host/port)
REDIS_HOSTRedis host (default: localhost)
REDIS_PORTRedis port (default: 6379)
REDIS_PASSWORDAuthentication password
REDIS_CLUSTER_NODESJSON array of {host, port} objects; switches to cluster mode (a malformed value throws instead of falling back to localhost)
REDIS_KEY_PREFIXKey prefix (default: cogitator: / {cogitator}:)

Auto-Detection

Detect whether a Redis server runs in standalone or cluster mode:

import { detectRedisMode, createRedisClient } from '@cogitator-ai/redis';

const mode = await detectRedisMode({ host: 'localhost', port: 6379 }); // reads INFO cluster

const redis = await createRedisClient(
  mode === 'cluster'
    ? { mode: 'cluster', nodes: [{ host: 'localhost', port: 6379 }] }
    : { host: 'localhost', port: 6379 }
);

Client Interface

The RedisClient interface provides key-value, sorted set, set, pub/sub, event and utility operations that work identically in both modes.

// Key-value
await redis.set('key', 'value');
await redis.get('key');
await redis.setex('key', 3600, 'value');
await redis.del('key1', 'key2');
await redis.expire('key', 60);
await redis.mget('key1', 'key2');
await redis.exists('key1', 'key2'); // number of keys that exist
await redis.incr('counter');
await redis.decr('counter');

// Sorted sets
await redis.zadd('scores', 100, 'player1');
await redis.zrange('scores', 0, -1); // stop may also be a string, e.g. '-1'
await redis.zrangebyscore('scores', 0, 100);
await redis.zrem('scores', 'player1');

// Sets
await redis.smembers('tags');

// Pub/sub
const sub = redis.duplicate();
await sub.subscribe('events', (channel, message) => {
  console.log(channel, message);
});
await redis.publish('events', JSON.stringify({ type: 'update' }));

// Connection events
redis.on('error', (error) => console.error(error));
redis.on('reconnecting', () => console.warn('reconnecting'));

// Utility
await redis.ping();
await redis.info('memory');
const keys = await redis.keys('cache:*');
await redis.del(...keys);

keys() is relative to keyPrefix: the prefix is added to the pattern and stripped from the results, so they can be passed straight to get/del. It uses SCAN (non-blocking) and visits every master node in cluster mode.

For a step-by-step walk, scan() has ioredis' SCAN cursor MATCH pattern COUNT count signature: start with cursor '0' and call again with the returned cursor until it is '0' again. Like keys(), the pattern and the returned keys are relative to keyPrefix. In cluster mode the cursor (<master index>:<node cursor>) walks every master node in turn.

let cursor = '0';
do {
  const [next, batch] = await redis.scan(cursor, 'MATCH', 'cache:*', 'COUNT', 100);
  if (batch.length > 0) await redis.del(...batch);
  cursor = next;
} while (cursor !== '0');

Pub/sub channels are not prefixed. Subscribe on a duplicate() connection: a subscribed connection can only run pub/sub commands. Several callbacks can share a channel; unsubscribe(channel) removes all of them.

Used By

  • Memory — RedisAdapter in @cogitator-ai/memory creates its own client from { url | host/port | cluster: { nodes, scaleReads }, password, keyPrefix }.
  • Workflows — RedisRunStore, RedisApprovalStore, RedisTimerStore and RedisCheckpointStore take any client with the commands they need, so a RedisClient from this package fits:
import { createRedisClient, createConfigFromEnv } from '@cogitator-ai/redis';
import { RedisApprovalStore, RedisRunStore } from '@cogitator-ai/workflows';

const redis = await createRedisClient(createConfigFromEnv());

const runStore = new RedisRunStore({ client: redis });
const approvalStore = new RedisApprovalStore({ client: redis, keyPrefix: '{approvals}' });

See Workflow Execution for what the stores do.

Tool-Cache Client

A RedisClient also satisfies RedisClientLike from @cogitator-ai/core, so it can back tool caching without ioredis-specific code. In cluster mode, use a {hashtag} key prefix: the cache storage sends multi-key commands (mget, del) that must stay in one slot.

import { createRedisClient } from '@cogitator-ai/redis';
import { withCache, webSearch } from '@cogitator-ai/core';

const redis = await createRedisClient({
  mode: 'cluster',
  nodes: [{ host: 'redis-1', port: 6379 }],
  keyPrefix: '{tools}:',
});

const cachedSearch = withCache(webSearch, {
  strategy: 'exact',
  ttl: '30m',
  maxSize: 1000,
  storage: 'redis',
  redisClient: redis,
});

Cluster Key Routing

Use hash tags to route related keys to the same slot:

const redis = await createRedisClient({
  mode: 'cluster',
  nodes: [{ host: 'redis-1', port: 6379 }],
  keyPrefix: '{myapp}:',
});

// Both keys route to the same slot via {myapp}
await redis.set('users:123', '...');
await redis.set('sessions:456', '...');

TLS and NAT

// TLS
const secure = await createRedisClient({
  url: 'rediss://secure.redis.host:6379',
  tls: true,
});

// NAT mapping for clusters behind load balancers
const natCluster = await createRedisClient({
  mode: 'cluster',
  nodes: [{ host: 'external.host', port: 6379 }],
  natMap: {
    '10.0.0.1:6379': { host: 'external-1.host', port: 6379 },
    '10.0.0.2:6379': { host: 'external-2.host', port: 6379 },
  },
});

Edge Runtimes

Run Cogitator agents on Cloudflare Workers, Deno and Bun with the Hono adapter — what works, what to configure, and what needs a Node.js server.

Worker Queues

Distributed job processing with BullMQ — background execution for agents, workflows, and swarms.

On this page

OverviewQuick StartStandaloneClusterOptionsEnvironment ConfigurationAuto-DetectionClient InterfaceUsed ByTool-Cache ClientCluster Key RoutingTLS and NAT