Merge pull request #334 from m-paternostro/mp/notifier

Capability notification for the MCP Server

src/examples/server/simpleStreamableHttp.ts

@@ -12,6 +12,21 @@ const server = new McpServer({
   version: '1.0.0',
 }, { capabilities: { logging: {} } });
 
+// Log the capability invocation details
+server.onCapabilityChange((event) => {
+  switch (event.action) {
+    case 'invoked':
+      console.log(`${event.capabilityType} invocation ${event.invocationIndex}: '${event.capabilityName}' started`);
+      break;
+    case 'completed':
+      console.log(`${event.capabilityType} invocation ${event.invocationIndex}: '${event.capabilityName}' completed in ${event.durationMs}ms`);
+      break;
+    case 'error':
+      console.log(`${event.capabilityType} invocation ${event.invocationIndex}: '${event.capabilityName}' failed in ${event.durationMs}ms: ${event.error}`);
+      break;
+  }
+});
+
 // Register a simple tool that returns a greeting
 server.tool(
   'greet',
@@ -291,4 +306,4 @@ process.on('SIGINT', async () => {
   await server.close();
   console.log('Server shutdown complete');
   process.exit(0);
-});
+});

src/server/index.ts

@@ -104,6 +104,13 @@ export class Server<
     );
   }
 
