From 271df2baced094090e81f115ffc376d511a2e439 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Thu, 15 Oct 2026 11:30:20 +0200 Subject: [PATCH] feat(redis): accept an ioredis client or URL as the redis option The redis option only took RedisOptions, so the plugin always created and owned its connection. Callers who want to manage the connection themselves (e.g. attach their own 'error' handling, or share one client across the app) had no way to pass it in. The option now also accepts a Redis URL string and an ioredis Redis client. A provided client is used as-is: the plugin never quits it and adds no 'error' listener to it. The pub/sub connections the message broker derives from it with duplicate() are still owned and closed by the plugin. A client is detected with instanceof, falling back to a method-shape check so clients from another copy of ioredis are recognised too. Fixes #59 Co-authored-by: David Wood Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UF4S3w7iA83V9DVsasLJzM --- README.md | 27 ++++++ src/index.ts | 27 ++++-- src/redis-client.ts | 20 +++++ src/types.ts | 15 +++- test-d/index.test-d.ts | 7 +- test/redis-client-option.test.ts | 140 +++++++++++++++++++++++++++++++ 6 files changed, 223 insertions(+), 13 deletions(-) create mode 100644 src/redis-client.ts create mode 100644 test/redis-client-option.test.ts diff --git a/README.md b/README.md index bfd15b35..52832ae0 100644 --- a/README.md +++ b/README.md @@ -887,6 +887,33 @@ await app.register(mcpPlugin, { }) ``` +The `redis` option accepts three forms: + +| Form | Example | Who owns the connection | +| --- | --- | --- | +| `RedisOptions` object | `{ host: 'localhost', port: 6379 }` | The plugin: it creates the client, logs its errors, and quits it on `app.close()` | +| Redis URL string | `'redis://:secret@redis.example.com:6379/0'` | The plugin, as above | +| ioredis `Redis` client | `new Redis(...)` | You: the plugin uses it but never quits it and adds no `error` listener to it | + +Pass your own client when you want to control connecting, reconnecting and error handling yourself, or share one connection with the rest of your application: + +```typescript +import { Redis } from 'ioredis' + +const redis = new Redis(process.env.REDIS_URL) +redis.on('error', (err) => app.log.error({ err }, 'Redis error')) + +await app.register(mcpPlugin, { enableSSE: true, redis }) + +// ...later, on shutdown: close the plugin first, then your client +await app.close() +await redis.quit() +``` + +If you would rather close it from an `onClose` hook, add that hook before registering the plugin: Fastify runs `onClose` hooks in reverse order, so it then runs after the plugin's own cleanup. + +The message broker needs dedicated pub/sub connections, which the plugin derives from your client with `redis.duplicate()`. Those duplicates belong to the plugin and are closed on `app.close()`; your client stays open. + ### Message Broker Exports `MemoryMessageBroker`, `RedisMessageBroker`, and the `MessageBroker` interface are exported from the package's public entry point, so you can use them directly without deep-importing into `dist/brokers/...`: diff --git a/src/index.ts b/src/index.ts index a57f9b31..3596ba29 100644 --- a/src/index.ts +++ b/src/index.ts @@ -21,6 +21,7 @@ import { createAuthPreHandler } from './auth/prehandler.ts' import oauthClientPlugin from './auth/oauth-client.ts' import authRoutesPlugin from './routes/auth-routes.ts' import { quitWithTimeout } from './redis-quit-with-timeout.ts' +import { isRedisClient } from './redis-client.ts' import { createJsonSchemaValidator } from './validation/json-schema-validator.ts' import { createMcpClient, @@ -110,17 +111,27 @@ const mcpPlugin = fp(async function (app: FastifyInstance, opts: MCPPluginOption // Initialize stores and brokers based on configuration let sessionStore: SessionStore let messageBroker: MessageBroker - let redis: Redis | null = null + // Only a connection the plugin opened itself is closed on shutdown; a + // client passed in by the caller is theirs to manage. + let ownedRedis: Redis | null = null let taskStore: TaskStore | undefined if (opts.redis) { // Redis implementations for horizontal scaling - redis = new Redis(opts.redis) - // Without an 'error' listener a dropped connection is reported as unhandled; log it and let ioredis reconnect. - redis.on('error', (err) => { - app.log.error({ err }, 'Redis connection error') - }) + let redis: Redis + if (isRedisClient(opts.redis)) { + // Error handling on a caller-owned client is left to the caller + redis = opts.redis + } else { + // ioredis accepts either a URL or an options object + redis = typeof opts.redis === 'string' ? new Redis(opts.redis) : new Redis(opts.redis) + ownedRedis = redis + // Without an 'error' listener a dropped connection is reported as unhandled; log it and let ioredis reconnect. + redis.on('error', (err) => { + app.log.error({ err }, 'Redis connection error') + }) + } sessionStore = new RedisSessionStore({ redis, maxMessages: 100, sessionTtlMs: opts.sessionTtlMs }) messageBroker = new RedisMessageBroker(redis, { onCloseTimeout: (closeTimeoutMs) => { @@ -317,8 +328,8 @@ const mcpPlugin = fp(async function (app: FastifyInstance, opts: MCPPluginOption try { await messageBroker.close() } finally { - if (redis) { - await quitWithTimeout(redis, REDIS_QUIT_TIMEOUT_MS, () => { + if (ownedRedis) { + await quitWithTimeout(ownedRedis, REDIS_QUIT_TIMEOUT_MS, () => { app.log.warn({ timeoutMs: REDIS_QUIT_TIMEOUT_MS }, 'Redis client quit timed out; forcing disconnect') }) } diff --git a/src/redis-client.ts b/src/redis-client.ts new file mode 100644 index 00000000..d7367a66 --- /dev/null +++ b/src/redis-client.ts @@ -0,0 +1,20 @@ +import { Redis } from 'ioredis' + +const CLIENT_METHODS = ['duplicate', 'quit', 'disconnect', 'on', 'get', 'hset'] as const + +/** + * Tell a caller-provided ioredis client apart from RedisOptions or a URL. + * `instanceof` alone misses clients created by a different copy of ioredis + * (e.g. a duplicated dependency), so fall back to checking the methods the + * plugin relies on; a plain options object never has them. + */ +export function isRedisClient (value: unknown): value is Redis { + if (typeof value !== 'object' || value === null) { + return false + } + if (value instanceof Redis) { + return true + } + const candidate = value as Record + return CLIENT_METHODS.every((method) => typeof candidate[method] === 'function') +} diff --git a/src/types.ts b/src/types.ts index 35d21dbd..b485e13b 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1,6 +1,6 @@ import type { FastifyReply, FastifyRequest, FastifySchema, HTTPMethods } from 'fastify' import type { Options } from 'ajv' -import type { RedisOptions } from 'ioredis' +import type { Redis, RedisOptions } from 'ioredis' import type { JSONRPCMessage, JSONRPCNotification, @@ -401,10 +401,17 @@ export interface MCPPluginOptions { */ validateJsonSchemaInputs?: Options /** - * Redis connection options, passed as-is to the ioredis constructor. - * Providing this option switches sessions, messaging and tasks to the Redis backends. + * Redis connection for the session, messaging and task backends. Providing + * this option switches them from memory to Redis. Accepts: + * - an ioredis `Redis` client: used as-is and never closed by the plugin, + * so the caller keeps ownership (including its 'error' handling). The + * pub/sub connections are derived from it with `duplicate()` and closed + * by the plugin on shutdown. + * - a Redis URL (e.g. `redis://localhost:6379/0`), or + * - ioredis `RedisOptions`, passed as-is to the ioredis constructor. + * In the last two cases the plugin owns the connection and quits it on close. */ - redis?: RedisOptions + redis?: Redis | RedisOptions | string authorization?: AuthorizationConfig /** * Optional OpenTelemetry instrumentation. diff --git a/test-d/index.test-d.ts b/test-d/index.test-d.ts index 10e2df07..894605aa 100644 --- a/test-d/index.test-d.ts +++ b/test-d/index.test-d.ts @@ -6,7 +6,7 @@ import { } from 'tsd' import { Type } from '@sinclair/typebox' import type { FastifyInstance, FastifyReply, FastifyRequest, FastifySchema, HTTPMethods } from 'fastify' -import type { RedisOptions } from 'ioredis' +import type { Redis, RedisOptions } from 'ioredis' import { RedisMessageBroker, MemoryMessageBroker, RedisSessionStore, MemorySessionStore, RateLimiter } from '../dist/index.js' import type { ToolHandler, @@ -306,6 +306,11 @@ expectAssignable({ redis: ioredisOptions }) expectNotAssignable({ redis: { port: '6379' } }) expectNotAssignable({ redis: { maxRetriesPerRequest: 'three' } }) +// Redis config also accepts a caller-owned ioredis client or a Redis URL +expectAssignable({ redis: {} as Redis }) +expectAssignable({ redis: 'redis://localhost:6379/0' }) +expectNotAssignable({ redis: 6379 }) + // RateLimiter is exported from the main entry point const rateLimiter = new RateLimiter(100, 60000) expectType(rateLimiter.isAllowed('session-1')) diff --git a/test/redis-client-option.test.ts b/test/redis-client-option.test.ts new file mode 100644 index 00000000..8748ebc3 --- /dev/null +++ b/test/redis-client-option.test.ts @@ -0,0 +1,140 @@ +import { describe, test } from 'node:test' +import assert from 'node:assert' +import fastify from 'fastify' +import { Redis } from 'ioredis' +import mcpPlugin from '../src/index.ts' +import { isRedisClient } from '../src/redis-client.ts' +import { testWithRedis } from './redis-test-utils.ts' + +async function initializeSession (app: ReturnType): Promise { + const response = await app.inject({ + method: 'POST', + url: '/mcp', + payload: { + jsonrpc: '2.0', + method: 'initialize', + params: { + protocolVersion: '2025-06-18', + capabilities: {}, + clientInfo: { name: 'test-client', version: '1.0.0' } + }, + id: 1 + } + }) + if (response.statusCode !== 200) { + throw new Error(`initialize failed with ${response.statusCode}: ${response.payload}`) + } + const sessionId = response.headers['mcp-session-id'] + if (typeof sessionId !== 'string') { + throw new Error('initialize did not return a session id') + } + return sessionId +} + +describe('redis option: externally managed client', () => { + testWithRedis('uses a provided ioredis client and does not quit it on close', async (redis, t) => { + const client = new Redis({ + host: redis.options.host!, + port: redis.options.port!, + db: redis.options.db! + }) + t.after(() => client.disconnect()) + + const duplicates: Redis[] = [] + const duplicate = client.duplicate.bind(client) + client.duplicate = (...args: Parameters) => { + const conn = duplicate(...args) + duplicates.push(conn) + return conn + } + const errorListeners = client.listenerCount('error') + + const app = fastify() + await app.register(mcpPlugin, { enableSSE: true, redis: client }) + await app.ready() + + const sessionId = await initializeSession(app) + assert.strictEqual(await redis.exists(`session:${sessionId}`), 1) + + await app.close() + + // The caller's client stays open and usable + assert.strictEqual(client.status, 'ready') + assert.strictEqual(await client.ping(), 'PONG') + // The plugin did not leave its own listeners behind on the caller's client + assert.strictEqual(client.listenerCount('error'), errorListeners) + // Connections the plugin derived from the client (pub/sub) are closed + assert.ok(duplicates.length > 0) + for (const conn of duplicates) { + assert.strictEqual(conn.status, 'end') + } + }) + + testWithRedis('uses a provided client for the task store', async (redis, t) => { + const client = new Redis({ + host: redis.options.host!, + port: redis.options.port!, + db: redis.options.db! + }) + t.after(() => client.disconnect()) + + const app = fastify() + await app.register(mcpPlugin, { enableSSE: true, enableTasks: true, redis: client }) + await app.ready() + await app.close() + + assert.strictEqual(client.status, 'ready') + }) + + testWithRedis('accepts a Redis URL string', async (redis, t) => { + const url = `redis://${redis.options.host}:${redis.options.port}/${redis.options.db}` + const app = fastify() + t.after(() => app.close()) + + await app.register(mcpPlugin, { enableSSE: true, redis: url }) + await app.ready() + + const sessionId = await initializeSession(app) + // Stored in the database selected by the URL + assert.strictEqual(await redis.exists(`session:${sessionId}`), 1) + }) + + testWithRedis('still accepts RedisOptions', async (redis, t) => { + const app = fastify() + t.after(() => app.close()) + + await app.register(mcpPlugin, { + enableSSE: true, + redis: { + host: redis.options.host!, + port: redis.options.port!, + db: redis.options.db! + } + }) + await app.ready() + + const sessionId = await initializeSession(app) + assert.strictEqual(await redis.exists(`session:${sessionId}`), 1) + }) +}) + +describe('isRedisClient', () => { + test('recognises an ioredis instance', (t) => { + const client = new Redis({ lazyConnect: true }) + t.after(() => client.disconnect()) + assert.strictEqual(isRedisClient(client), true) + }) + + test('recognises a client from another copy of ioredis by shape', () => { + const fn = () => {} + const foreign = { duplicate: fn, quit: fn, disconnect: fn, on: fn, get: fn, hset: fn } + assert.strictEqual(isRedisClient(foreign), true) + }) + + test('rejects options objects, strings and nullish values', () => { + assert.strictEqual(isRedisClient({ host: 'localhost', port: 6379 }), false) + assert.strictEqual(isRedisClient('redis://localhost:6379'), false) + assert.strictEqual(isRedisClient(undefined), false) + assert.strictEqual(isRedisClient(null), false) + }) +})