examples

src/examples/server/simpleStreamableHttp.ts

@@ -1,9 +1,75 @@
 import express, { Request, Response } from 'express';
 import { randomUUID } from 'node:crypto';
 import { McpServer } from '../../server/mcp.js';
-import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js';
+import { EventStore, StreamableHTTPServerTransport } from '../../server/streamableHttp.js';
 import { z } from 'zod';
-import { CallToolResult, GetPromptResult, ReadResourceResult } from '../../types.js';
+import { CallToolResult, GetPromptResult, JSONRPCMessage, ReadResourceResult } from '../../types.js';
+
+// Create a simple in-memory EventStore for resumability
+class InMemoryEventStore implements EventStore {
+  private events: Map<string, { streamId: string, message: JSONRPCMessage }> = new Map();
+
+  /**
+   * Generates a unique event ID for a given stream ID
+   */
+  generateEventId(streamId: string): string {
+    return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`;
+  }
+
+  getStreamIdFromEventId(eventId: string): string {
+    const parts = eventId.split('_');
+    return parts.length > 0 ? parts[0] : '';
+  }
+
+
+  /**
+   * Stores an event with a generated event ID
+   */
+  async storeEvent(streamId: string, message: JSONRPCMessage): Promise<string> {
+    const eventId = this.generateEventId(streamId);
+    console.log(`Storing event ${eventId} for stream ${streamId}`);
+    this.events.set(eventId, { streamId, message });
+    return eventId;
+  }
+
+  /**
+   * Retrieves events that occurred after a specific event
+   */
+  async getEventsAfter(lastEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>> {
+    if (!lastEventId || !this.events.has(lastEventId)) {
+      return [];
+    }
+
+    // Extract the stream ID from the event ID
+    const streamId = lastEventId.split('_')[0];
+    const result: Array<{ eventId: string, message: JSONRPCMessage }> = [];
+    let foundLastEvent = false;
+
+    // Sort events by eventId for chronological ordering
+    const sortedEvents = [...this.events.entries()].sort((a, b) => a[0].localeCompare(b[0]));
+
+    for (const [eventId, { streamId: eventStreamId, message }] of sortedEvents) {
+      // Only include events from the same stream
+      if (eventStreamId !== streamId) {
+        continue;
+      }
+
+      // Start collecting events after we find the lastEventId
+      if (eventId === lastEventId) {
+        foundLastEvent = true;
+        continue;
+      }
+
+      if (foundLastEvent) {
+        result.push({ eventId, message });
+      }
+    }
+
+    console.log(`Found ${result.length} events after ${lastEventId} for replay`);
+    return result;
+  }
+
+}
 
 // Create an MCP server with implementation details
 const server = new McpServer({
@@ -92,6 +158,43 @@ server.prompt(
   }
 );
 
+// Register a tool specifically for testing resumability
+server.tool(
+  'start-notification-stream',
+  'Starts sending periodic notifications for testing resumability',
+  {
+    interval: z.number().describe('Interval in milliseconds between notifications').default(100),
+    count: z.number().describe('Number of notifications to send (0 for 100)').default(50),
+  },
+  async ({ interval, count }, { sendNotification }): Promise<CallToolResult> => {
+    const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
+    let counter = 0;
+
+    while (count === 0 || counter < count) {
+      counter++;
+      await sendNotification({
+        method: "notifications/message",
+        params: {
+          level: "info",
+          data: `Periodic notification #${counter} at ${new Date().toISOString()}`
+        }
+      });
+
+      // Wait for the specified interval
+      await sleep(interval);
+    }
+
+    return {
+      content: [
+        {
+          type: 'text',
+          text: `Started sending periodic notifications every ${interval}ms`,
+        }
+      ],
+    };
+  }
+);
+
 // Create a simple resource at a fixed URI
 server.resource(
   'greeting-resource',
@@ -127,8 +230,10 @@ app.post('/mcp', async (req: Request, res: Response) => {
       transport = transports[sessionId];
     } else if (!sessionId && isInitializeRequest(req.body)) {
       // New initialization request
+      const eventStore = new InMemoryEventStore();
       transport = new StreamableHTTPServerTransport({
         sessionIdGenerator: () => randomUUID(),
+        eventStore, // Enable resumability
       });
 
       // Connect the transport to the MCP server BEFORE handling the request
@@ -182,7 +287,14 @@ app.get('/mcp', async (req: Request, res: Response) => {
     return;
   }
 
-  console.log(`Establishing SSE stream for session ${sessionId}`);
+  // Check for Last-Event-ID header for resumability
+  const lastEventId = req.headers['last-event-id'] as string | undefined;
+  if (lastEventId) {
+    console.log(`Client reconnecting with Last-Event-ID: ${lastEventId}`);
+  } else {
+    console.log(`Establishing new SSE stream for session ${sessionId}`);
+  }
+
   const transport = transports[sessionId];
   await transport.handleRequest(req, res);
 });