+  /**
+   * The server's name and version.
+   */
+  getVersion(): { readonly name: string; readonly version: string } {
+    return this._serverInfo;
+  }
+
   /**
    * Registers new capabilities. This can only be called before connecting to a transport.
    *

src/server/mcp.test.ts

@@ -14,7 +14,7 @@ import {
   LoggingMessageNotificationSchema,
   Notification,
 } from "../types.js";
-import { ResourceTemplate } from "./mcp.js";
+import { ResourceTemplate, CapabilityEvent } from "./mcp.js";
 import { completable } from "./completable.js";
 import { UriTemplate } from "../shared/uriTemplate.js";
 
@@ -26,6 +26,10 @@ describe("McpServer", () => {
     });
 
     expect(mcpServer.server).toBeDefined();
+    expect(mcpServer.server.getVersion()).toEqual({
+      name: "test server",
+      version: "1.0",
+    });
   });
 
   test("should allow sending notifications via Server", async () => {
@@ -862,6 +866,95 @@ describe("tool()", () => {
       ),
     ).rejects.toThrow(/Tool nonexistent-tool not found/);
   });
+
+  test("should include duration in completed and error events", async () => {
+    const mcpServer = new McpServer({
+      name: "test server",
+      version: "1.0",
+    });
+
+    const client = new Client({
+      name: "test client",
+      version: "1.0",
+    });
+
+    const events: Array<CapabilityEvent> = [];
+    const subscription = mcpServer.onCapabilityChange((event) => {
+      events.push(event);
+    });
+
+    // Register tools with different behaviors
+    mcpServer.tool("success-tool", async () => {
+      // Simulate some work
+      await new Promise(resolve => setTimeout(resolve, 10));
+      return { content: [{ type: "text", text: "Success" }] };
+    });
+
+    mcpServer.tool("error-tool", async () => {
+      // Simulate some work
+      await new Promise(resolve => setTimeout(resolve, 10));
+      throw new Error("Simulated error");
+    });
+
+    const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair();
+
+    await Promise.all([
+      client.connect(clientTransport),
+      mcpServer.connect(serverTransport),
+    ]);
+
+    // Clear events from registration
+    events.length = 0;
+
+    // Call the success tool
+    await client.request(
+      {
+        method: "tools/call",
+        params: {
+          name: "success-tool",
+        },
+      },
+      CallToolResultSchema,
+    );
+
+    // Find the completed event
+    const completedEvent = events.find(e =>
+      e.capabilityType === "tool" &&
+      e.capabilityName === "success-tool" &&
+      e.action === "completed"
+    ) as Extract<CapabilityEvent, { action: "completed" }>;
+
+    expect(completedEvent).toBeDefined();
+    expect(completedEvent.durationMs).toBeDefined();
+    expect(completedEvent.durationMs).toBeGreaterThan(0);
+
+    // Clear events
+    events.length = 0;
+
+    // Call the error tool
+    await client.request(
+      {
+        method: "tools/call",
+        params: {
+          name: "error-tool",
+        },
+      },
+      CallToolResultSchema,
+    );
+
+    // Find the error event
+    const errorEvent = events.find(e =>
+      e.capabilityType === "tool" &&
+      e.capabilityName === "error-tool" &&
+      e.action === "error"
+    ) as Extract<CapabilityEvent, { action: "error" }>;
+
+    expect(errorEvent).toBeDefined();
+    expect(errorEvent.durationMs).toBeDefined();
+    expect(errorEvent.durationMs).toBeGreaterThan(0);
+
+    subscription.close();
+  });
 });
 
 describe("resource()", () => {
@@ -2512,3 +2605,322 @@ describe("prompt()", () => {
     expect(result.completion.total).toBe(1);
   });
 });
+
+describe("CapabilityEvents", () => {
+  test("should emit capability events when registering and interacting with tools", async () => {
+    const mcpServer = new McpServer({
+      name: "test server",
+      version: "1.0",
+    });
+
+    const events: Array<CapabilityEvent> = [];
+    const subscription = mcpServer.onCapabilityChange((event) => {
+      events.push(event);
+    });
+
+    // Register a tool (should trigger "added" event)
+    const tool = mcpServer.tool("test-tool", async () => ({
+      content: [{ type: "text", text: "Test response" }],
+    }));
+
+    expect(events.length).toBeGreaterThan(0);
+    expect(events[0]).toMatchObject({
+      serverInfo: { name: "test server", version: "1.0" },
+      capabilityType: "tool",
+      capabilityName: "test-tool",
+      action: "added",
+    });
+
+    // Update the tool (should trigger "updated" event)
+    tool.update({
+      description: "Updated description",
+    });
+
+    expect(events.some(e =>
+      e.capabilityType === "tool" &&
+      e.capabilityName === "test-tool" &&
+      e.action === "updated"
+    )).toBe(true);
+
+    // Disable the tool (should trigger "disabled" event)
+    tool.disable();
+
+    expect(events.some(e =>
+      e.capabilityType === "tool" &&
+      e.capabilityName === "test-tool" &&
+      e.action === "disabled"
+    )).toBe(true);
+
+    // Enable the tool (should trigger "enabled" event)
+    tool.enable();
+
+    expect(events.some(e =>
+      e.capabilityType === "tool" &&
+      e.capabilityName === "test-tool" &&
+      e.action === "enabled"
+    )).toBe(true);
+
+    // Clean up
+    subscription.close();
+  });
+
+  test("should emit capability events when tools are invoked", async () => {
+    const mcpServer = new McpServer({
+      name: "test server",
+      version: "1.0",
+    });
+
+    const client = new Client({
+      name: "test client",
+      version: "1.0",
+    });
+
+    const events: Array<CapabilityEvent> = [];
+    const subscription = mcpServer.onCapabilityChange((event) => {
+      events.push(event);
+    });
+
+    mcpServer.tool("test-tool", async () => ({
+      content: [{ type: "text", text: "Test response" }],
+    }));
+
+    // Clear events from registration
+    events.length = 0;
+
+    const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair();
+
+    await Promise.all([
+      client.connect(clientTransport),
+      mcpServer.connect(serverTransport),
+    ]);
+
+    await client.request(
+      {
+        method: "tools/call",
+        params: {
+          name: "test-tool",
+        },
+      },
+      CallToolResultSchema,
+    );
+
+    // Should have "invoked" and "completed" events
+    expect(events.some(e =>
+      e.capabilityType === "tool" &&
+      e.capabilityName === "test-tool" &&
+      e.action === "invoked"
+    )).toBe(true);
+
+    expect(events.some(e =>
+      e.capabilityType === "tool" &&
+      e.capabilityName === "test-tool" &&
+      e.action === "completed"
+    )).toBe(true);
+
+    // The invoked and completed events should have the same invocationIndex
+    const invokedEvent = events.find(e => e.action === "invoked");
+    const completedEvent = events.find(e => e.action === "completed");
+    expect(invokedEvent?.invocationIndex).toBe(completedEvent?.invocationIndex);
+
+    subscription.close();
+  });
+
+  test("should emit capability events for resources", async () => {
+    const mcpServer = new McpServer({
+      name: "test server",
+      version: "1.0",
+    });
+
+    const client = new Client({
+      name: "test client",
+      version: "1.0",
+    });
+
+    const events: Array<CapabilityEvent> = [];
+    const subscription = mcpServer.onCapabilityChange((event) => {
+      events.push(event);
+    });
+
+    // Register a resource
+    const resource = mcpServer.resource("test-resource", "test://resource", async () => ({
+      contents: [{ uri: "test://resource", text: "Test content" }],
+    }));
+
+    expect(events.some(e =>
+      e.capabilityType === "resource" &&
+      e.capabilityName === "test-resource" &&
+      e.action === "added"
+    )).toBe(true);
+
+    // Clear events from registration
+    events.length = 0;
+
+    const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair();
+
+    await Promise.all([
+      client.connect(clientTransport),
+      mcpServer.connect(serverTransport),
+    ]);
+
+    // Read the resource
+    await client.request(
+      {
+        method: "resources/read",
+        params: {
+          uri: "test://resource",
+        },
+      },
+      ReadResourceResultSchema,
+    );
+
+    // Should have "invoked" event
+    expect(events.some(e =>
+      e.capabilityType === "resource" &&
+      e.capabilityName === "test://resource" &&
+      e.action === "invoked"
+    )).toBe(true);
+
+    // Clear events
+    events.length = 0;
+
+    // Remove the resource
+    resource.remove();
+
+    // Should have "removed" event
+    expect(events.some(e =>
+      e.capabilityType === "resource" &&
+      e.capabilityName === "test-resource" &&
+      e.action === "removed"
+    )).toBe(true);
+
+    subscription.close();
+  });
+
+  test("should emit capability events for prompts", async () => {
+    const mcpServer = new McpServer({
+      name: "test server",
+      version: "1.0",
+    });
+
+    const client = new Client({
+      name: "test client",
+      version: "1.0",
+    });
+
+    const events: Array<CapabilityEvent> = [];
+    const subscription = mcpServer.onCapabilityChange((event) => {
+      events.push(event);
+    });
+
+    // Register a prompt
+    mcpServer.prompt("test-prompt", async () => ({
+      messages: [
+        {
+          role: "assistant",
+          content: {
+            type: "text",
+            text: "Test response",
+          },
+        },
+      ],
+    }));
+
+    expect(events.some(e =>
+      e.capabilityType === "prompt" &&
+      e.capabilityName === "test-prompt" &&
+      e.action === "added"
+    )).toBe(true);
+
+    // Clear events from registration
+    events.length = 0;
+
+    const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair();
+
+    await Promise.all([
+      client.connect(clientTransport),
+      mcpServer.connect(serverTransport),
+    ]);
+
+    // Get the prompt
+    await client.request(
+      {
+        method: "prompts/get",
+        params: {
+          name: "test-prompt",
+        },
+      },
+      GetPromptResultSchema,
+    );
+
+    // Should have "invoked" and "completed" events
+    expect(events.some(e =>
+      e.capabilityType === "prompt" &&
+      e.capabilityName === "test-prompt" &&
+      e.action === "invoked"
+    )).toBe(true);
+
+    expect(events.some(e =>
+      e.capabilityType === "prompt" &&
+      e.capabilityName === "test-prompt" &&
+      e.action === "completed"
+    )).toBe(true);
+
+    subscription.close();
+  });
+
+  test("should emit error events when tool execution fails", async () => {
+    const mcpServer = new McpServer({
+      name: "test server",
+      version: "1.0",
+    });
+
+    const client = new Client({
+      name: "test client",
+      version: "1.0",
+    });
+
+    const events: Array<CapabilityEvent> = [];
+    const subscription = mcpServer.onCapabilityChange((event) => {
+      // Only capture error events
+      if (event.action === "error") {
+        events.push(event);
+      }
+    });
+
+    mcpServer.tool("error-tool", async () => {
+      throw new Error("Simulated error");
+    });
+
+    const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair();
+
+    await Promise.all([
+      client.connect(clientTransport),
+      mcpServer.connect(serverTransport),
+    ]);
+
+    // Call the tool that throws an error
+    await client.request(
+      {
+        method: "tools/call",
+        params: {
+          name: "error-tool",
+        },
+      },
+      CallToolResultSchema,
+    );
+
+    // Should have error event
+    expect(events.length).toBe(1);
+    expect(events[0]).toMatchObject({
+      capabilityType: "tool",
+      capabilityName: "error-tool",
+      action: "error",
+    });
+
+    // Type assert to the specific variant of CapabilityEvent that has the error property
+    const errorEvent = events[0] as Extract<CapabilityEvent, { action: "error" }>;
+    expect(errorEvent.error).toBeDefined();
+
+    subscription.close();
+  });
+});

src/shared/eventNotifier.test.ts

@@ -0,0 +1,145 @@
+import { createEventNotifier } from "./eventNotifier.js";
+
+describe("EventNotifier", () => {
+  let notifier: ReturnType<typeof createEventNotifier<string>>;
+
+  beforeEach(() => {
+    notifier = createEventNotifier<string>();
+  });
+
+  test("should notify listeners in registration order", () => {
+    const events: string[] = [];
+    notifier.onEvent((event) => events.push(`first: ${event}`));
+    notifier.onEvent((event) => events.push(`second: ${event}`));
+    notifier.onEvent((event) => events.push(`third: ${event}`));
+
+    notifier.notify("test event");
+
+    expect(events).toEqual([
+      "first: test event",
+      "second: test event",
+      "third: test event",
+    ]);
+  });
+
+  test("should not notify unsubscribed listeners", () => {
+    const events: string[] = [];
+    const subscription = notifier.onEvent((event) => events.push(event));
+
+    notifier.notify("first event");
+    subscription.close();
+    notifier.notify("second event");
+
+    expect(events).toEqual(["first event"]);
+  });
+
+  test("should handle function events", () => {
+    const events: string[] = [];
+    notifier.onEvent((event) => events.push(event));
+
+    notifier.notify(() => "dynamic event");
+
+    expect(events).toEqual(["dynamic event"]);
+  });
+
+  test("should handle errors through error handler", () => {
+    const errors: Error[] = [];
+    notifier.onError((error) => errors.push(error));
+
+    notifier.onEvent(() => {
+      throw new Error("test error");
+    });
+
+    notifier.notify("test event");
+
+    expect(errors).toHaveLength(1);
+    expect(errors[0]).toBeInstanceOf(Error);
+    expect(errors[0].message).toBe("test error");
+  });
+
+  test("should not notify after close", () => {
+    const events: string[] = [];
+    notifier.onEvent((event) => events.push(event));
+
+    notifier.notify("first event");
+    notifier.close();
+    notifier.notify("second event");
+
+    expect(events).toEqual(["first event"]);
+  });
+
+  test("should handle multiple subscriptions and unsubscriptions", () => {
+    const events: string[] = [];
+    const subscription1 = notifier.onEvent((event) =>
+      events.push(`1: ${event}`)
+    );
+    const subscription2 = notifier.onEvent((event) =>
+      events.push(`2: ${event}`)
+    );
+
+    notifier.notify("first event");
+    subscription1.close();
+    notifier.notify("second event");
+    subscription2.close();
+    notifier.notify("third event");
+
+    expect(events).toEqual([
+      "1: first event",
+      "2: first event",
+      "2: second event",
+    ]);
+  });
+
+  test("should handle error handler after close", () => {
+    const errors: Error[] = [];
+    notifier.onError((error) => errors.push(error));
+    notifier.close();
+
+    notifier.onEvent(() => {
+      throw new Error("test error");
+    });
+
+    notifier.notify("test event");
+
+    expect(errors).toHaveLength(0);
+  });
+
+  test("should clear error handler on close", () => {
+    const errors: Error[] = [];
+    notifier.onError((error) => errors.push(error));
+
+    // Close should clear the error handler
+    notifier.close();
+
+    // Setting up a new listener after close
+    notifier.onEvent(() => {
+      throw new Error("test error");
+    });
+
+    // This should not trigger the error handler since the notifier is closed
+    notifier.notify("test event");
+
+    expect(errors).toHaveLength(0);
+  });
+
+  test("should use the last set error handler", () => {
+    const errors1: Error[] = [];
+    const errors2: Error[] = [];
+
+    // First error handler
+    notifier.onError((error) => errors1.push(error));
+
+    // Second error handler should replace the first one
+    notifier.onError((error) => errors2.push(error));
+
+    notifier.onEvent(() => {
+      throw new Error("test error");
+    });
+
+    notifier.notify("test event");
+
+    expect(errors1).toHaveLength(0);
+    expect(errors2).toHaveLength(1);
+    expect(errors2[0].message).toBe("test error");
+  });
+});

src/shared/eventNotifier.ts

@@ -0,0 +1,152 @@
+/**
+ * Provides a simple, type-safe event notification implementation. This module allows components to implement the
+ * observer pattern with minimal boilerplate and proper type checking.
+ */
+
+/**
+ * A type-safe event notifier that manages event listeners and notifications.
+ *
+ * @template T The type of events this notifier will handle
+ * @template E The type of error that can be handled (defaults to Error)
+ *
+ *   EventNotifier provides:
+ *
+ *   - Type-safe event subscriptions via `onEvent`
+ *   - Synchronized event notifications via `notify`
+ *   - Automatic cleanup of resources via `close`
+ *   - Status tracking via `active` property
+ *   - Error handling via optional error callback
+ */
+export type EventNotifier<T, E = Error> = {
+  /**
+   * Registers a listener function to be called when events are notified. Listeners are notified in the order they
+   * were registered.
+   *
+   * @example
+   *
+   * ```ts
+   * const notifier = createEventNotifier<string>();
+   * const subscription = notifier.onEvent((message) => {
+   *   console.log(`Received message: ${message}`);
+   * });
+   *
+   * // Later, to stop listening:
+   * subscription.close();
+   * ```
+   *
+   * @param listener A function that will be called with the notified event
+   * @returns A closeable to unregister the listener (all listeners are unregistered when the notifier is closed).
+   */
+  onEvent: (listener: (event: T) => unknown) => { close: () => void };
+
+  /**
+   * Notifies all registered listeners with the provided event.
+   *
+   * This method:
+   *
+   * - Calls all registered listeners with the event in their registration order
+   * - Ignores errors thrown by listeners (they won't affect other listeners)
+   * - Ignores returned promises (results are not awaited)
+   * - Does nothing if there are no listeners
+   * - If the event is a function, it will be called if there are listeners and its return value will be
+   *   used as the event.
+   *
+   * @example
+   *
+   * ```ts
+   * const notifier = createEventNotifier<{ type: string; value: number }>();
+   * notifier.onEvent((event) => {
+   *   console.log(`Received ${event.type} with value:`, event.value);
+   * });
+   *
+   * notifier.notify({ type: 'progress', value: 75 });
+   * ```
+   *
+   * @param event The event to send to all listeners or a function that returns such event.
+   */
+  notify: (event: T | (() => T)) => void;
+
+  /**
+   * Sets an error handler for the notifier. This handler will be called when a listener throws an error.
+   *
+   * @param handler A function that will be called with any errors thrown by listeners.
+   */
+  onError: (handler: (error: E) => void) => void;
+
+  /**
+   * Closes the notifier and removes all listeners.
+   *
+   * @warning Failing to call close() on subscriptions or the notifier itself may lead to memory leaks.
+   */
+  close: () => void;
+};
+
+/**
+ * Creates a type-safe event notifier.
+ *
+ * @example
+ *
+ * ```ts
+ * // Simple string event notifier
+ * const stringNotifier = createEventNotifier<string>();
+ *
+ * // Complex object event notifier
+ * interface TaskEvent {
+ *   type: 'started' | 'completed' | 'failed';
+ *   taskId: number;
+ *   details: {
+ *     name: string;
+ *     duration?: number;
+ *   };
+ * }
+ * const taskNotifier = createEventNotifier<TaskEvent>();
+ * ```
+ *
+ * @template T The type of events this notifier will handle
+ * @template E The type of error that can be handled (defaults to Error)
+ * @returns A new EventNotifier instance.
+ */
+export const createEventNotifier = <T, E = Error>(): EventNotifier<T, E> => {
+  const listeners = new Set<(event: T) => unknown>();
+  let errorHandler: ((error: E) => void) | undefined;
+
+  return {
+    close: () => {
+      listeners.clear();
+      errorHandler = undefined;
+    },
+
+    onEvent: (listener) => {
+      listeners.add(listener);
+      return {
+        close: () => {
+          listeners.delete(listener);
+        },
+      };
+    },
+
+    notify: (event: T | (() => T)) => {
+      if (!listeners.size) {
+        return;
+      }
+
+      if (typeof event === "function") {
+        event = (event as () => T)();
+      }
+
+      for (const listener of listeners) {
+        try {
+          void listener(event);
+        } catch (error) {
+          if (errorHandler) {
+            errorHandler(error as E);
+          }
+        }
+      }
+    },
+
+    onError: (handler) => {
+      errorHandler = handler;
+    },
+  };
+};