chore: publish resource updated notification only when there are subscriptions

Author: himanshusinghs

src/resources/common/exportedData.ts

@@ -15,15 +15,12 @@ export class ExportedData {
     private server?: Server;
 
     constructor(private readonly session: Session) {
-        this.session.exportsManager.on("export-available", (uri) => {
-            this.server?.mcpServer.sendResourceListChanged();
-            void this.server?.mcpServer.server.sendResourceUpdated({
-                uri,
-            });
-        });
-        this.session.exportsManager.on("export-expired", () => {
-            this.server?.mcpServer.sendResourceListChanged();
-        });
+        const onExportChanged = (uri: string): void => {
+            this.server?.sendResourceListChanged();
+            this.server?.sendResourceUpdated(uri);
+        };
+        this.session.exportsManager.on("export-available", onExportChanged);
+        this.session.exportsManager.on("export-expired", onExportChanged);
     }
 
     public register(server: Server): void {

src/resources/resource.ts

@@ -83,10 +83,10 @@ export abstract class ReactiveResource<Value, RelevantEvents extends readonly (k
         ],
     });
 
-    private async triggerUpdate(): Promise<void> {
+    private triggerUpdate(): void {
         try {
-            await this.server?.mcpServer.server.sendResourceUpdated({ uri: this.uri });
-            this.server?.mcpServer.sendResourceListChanged();
+            this.server?.sendResourceListChanged();
+            this.server?.sendResourceUpdated(this.uri);
         } catch (error: unknown) {
             this.session.logger.warning({
                 id: LogId.resourceUpdateFailure,

src/server.ts

@@ -9,7 +9,12 @@ import { Telemetry } from "./telemetry/telemetry.js";
 import { UserConfig } from "./common/config.js";
 import { type ServerEvent } from "./telemetry/types.js";
 import { type ServerCommand } from "./telemetry/types.js";
-import { CallToolRequestSchema, CallToolResult } from "@modelcontextprotocol/sdk/types.js";
+import {
+    CallToolRequestSchema,
+    CallToolResult,
+    SubscribeRequestSchema,
+    UnsubscribeRequestSchema,
+} from "@modelcontextprotocol/sdk/types.js";
 import assert from "assert";
 import { ToolBase } from "./tools/tool.js";
 
@@ -27,6 +32,7 @@ export class Server {
     public readonly userConfig: UserConfig;
     public readonly tools: ToolBase[] = [];
     private readonly startTime: number;
+    private readonly subscriptions = new Set<string>();
 
     constructor({ session, mcpServer, userConfig, telemetry }: ServerOptions) {
         this.startTime = Date.now();
@@ -42,7 +48,7 @@ export class Server {
         this.registerResources();
         await this.validateConfig();
 
-        this.mcpServer.server.registerCapabilities({ logging: {}, resources: { listChanged: true } });
+        this.mcpServer.server.registerCapabilities({ logging: {}, resources: { listChanged: true, subscribe: true } });
 
         // TODO: Eventually we might want to make tools reactive too instead of relying on custom logic.
         this.registerTools();
@@ -70,6 +76,26 @@ export class Server {
             return existingHandler(request, extra);
         });
 
+        this.mcpServer.server.setRequestHandler(SubscribeRequestSchema, ({ params }) => {
+            this.subscriptions.add(params.uri);
+            this.session.logger.debug({
+                id: LogId.serverInitialized,
+                context: "resources",
+                message: `Client subscribed to resource: ${params.uri}`,
+            });
+            return {};
+        });
+
+        this.mcpServer.server.setRequestHandler(UnsubscribeRequestSchema, ({ params }) => {
+            this.subscriptions.delete(params.uri);
+            this.session.logger.debug({
+                id: LogId.serverInitialized,
+                context: "resources",
+                message: `Client unsubscribed from resource: ${params.uri}`,
+            });
+            return {};
+        });
+
         this.mcpServer.server.oninitialized = (): void => {
             this.session.setAgentRunner(this.mcpServer.server.getClientVersion());
 
@@ -101,6 +127,16 @@ export class Server {
         await this.mcpServer.close();
     }
 
+    public sendResourceListChanged(): void {
+        this.mcpServer.sendResourceListChanged();
+    }
+
+    public sendResourceUpdated(uri: string): void {
+        if (this.subscriptions.has(uri)) {
+            void this.mcpServer.server.sendResourceUpdated({ uri });
+        }
+    }
+
     /**
      * Emits a server event
      * @param command - The server command (e.g., "start", "stop", "register", "deregister")

tests/integration/helpers.ts

@@ -286,6 +286,7 @@ export function timeout(ms: number): Promise<void> {
  */
 export function resourceChangedNotification(client: Client, uri: string): Promise<void> {
     return new Promise<void>((resolve) => {
+        void client.subscribeResource({ uri });
         client.setNotificationHandler(ResourceUpdatedNotificationSchema, (notification) => {
             if (notification.params.uri === uri) {
                 resolve();