feat(redis): accept an ioredis client or URL as the redis option

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/...`:

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')
         })
       }

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<string, unknown>
+  return CLIENT_METHODS.every((method) => typeof candidate[method] === 'function')
+}

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.

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<MCPPluginOptions>({ redis: ioredisOptions })
 expectNotAssignable<MCPPluginOptions>({ redis: { port: '6379' } })
 expectNotAssignable<MCPPluginOptions>({ redis: { maxRetriesPerRequest: 'three' } })
 
+// Redis config also accepts a caller-owned ioredis client or a Redis URL
+expectAssignable<MCPPluginOptions>({ redis: {} as Redis })
+expectAssignable<MCPPluginOptions>({ redis: 'redis://localhost:6379/0' })
+expectNotAssignable<MCPPluginOptions>({ redis: 6379 })
+
 // RateLimiter is exported from the main entry point
 const rateLimiter = new RateLimiter(100, 60000)
 expectType<boolean>(rateLimiter.isAllowed('session-1'))

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<typeof fastify>): Promise<string> {
+  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<Redis['duplicate']>) => {
+      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)
+  })
+})