Guide · Event-Driven & Async Patterns
MCP Server Redis Streams — XADD, XREADGROUP, XACK, consumer groups, XAUTOCLAIM
Three Redis Streams behaviours catch MCP server authors off-guard: XREAD without COUNT returns all messages since the ID — always pass COUNT to avoid reading the entire stream into memory — unlike Redis pub/sub where messages are ephemeral, a Redis Stream is a persistent log; calling XREAD COUNT 0 STREAMS mystream 0-0 (COUNT 0 means no limit) on a stream with millions of entries returns all entries in a single blocking response and will exhaust the calling process's memory; always pass an explicit COUNT of 100–500 and use a cursor loop; consumer groups track the Pending Entries List (PEL) per consumer, not per group — a dead consumer's unacknowledged messages sit in PEL forever until you run XAUTOCLAIM — when a consumer in a group crashes mid-processing, the messages it acknowledged but did not XACK remain in the PEL attributed to that consumer; other consumers in the group will not see those messages again unless you call XPENDING to inspect the PEL and XAUTOCLAIM (Redis 6.2+) or XCLAIM (older) to transfer stuck messages to an active consumer; without this, messages are silently lost to processing; and the ID $ in XGROUP CREATE means "start from now", and the special ID > in XREADGROUP means "give me undelivered messages" — these two operators are related but distinct — when creating a consumer group with XGROUP CREATE stream group $, the $ instructs Redis to set the group's last-delivered-id to the current last entry ID (skip all existing messages); subsequently, XREADGROUP GROUP group consumer COUNT N STREAMS stream > uses > to request only messages delivered after the group's last-delivered-id; the two IDs serve different phases: $ at group creation, > during normal consumption.
TL;DR
Use ioredis. Append entries with client.xadd(stream, '*', ...fields) — '*' auto-generates the ID. Create consumer groups with XGROUP CREATE stream group $ MKSTREAM — MKSTREAM creates the stream if it doesn't exist, $ starts from now. Read with XREADGROUP GROUP g consumer COUNT 100 BLOCK 2000 STREAMS stream >. Acknowledge with XACK stream group id immediately after processing. Run XAUTOCLAIM stream group consumer 60000 0-0 periodically to reclaim messages stuck in the PEL for more than 60 seconds. Trim streams with XADD stream MAXLEN ~ 100000 '*' ... to bound memory usage.
Appending events with XADD
Redis Streams store entries as flat key-value maps. All field values must be strings — serialize nested objects to JSON before writing. The auto-generated entry ID is a millisecond-precision timestamp plus a sequence counter (1234567890123-0), which makes IDs globally ordered by wall-clock time within the same Redis node. Use MAXLEN ~ N in XADD to trim the stream to an approximate length — the ~ operator allows Redis to trim in O(1) by deleting full radix tree nodes rather than individual entries.
import Redis from 'ioredis';
import { z } from 'zod';
const redis = new Redis({
host: process.env.REDIS_HOST ?? 'localhost',
port: parseInt(process.env.REDIS_PORT ?? '6379'),
password: process.env.REDIS_PASSWORD,
tls: process.env.REDIS_TLS === 'true' ? {} : undefined,
});
const STREAM_NAME = process.env.REDIS_STREAM ?? 'mcp-events';
const CONSUMER_GROUP = process.env.REDIS_GROUP ?? 'mcp-consumers';
const MAX_STREAM_LEN = 100_000; // approximate — use ~ for O(1) trim
// MCP tool — append an event to the stream
server.tool(
'stream_append',
{
event_type: z.string().min(1),
payload: z.record(z.unknown()),
stream: z.string().optional(),
},
async ({ event_type, payload, stream }) => {
const streamName = stream ?? STREAM_NAME;
// All field values must be strings — serialize objects
const id = await redis.xadd(
streamName,
'MAXLEN', '~', String(MAX_STREAM_LEN),
'*', // auto-generate entry ID
'type', event_type,
'payload', JSON.stringify(payload),
'timestamp', new Date().toISOString(),
);
return {
content: [{
type: 'text',
text: JSON.stringify({ appended: true, stream: streamName, entry_id: id }),
}],
};
}
);
// MCP tool — read the most recent N entries from a stream (no consumer group)
server.tool(
'stream_read_latest',
{
count: z.number().int().min(1).max(500).default(50),
stream: z.string().optional(),
},
async ({ count, stream }) => {
const streamName = stream ?? STREAM_NAME;
// XREVRANGE reads from newest to oldest — reverse for chronological display
const entries = await redis.xrevrange(streamName, '+', '-', 'COUNT', count);
const parsed = entries.reverse().map(([id, fields]) => {
const obj: Record<string, unknown> = { id };
for (let i = 0; i < fields.length; i += 2) {
const key = fields[i];
const val = fields[i + 1];
obj[key] = key === 'payload' ? JSON.parse(val) : val;
}
return obj;
});
return {
content: [{
type: 'text',
text: JSON.stringify({ stream: streamName, count: parsed.length, entries: parsed }, null, 2),
}],
};
}
);Consumer groups — XREADGROUP and XACK
Consumer groups enable competitive consumption: multiple consumers within a group each receive a unique subset of entries — each entry is delivered to exactly one consumer in the group. The group tracks which entries have been acknowledged (XACK). Unacknowledged entries sit in the group's Pending Entries List (PEL) and are re-delivered to the same consumer on subsequent XREADGROUP calls that use entry IDs (not >). The BLOCK timeout (milliseconds) instructs Redis to wait for new entries before returning — use this for long-polling consumers that should not busy-loop.
// Create consumer group if it doesn't exist
async function ensureConsumerGroup(streamName: string, groupName: string): Promise<void> {
try {
// MKSTREAM creates the stream if it doesn't exist
// $ means: only process entries added after group creation
await redis.xgroup('CREATE', streamName, groupName, '$', 'MKSTREAM');
} catch (err: any) {
if (!err.message?.includes('BUSYGROUP')) throw err;
// BUSYGROUP = group already exists — safe to continue
}
}
// MCP tool — consume events from a consumer group (one poll iteration)
server.tool(
'stream_consume',
{
consumer_name: z.string().min(1),
count: z.number().int().min(1).max(200).default(10),
block_ms: z.number().int().min(0).max(10_000).default(2000),
stream: z.string().optional(),
group: z.string().optional(),
},
async ({ consumer_name, count, block_ms, stream, group }) => {
const streamName = stream ?? STREAM_NAME;
const groupName = group ?? CONSUMER_GROUP;
await ensureConsumerGroup(streamName, groupName);
// XREADGROUP with '>' reads only new (undelivered) entries
// BLOCK waits up to block_ms milliseconds for new entries
const results = await redis.xreadgroup(
'GROUP', groupName, consumer_name,
'COUNT', count,
'BLOCK', block_ms,
'STREAMS', streamName, '>' // '>' = deliver undelivered messages only
) as Array<[string, Array<[string, string[]]>]> | null;
if (!results || results.length === 0) {
return { content: [{ type: 'text', text: JSON.stringify({ entries: [], count: 0 }) }] };
}
const [_streamName, entries] = results[0];
const parsed = entries.map(([id, fields]) => {
const obj: Record<string, unknown> = { id };
for (let i = 0; i < fields.length; i += 2) {
const key = fields[i];
const val = fields[i + 1];
obj[key] = key === 'payload' ? JSON.parse(val) : val;
}
return obj;
});
return {
content: [{
type: 'text',
text: JSON.stringify({
stream: streamName,
group: groupName,
consumer: consumer_name,
count: parsed.length,
entries: parsed,
instructions: 'Call stream_ack with the entry IDs after processing.',
}, null, 2),
}],
};
}
);
// MCP tool — acknowledge processed entries (removes from PEL)
server.tool(
'stream_ack',
{
entry_ids: z.array(z.string().min(1)).min(1).max(200),
stream: z.string().optional(),
group: z.string().optional(),
},
async ({ entry_ids, stream, group }) => {
const streamName = stream ?? STREAM_NAME;
const groupName = group ?? CONSUMER_GROUP;
const acked = await redis.xack(streamName, groupName, ...entry_ids);
return {
content: [{
type: 'text',
text: JSON.stringify({ acknowledged: acked, submitted: entry_ids.length }),
}],
};
}
);Reclaiming stuck messages — XPENDING and XAUTOCLAIM
When a consumer crashes or disconnects after receiving entries but before acknowledging them, those entries remain in the PEL indefinitely. Other consumers in the group will not see them because Redis considers them "owned" by the dead consumer. The recovery pattern is: periodically call XPENDING to find entries that have been sitting in the PEL longer than an acceptable idle timeout, then use XAUTOCLAIM (Redis 6.2+) to transfer those entries to an active consumer for reprocessing. Always design your event handlers to be idempotent — XAUTOCLAIM means an entry may be processed more than once.
// MCP tool — inspect pending entries in the PEL
server.tool(
'stream_pending',
{
stream: z.string().optional(),
group: z.string().optional(),
consumer: z.string().optional(),
count: z.number().int().min(1).max(500).default(50),
},
async ({ stream, group, consumer, count }) => {
const streamName = stream ?? STREAM_NAME;
const groupName = group ?? CONSUMER_GROUP;
// Summary form: XPENDING stream group - + count [consumer]
const pending = await redis.xpending(
streamName, groupName, '-', '+', count,
...(consumer ? [consumer] : [])
) as Array<[string, string, number, number]>;
const entries = (pending ?? []).map(([id, ownerConsumer, idleMs, deliveryCount]) => ({
id,
consumer: ownerConsumer,
idle_ms: idleMs,
idle_seconds: Math.floor(idleMs / 1000),
delivery_count: deliveryCount,
is_stuck: idleMs > 60_000,
}));
return {
content: [{
type: 'text',
text: JSON.stringify({
stream: streamName,
group: groupName,
pending_count: entries.length,
stuck_count: entries.filter(e => e.is_stuck).length,
entries,
}, null, 2),
}],
};
}
);
// MCP tool — reclaim stuck entries via XAUTOCLAIM
server.tool(
'stream_reclaim_stuck',
{
claim_to_consumer: z.string().min(1),
min_idle_ms: z.number().int().min(1000).default(60_000),
count: z.number().int().min(1).max(100).default(50),
stream: z.string().optional(),
group: z.string().optional(),
},
async ({ claim_to_consumer, min_idle_ms, count, stream, group }) => {
const streamName = stream ?? STREAM_NAME;
const groupName = group ?? CONSUMER_GROUP;
// XAUTOCLAIM: transfer entries idle for min_idle_ms to claim_to_consumer
// '0-0' = start from the beginning of the PEL
const result = await redis.xautoclaim(
streamName, groupName, claim_to_consumer,
min_idle_ms, '0-0', 'COUNT', count
) as [string, Array<[string, string[]]>, string[]];
const [nextId, claimed, deleted] = result;
return {
content: [{
type: 'text',
text: JSON.stringify({
claimed_count: claimed.length,
deleted_count: deleted.length, // entries deleted from PEL (already acked elsewhere)
next_cursor_id: nextId, // '0-0' means scan is complete
scan_complete: nextId === '0-0',
claimed_ids: claimed.map(([id]) => id),
}, null, 2),
}],
};
}
);
// Health probe — check stream exists and report lag
server.resource('redis_streams_health', 'event://redis/streams/health', async () => {
try {
const info = await redis.xinfo('STREAM', STREAM_NAME) as unknown[];
// xinfo STREAM returns flat array: ['length', N, 'radix-tree-keys', M, ...]
const infoMap: Record<string, unknown> = {};
for (let i = 0; i < info.length - 1; i += 2) {
infoMap[String(info[i])] = info[i + 1];
}
// Consumer group lag
const groups = await redis.xinfo('GROUPS', STREAM_NAME) as unknown[][];
const groupInfo = groups.map(g => {
const gMap: Record<string, unknown> = {};
const arr = g as unknown[];
for (let i = 0; i < arr.length - 1; i += 2) {
gMap[String(arr[i])] = arr[i + 1];
}
return { name: gMap['name'], pending: gMap['pending-count'], lag: gMap['lag'] };
});
return {
contents: [{
uri: 'event://redis/streams/health',
text: JSON.stringify({
status: 'ok',
stream: STREAM_NAME,
length: infoMap['length'],
groups: groupInfo,
}),
}],
};
} catch (err: any) {
return {
contents: [{
uri: 'event://redis/streams/health',
text: JSON.stringify({ status: 'error', stream: STREAM_NAME, message: err.message }),
}],
};
}
});