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