feat(resources): Add resources/subscribe and resources/unsubscribe support

src/decorators/meta.ts

@@ -3,18 +3,22 @@ import fp from 'fastify-plugin'
 import type {
   MCPTool,
   MCPResource,
-  MCPPrompt
+  MCPPrompt,
+  ResourceHandlers,
+  ResourceSubscribeHandler,
+  ResourceUnsubscribeHandler
 } from '../types.ts'
 import { schemaToArguments, validateToolSchema } from '../validation/index.ts'
 
 interface MCPDecoratorsOptions {
   tools: Map<string, MCPTool>
   resources: Map<string, MCPResource>
   prompts: Map<string, MCPPrompt>
+  resourceHandlers: ResourceHandlers
 }
 
 const mcpDecoratorsPlugin: FastifyPluginAsync<MCPDecoratorsOptions> = async (app, options) => {
-  const { tools, resources, prompts } = options
+  const { tools, resources, prompts, resourceHandlers } = options
 
   // Enhanced tool decorator with TypeBox schema support
   app.decorate('mcpAddTool', (
@@ -93,6 +97,15 @@ const mcpDecoratorsPlugin: FastifyPluginAsync<MCPDecoratorsOptions> = async (app
       handler
     })
   })
+
+  // Resource subscription handler setters
+  app.decorate('mcpSetResourceSubscribeHandler', (handler: ResourceSubscribeHandler) => {
+    resourceHandlers.subscribeHandler = handler
+  })
+
+  app.decorate('mcpSetResourceUnsubscribeHandler', (handler: ResourceUnsubscribeHandler) => {
+    resourceHandlers.unsubscribeHandler = handler
+  })
 }
 
 export default fp(mcpDecoratorsPlugin, {

src/handlers.ts

@@ -23,7 +23,7 @@ import {
   INVALID_PARAMS
 } from './schema.ts'
 
-import type { MCPTool, MCPResource, MCPPrompt, MCPPluginOptions } from './types.ts'
+import type { MCPTool, MCPResource, MCPPrompt, MCPPluginOptions, ResourceHandlers } from './types.ts'
 import type { AuthorizationContext } from './types/auth-types.ts'
 import { validate, CallToolRequestSchema, ReadResourceRequestSchema, GetPromptRequestSchema, isTypeBoxSchema } from './validation/index.ts'
 import { sanitizeToolParams, assessToolSecurity, SECURITY_WARNINGS } from './security.ts'
@@ -36,6 +36,7 @@ type HandlerDependencies = {
   tools: Map<string, MCPTool>
   resources: Map<string, MCPResource>
   prompts: Map<string, MCPPrompt>
+  resourceHandlers: ResourceHandlers
   request: FastifyRequest
   reply: FastifyReply
   authContext?: AuthorizationContext
@@ -259,7 +260,19 @@ async function handleResourcesRead (
   const params = paramsValidation.data
   const uri = params.uri
 
-  const resource = resources.get(uri)
+  // Try exact match first
+  let resource = resources.get(uri)
+
+  // If not found and URI has query params, try base URI for resources with uriSchema
+  if (!resource && uri.includes('?')) {
+    const baseUri = uri.split('?')[0]
+    const baseResource = resources.get(baseUri)
+    // Only use base resource if it has a uriSchema (expects query params)
+    if (baseResource?.definition?.uriSchema) {
+      resource = baseResource
+    }
+  }
+
   if (!resource) {
     return createError(request.id, METHOD_NOT_FOUND, `Resource '${uri}' not found`)
   }
@@ -440,6 +453,64 @@ async function handlePromptsGet (
   }
 }
 
+async function handleResourcesSubscribe (
+  request: JSONRPCRequest,
+  sessionId: string | undefined,
+  dependencies: HandlerDependencies
+): Promise<JSONRPCResponse | JSONRPCError> {
+  const { resourceHandlers } = dependencies
+
+  if (!resourceHandlers.subscribeHandler) {
+    return createError(request.id, METHOD_NOT_FOUND, 'resources/subscribe handler not configured')
+  }
+
+  const params = request.params as { uri: string }
+  if (!params?.uri) {
+    return createError(request.id, INVALID_PARAMS, 'Missing required parameter: uri')
+  }
+
+  try {
+    const result = await resourceHandlers.subscribeHandler(params, {
+      sessionId,
+      request: dependencies.request,
+      reply: dependencies.reply,
+      authContext: dependencies.authContext
+    })
+    return createResponse(request.id, result)
+  } catch (error: any) {
+    return createError(request.id, INTERNAL_ERROR, `Subscribe failed: ${error.message || error}`)
+  }
+}
+
+async function handleResourcesUnsubscribe (
+  request: JSONRPCRequest,
+  sessionId: string | undefined,
+  dependencies: HandlerDependencies
+): Promise<JSONRPCResponse | JSONRPCError> {
+  const { resourceHandlers } = dependencies
+
+  if (!resourceHandlers.unsubscribeHandler) {
+    return createError(request.id, METHOD_NOT_FOUND, 'resources/unsubscribe handler not configured')
+  }
+
+  const params = request.params as { uri: string }
+  if (!params?.uri) {
+    return createError(request.id, INVALID_PARAMS, 'Missing required parameter: uri')
+  }
+
+  try {
+    const result = await resourceHandlers.unsubscribeHandler(params, {
+      sessionId,
+      request: dependencies.request,
+      reply: dependencies.reply,
+      authContext: dependencies.authContext
+    })
+    return createResponse(request.id, result)
+  } catch (error: any) {
+    return createError(request.id, INTERNAL_ERROR, `Unsubscribe failed: ${error.message || error}`)
+  }
+}
+
 export async function handleRequest (
   request: JSONRPCRequest,
   sessionId: string | undefined,
@@ -469,6 +540,10 @@ export async function handleRequest (
         return await handleToolsCall(request, sessionId, dependencies)
       case 'resources/read':
         return await handleResourcesRead(request, sessionId, dependencies)
+      case 'resources/subscribe':
+        return await handleResourcesSubscribe(request, sessionId, dependencies)
+      case 'resources/unsubscribe':
+        return await handleResourcesUnsubscribe(request, sessionId, dependencies)
       case 'prompts/get':
         return await handlePromptsGet(request, sessionId, dependencies)
       default:

src/index.ts

@@ -7,7 +7,7 @@ import { MemorySessionStore } from './stores/memory-session-store.ts'
 import { MemoryMessageBroker } from './brokers/memory-message-broker.ts'
 import { RedisSessionStore } from './stores/redis-session-store.ts'
 import { RedisMessageBroker } from './brokers/redis-message-broker.ts'
-import type { MCPPluginOptions, MCPTool, MCPResource, MCPPrompt } from './types.ts'
+import type { MCPPluginOptions, MCPTool, MCPResource, MCPPrompt, ResourceHandlers } from './types.ts'
 import pubsubDecorators from './decorators/pubsub.ts'
 import metaDecorators from './decorators/meta.ts'
 import routes from './routes/mcp.ts'
@@ -50,6 +50,7 @@ const mcpPlugin = fp(async function (app: FastifyInstance, opts: MCPPluginOption
   const tools = new Map<string, MCPTool>()
   const resources = new Map<string, MCPResource>()
   const prompts = new Map<string, MCPPrompt>()
+  const resourceHandlers: ResourceHandlers = {}
 
   // Initialize stores and brokers based on configuration
   let sessionStore: SessionStore
@@ -98,7 +99,8 @@ const mcpPlugin = fp(async function (app: FastifyInstance, opts: MCPPluginOption
   app.register(metaDecorators, {
     tools,
     resources,
-    prompts
+    prompts,
+    resourceHandlers
   })
   app.register(pubsubDecorators, {
     enableSSE,
@@ -116,6 +118,7 @@ const mcpPlugin = fp(async function (app: FastifyInstance, opts: MCPPluginOption
     tools,
     resources,
     prompts,
+    resourceHandlers,
     sessionStore,
     messageBroker,
     localStreams
@@ -188,7 +191,10 @@ export type {
   UnsafeToolHandler,
   UnsafeResourceHandler,
   UnsafePromptHandler,
-  SSESession
+  SSESession,
+  ResourceHandlers,
+  ResourceSubscribeHandler,
+  ResourceUnsubscribeHandler
 } from './types.ts'
 
 // Export authorization types

src/routes/mcp.ts

@@ -3,7 +3,7 @@ import type { FastifyRequest, FastifyReply, FastifyPluginAsync } from 'fastify'
 import fp from 'fastify-plugin'
 import type { JSONRPCMessage } from '../schema.ts'
 import { JSONRPC_VERSION, INTERNAL_ERROR } from '../schema.ts'
-import type { MCPPluginOptions, MCPTool, MCPResource, MCPPrompt } from '../types.ts'
+import type { MCPPluginOptions, MCPTool, MCPResource, MCPPrompt, ResourceHandlers } from '../types.ts'
 import type { SessionStore, SessionMetadata } from '../stores/session-store.ts'
 import type { MessageBroker } from '../brokers/message-broker.ts'
 import type { AuthorizationContext } from '../types/auth-types.ts'
@@ -17,13 +17,14 @@ interface MCPPubSubRoutesOptions {
   tools: Map<string, MCPTool>
   resources: Map<string, MCPResource>
   prompts: Map<string, MCPPrompt>
+  resourceHandlers: ResourceHandlers
   sessionStore: SessionStore
   messageBroker: MessageBroker
   localStreams: Map<string, Set<any>>
 }
 
 const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async (app, options) => {
-  const { enableSSE, opts, capabilities, serverInfo, tools, resources, prompts, sessionStore, messageBroker, localStreams } = options
+  const { enableSSE, opts, capabilities, serverInfo, tools, resources, prompts, resourceHandlers, sessionStore, messageBroker, localStreams } = options
 
   async function createSSESession (): Promise<SessionMetadata> {
     const sessionId = randomUUID()
@@ -185,6 +186,7 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
         tools,
         resources,
         prompts,
+        resourceHandlers,
         request,
         reply,
         authContext

src/types.ts

@@ -24,6 +24,23 @@ export interface HandlerContext {
   authContext?: AuthorizationContext
 }
 
+// Resource subscription handler types
+export type ResourceSubscribeHandler = (
+  params: { uri: string },
+  context: HandlerContext
+) => Promise<Record<string, unknown>> | Record<string, unknown>
+
+export type ResourceUnsubscribeHandler = (
+  params: { uri: string },
+  context: HandlerContext
+) => Promise<Record<string, unknown>> | Record<string, unknown>
+
+// Resource handlers container
+export interface ResourceHandlers {
+  subscribeHandler?: ResourceSubscribeHandler
+  unsubscribeHandler?: ResourceUnsubscribeHandler
+}
+
 // Generic handler types with TypeBox schema support
 export type ToolHandler<TSchema extends TObject = TObject> = (
   params: Static<TSchema>,
@@ -107,6 +124,10 @@ declare module 'fastify' {
       requestedSchema: ElicitRequest['params']['requestedSchema'],
       requestId?: RequestId
     ) => Promise<boolean>
+
+    // Resource subscription handler setters
+    mcpSetResourceSubscribeHandler: (handler: ResourceSubscribeHandler) => void
+    mcpSetResourceUnsubscribeHandler: (handler: ResourceUnsubscribeHandler) => void
   }
 }
 

test/integration.test.ts

@@ -160,7 +160,11 @@ describe('MCP Integration Tests', () => {
         }
       }, CallToolResultSchema)
 
-      t.assert.strictEqual(divResult.content[0].text, 'Result: 5')
+      const divContent = divResult.content[0]
+      t.assert.strictEqual(divContent.type, 'text')
+      if (divContent.type === 'text') {
+        t.assert.strictEqual(divContent.text, 'Result: 5')
+      }
 
       // Test tool execution with error
       const errorResult = await client.request({
@@ -172,7 +176,11 @@ describe('MCP Integration Tests', () => {
       }, CallToolResultSchema)
 
       t.assert.strictEqual(errorResult.isError, true)
-      t.assert.ok((errorResult.content[0].text as string).includes('Invalid operation'))
+      const errorContent = errorResult.content[0]
+      t.assert.strictEqual(errorContent.type, 'text')
+      if (errorContent.type === 'text') {
+        t.assert.ok(errorContent.text.includes('Invalid operation'))
+      }
 
       // Test resources listing
       const resourcesResult = await client.request({
@@ -189,9 +197,11 @@ describe('MCP Integration Tests', () => {
         params: { uri: 'config://settings.json' }
       }, ReadResourceResultSchema)
 
-      t.assert.strictEqual(configResult.contents[0].uri, 'config://settings.json')
-      t.assert.strictEqual(configResult.contents[0].mimeType, 'application/json')
-      const config = JSON.parse(configResult.contents[0].text as string)
+      const configContent = configResult.contents[0]
+      t.assert.strictEqual(configContent.uri, 'config://settings.json')
+      t.assert.strictEqual(configContent.mimeType, 'application/json')
+      t.assert.ok('text' in configContent, 'Expected text content')
+      const config = JSON.parse((configContent as { text: string }).text)
       t.assert.strictEqual(config.mode, 'test')
       t.assert.strictEqual(config.debug, true)
 
@@ -215,7 +225,11 @@ describe('MCP Integration Tests', () => {
 
       t.assert.strictEqual(promptResult.messages.length, 1)
       t.assert.strictEqual(promptResult.messages[0].role, 'user')
-      t.assert.ok((promptResult.messages[0].content.text as string).includes('typescript'))
+      const promptContent = promptResult.messages[0].content
+      t.assert.strictEqual(promptContent.type, 'text')
+      if (promptContent.type === 'text') {
+        t.assert.ok(promptContent.text.includes('typescript'))
+      }
     } catch (error) {
       t.assert.fail(`MCP SDK integration test failed: ${error}`)
     }
@@ -341,6 +355,10 @@ describe('MCP Integration Tests', () => {
     }, CallToolResultSchema)
 
     t.assert.strictEqual(callResult.isError, true)
-    t.assert.ok((callResult.content[0].text as string).includes('no handler implementation'))
+    const noHandlerContent = callResult.content[0]
+    t.assert.strictEqual(noHandlerContent.type, 'text')
+    if (noHandlerContent.type === 'text') {
+      t.assert.ok(noHandlerContent.text.includes('no handler implementation'))
+    }
   })
 })