feat: add conformance tests for SEP-1699 SSE polling

examples/clients/typescript/sse-retry-test.ts

@@ -0,0 +1,89 @@
+#!/usr/bin/env node
+
+/**
+ * SSE Retry Test Client
+ *
+ * Tests that the MCP client respects the SSE retry field when reconnecting.
+ * This client connects to a test server that closes the SSE stream mid-tool-call,
+ * then waits for the client to reconnect and sends the tool result.
+ */
+
+import { Client } from '@modelcontextprotocol/sdk/client/index.js';
+import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js';
+import { CallToolResultSchema } from '@modelcontextprotocol/sdk/types.js';
+
+async function main(): Promise<void> {
+  const serverUrl = process.argv[2];
+
+  if (!serverUrl) {
+    console.error('Usage: sse-retry-test <server-url>');
+    process.exit(1);
+  }
+
+  console.log(`Connecting to MCP server at: ${serverUrl}`);
+  console.log('This test validates SSE retry field compliance (SEP-1699)');
+
+  try {
+    const client = new Client(
+      {
+        name: 'sse-retry-test-client',
+        version: '1.0.0'
+      },
+      {
+        capabilities: {}
+      }
+    );
+
+    const transport = new StreamableHTTPClientTransport(new URL(serverUrl), {
+      reconnectionOptions: {
+        initialReconnectionDelay: 1000,
+        maxReconnectionDelay: 10000,
+        reconnectionDelayGrowFactor: 1.5,
+        maxRetries: 3
+      }
+    });
+
+    transport.onerror = (error) => {
+      console.log(`Transport error: ${error.message}`);
+    };
+
+    transport.onclose = () => {
+      console.log('Transport closed');
+    };
+
+    console.log('Initiating connection...');
+    await client.connect(transport);
+    console.log('Connected to MCP server');
+
+    console.log('Calling test_reconnection tool...');
+    console.log(
+      'Server will close SSE stream mid-call and send result after reconnection'
+    );
+
+    const result = await client.request(
+      {
+        method: 'tools/call',
+        params: {
+          name: 'test_reconnection',
+          arguments: {}
+        }
+      },
+      CallToolResultSchema
+    );
+
+    console.log('Tool call completed:', JSON.stringify(result, null, 2));
+
+    await transport.close();
+    console.log('Connection closed successfully');
+
+    process.exit(0);
+  } catch (error) {
+    console.error('Test failed:', error);
+    process.exit(1);
+  }
+}
+
+main().catch((error) => {
+  console.error('Unhandled error:', error);
+  process.exit(1);
+});

examples/servers/typescript/everything-server.ts

@@ -12,7 +12,12 @@ import {
   McpServer,
   ResourceTemplate
 } from '@modelcontextprotocol/sdk/server/mcp.js';
-import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js';
+import {
+  StreamableHTTPServerTransport,
+  EventStore,
+  EventId,
+  StreamId
+} from '@modelcontextprotocol/sdk/server/streamableHttp.js';
 import {
   ElicitResultSchema,
   ListToolsRequestSchema,
@@ -33,6 +38,41 @@ const watchedResourceContent = 'Watched resource content';
 const transports: { [sessionId: string]: StreamableHTTPServerTransport } = {};
 const servers: { [sessionId: string]: McpServer } = {};
 
+// In-memory event store for SEP-1699 resumability
+const eventStoreData = new Map<
+  string,
+  { eventId: string; message: any; streamId: string }
+>();
+
+function createEventStore(): EventStore {
+  return {
+    async storeEvent(streamId: StreamId, message: any): Promise<EventId> {
+      const eventId = `${streamId}::${Date.now()}_${randomUUID()}`;
+      eventStoreData.set(eventId, { eventId, message, streamId });
+      return eventId;
+    },
+    async replayEventsAfter(
+      lastEventId: EventId,
+      { send }: { send: (eventId: EventId, message: any) => Promise<void> }
+    ): Promise<StreamId> {
+      const streamId = lastEventId.split('::')[0];
+      const eventsToReplay: Array<[string, { message: any }]> = [];
+      for (const [eventId, data] of eventStoreData.entries()) {
+        if (data.streamId === streamId && eventId > lastEventId) {
+          eventsToReplay.push([eventId, data]);
+        }
+      }
+      eventsToReplay.sort(([a], [b]) => a.localeCompare(b));
+      for (const [eventId, { message }] of eventsToReplay) {
+        if (Object.keys(message).length > 0) {
+          await send(eventId, message);
+        }
+      }
+      return streamId;
+    }
+  };
+}
+
 // Sample base64 encoded 1x1 red PNG pixel for testing
 const TEST_IMAGE_BASE64 =
   'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg==';
@@ -309,6 +349,46 @@ function createMcpServer() {
     }
   );
 
+  // SEP-1699: Reconnection test tool - closes SSE stream mid-call to test client reconnection
+  mcpServer.registerTool(
+    'test_reconnection',
+    {
+      description:
+        'Tests SSE stream disconnection and client reconnection (SEP-1699). Server will close the stream mid-call and send the result after client reconnects.',
+      inputSchema: {}
+    },
+    async (_args, { sessionId, requestId }) => {
+      const sleep = (ms: number) =>
+        new Promise((resolve) => setTimeout(resolve, ms));
+
+      console.log(`[${sessionId}] Starting test_reconnection tool...`);
+
+      // Get the transport for this session
+      const transport = sessionId ? transports[sessionId] : undefined;
+      if (transport && requestId) {
+        // Close the SSE stream to trigger client reconnection
+        console.log(
+          `[${sessionId}] Closing SSE stream to trigger client polling...`
+        );
+        transport.closeSSEStream(requestId);
+      }
+
+      // Wait for client to reconnect (should respect retry field)
+      await sleep(100);
+
+      console.log(`[${sessionId}] test_reconnection tool complete`);
+
+      return {
+        content: [
+          {
+            type: 'text',
+            text: 'Reconnection test completed successfully. If you received this, the client properly reconnected after stream closure.'
+          }
+        ]
+      };
+    }
+  );
+
   // Sampling tool - requests LLM completion from client
   mcpServer.registerTool(
     'test_sampling',
@@ -1003,6 +1083,8 @@ app.post('/mcp', async (req, res) => {
 
       transport = new StreamableHTTPServerTransport({
         sessionIdGenerator: () => randomUUID(),
+        eventStore: createEventStore(),
+        retryInterval: 5000, // 5 second retry interval for SEP-1699
         onsessioninitialized: (newSessionId) => {
           transports[newSessionId] = transport;
           servers[newSessionId] = mcpServer;