feat: add HTTP transport support with new StreamableHTTPServerTransport class

chrisribe · chrisribe · commit afb99f101d20 · 2025-11-11T00:32:39.000-05:00
diff --git a/package.json b/package.json
@@ -23,6 +23,8 @@
     "version:major": "node scripts/bump-version.js major",
     "prepublishOnly": "npm run build",
     "start": "node dist/index.js",
+    "start:http": "node dist/index.js --http",
+    "start:both": "node dist/index.js --both",
     "cli": "node dist/index.js",
     "dev": "tsx src/index.ts",
     "test": "npm run build && node dist/tests/memory-server-tests.js",
diff --git a/src/index.ts b/src/index.ts
@@ -17,6 +17,7 @@ import { toolRegistry } from './tools/index.js';
 import type { ToolContext } from './types/tools.js';
 import { debugLog } from './utils/debug.js';
 import { checkDatabaseIntegrity, rebuildHashIndex } from './utils/db-integrity-check.js';
+import { StreamableHTTPServerTransport } from './transports/streamable-http.js';
 
 // Initialize server
 const server = new Server(
@@ -117,6 +118,14 @@ server.setRequestHandler(GetPromptRequestSchema, async (request) => {
 async function main() {
   const args = process.argv.slice(2);
   
+  // Check for --http or --both flags
+  const useHttp = args.includes('--http') || args.includes('--both');
+  const useBoth = args.includes('--both');
+  const useStdio = !args.includes('--http') || useBoth;
+  
+  // Remove transport flags from args
+  const cliArgs = args.filter(arg => arg !== '--http' && arg !== '--both');
+  
   // Initialize services (backup auto-configures from env vars)
   memoryService = initializeServices();
   
@@ -126,9 +135,9 @@ async function main() {
     config: {}
   };
   
-  if (args.length > 0) {
+  if (cliArgs.length > 0) {
     // CLI mode - check for integrity commands first
-    if (args[0] === 'check-integrity') {
+    if (cliArgs[0] === 'check-integrity') {
       const dbPath = process.env.MEMORY_DB || './memory.db';
       console.log('Running database integrity check...\n');
       const result = checkDatabaseIntegrity(dbPath);
@@ -151,14 +160,14 @@ async function main() {
       }
       
       process.exit(result.orphanedMemories.length > 0 ? 1 : 0);
-    } else if (args[0] === 'rebuild-index') {
+    } else if (cliArgs[0] === 'rebuild-index') {
       const dbPath = process.env.MEMORY_DB || './memory.db';
       rebuildHashIndex(dbPath);
       process.exit(0);
     }
     
     // CLI mode - handle tool execution
-    const [toolName, ...toolArgs] = args;
+    const [toolName, ...toolArgs] = cliArgs;
     
     if (!toolRegistry.hasTool(toolName)) {
       console.error(`Unknown tool: ${toolName}`);
@@ -176,10 +185,40 @@ async function main() {
       process.exit(1);
     }
   } else {
-    // MCP mode
-    const transport = new StdioServerTransport();
-    await server.connect(transport);
-    debugLog('Simple Memory MCP server running on stdio');
+    // MCP mode - connect transport(s)
+    if (useHttp) {
+      // HTTP transport mode
+      const httpPort = parseInt(process.env.MCP_PORT || '3000', 10);
+      const httpHost = process.env.MCP_HOST || 'localhost';
+      
+      const httpTransport = new StreamableHTTPServerTransport({
+        port: httpPort,
+        host: httpHost
+      });
+      
+      try {
+        await server.connect(httpTransport);
+        console.log(`✅ Simple Memory MCP server running on HTTP: http://${httpHost}:${httpPort}/mcp`);
+      } catch (error) {
+        console.error('Failed to start HTTP transport:', error instanceof Error ? error.message : error);
+        if (error instanceof Error && error.stack) {
+          debugLog('Stack trace:', error.stack);
+        }
+        process.exit(1);
+      }
+    }
+    
+    if (useStdio) {
+      // Stdio transport mode (default or hybrid)
+      if (useBoth) {
+        debugLog('🔌 Hybrid mode: stdio + HTTP');
+        // Servers are already connected above in HTTP block
+      } else {
+        const transport = new StdioServerTransport();
+        await server.connect(transport);
+        debugLog('🔌 Simple Memory MCP server running on stdio');
+      }
+    }
   }
 }
 
diff --git a/src/transports/streamable-http.ts b/src/transports/streamable-http.ts
@@ -0,0 +1,254 @@
+/**
+ * StreamableHTTPServerTransport - MCP transport over HTTP
+ * 
+ * Implements the MCP Server transport interface using HTTP instead of stdio.
+ * Supports multiple clients connecting via HTTP POST requests.
+ */
+
+import { createServer, IncomingMessage, ServerResponse } from 'http';
+import { JSONRPCMessage } from '@modelcontextprotocol/sdk/types.js';
+import { debugLog } from '../utils/debug.js';
+
+interface HTTPTransportOptions {
+  port?: number;
+  host?: string;
+}
+
+/**
+ * Simple request-scoped response buffer for storing responses
+ * Each HTTP request gets its own buffer to collect responses
+ */
+class ResponseBuffer {
+  private responses: JSONRPCMessage[] = [];
+
+  add(message: JSONRPCMessage): void {
+    this.responses.push(message);
+  }
+
+  getAll(): JSONRPCMessage[] {
+    return this.responses;
+  }
+}
+
+// Thread-local style buffer (request-scoped)
+let currentBuffer: ResponseBuffer | null = null;
+
+/**
+ * Helper functions for HTTP responses
+ */
+function sendJSON(res: ServerResponse, statusCode: number, data: any): void {
+  res.writeHead(statusCode, { 'Content-Type': 'application/json' });
+  res.end(JSON.stringify(data));
+}
+
+function sendError(res: ServerResponse, statusCode: number, code: number, message: string, id: any = null): void {
+  sendJSON(res, statusCode, {
+    jsonrpc: '2.0',
+    error: { code, message },
+    id
+  });
+}
+
+function sendSuccess(res: ServerResponse, data: any): void {
+  sendJSON(res, 200, data);
+}
+
+function sendNotFound(res: ServerResponse): void {
+  sendJSON(res, 404, { error: 'Not found' });
+}
+
+function sendPayloadTooLarge(res: ServerResponse): void {
+  sendJSON(res, 413, {
+    error: 'Payload too large',
+    code: -32600
+  });
+}
+
+function sendAccepted(res: ServerResponse): void {
+  res.writeHead(202);
+  res.end();
+}
+
+export class StreamableHTTPServerTransport {
+  private port: number;
+  private host: string;
+  private httpServer: any = null;
+
+  // Transport interface properties (standard MCP interface)
+  onmessage?: (message: JSONRPCMessage) => void;
+  onclose?: () => void;
+  onerror?: (error: Error) => void;
+
+  constructor(options: HTTPTransportOptions = {}) {
+    this.port = options.port || 3000;
+    this.host = options.host || 'localhost';
+  }
+
+  /**
+   * Start the HTTP server
+   */
+  async start(): Promise<void> {
+    if (this.httpServer) {
+      throw new Error('Server already started');
+    }
+
+    return new Promise((resolve, reject) => {
+      this.httpServer = createServer(this.handleRequest.bind(this));
+      
+      this.httpServer.once('error', (error: any) => {
+        this.httpServer = null;
+        if (error.code === 'EADDRINUSE') {
+          reject(new Error(`Port ${this.port} is already in use`));
+        } else {
+          reject(error);
+        }
+      });
+
+      this.httpServer.once('listening', () => {
+        console.log(`✅ HTTP MCP Transport listening on http://${this.host}:${this.port}/mcp`);
+        resolve();
+      });
+
+      this.httpServer.listen(this.port, this.host);
+    });
+  }
+
+  /**
+   * Close the HTTP server
+   */
+  async close(): Promise<void> {
+    return new Promise((resolve) => {
+      if (this.httpServer) {
+        this.httpServer.close(() => {
+          console.log('✓ HTTP MCP Transport closed');
+          resolve();
+        });
+      } else {
+        resolve();
+      }
+    });
+  }
+
+  /**
+   * Send a message to the client
+   * In HTTP mode, responses are buffered and sent in the response body
+   */
+  async send(message: JSONRPCMessage): Promise<void> {
+    debugLog('[HTTP Transport] Sending message:', message);
+    
+    // If we're in a request context, buffer the response
+    if (currentBuffer) {
+      currentBuffer.add(message);
+    } else {
+      debugLog('[HTTP Transport] Warning: send() called outside of request context');
+    }
+  }
+
+  /**
+   * Handle incoming HTTP requests
+   */
+  private async handleRequest(req: IncomingMessage, res: ServerResponse): Promise<void> {
+    // Add CORS headers
+    res.setHeader('Access-Control-Allow-Origin', '*');
+    res.setHeader('Access-Control-Allow-Methods', 'POST, GET, OPTIONS');
+    res.setHeader('Access-Control-Allow-Headers', 'Content-Type');
+
+    // Handle OPTIONS requests
+    if (req.method === 'OPTIONS') {
+      res.writeHead(200);
+      res.end();
+      return;
+    }
+
+    // Health check endpoint
+    if (req.url === '/health' && req.method === 'GET') {
+      sendSuccess(res, { status: 'ok', transport: 'http' });
+      return;
+    }
+
+    // MCP protocol endpoint
+    if (req.url === '/mcp' && req.method === 'POST') {
+      await this.handleMCPRequest(req, res);
+      return;
+    }
+
+    // Unknown endpoint
+    sendNotFound(res);
+  }
+
+  /**
+   * Handle MCP protocol POST requests
+   */
+  private async handleMCPRequest(req: IncomingMessage, res: ServerResponse): Promise<void> {
+    let body = '';
+
+    // Read request body
+    req.on('data', (chunk: Buffer) => {
+        body += chunk.toString('utf-8');
+
+        // Prevent body from growing too large
+        if (body.length > 1024 * 1024) {
+          req.removeAllListeners('data');
+          sendPayloadTooLarge(res);
+        }
+      });
+
+      req.on('end', async () => {
+        try {
+          // Parse JSON-RPC request
+          let request: JSONRPCMessage;
+
+          try {
+            request = JSON.parse(body);
+            debugLog('[HTTP Transport] Received request:', request);
+          } catch (parseError) {
+            debugLog('[HTTP Transport] Parse error:', parseError);
+            sendError(res, 400, -32700, 'Parse error');
+            return;
+          }
+
+          // Create response buffer for this request
+          const buffer = new ResponseBuffer();
+          currentBuffer = buffer;
+
+          try {
+            // Call the message handler (set by MCP Server)
+            if (this.onmessage) {
+              this.onmessage(request);
+              
+              // Wait for handler to process and buffer responses
+              // TODO: Replace with proper async handling
+              await new Promise(resolve => setTimeout(resolve, 100));
+            }
+
+            // Send buffered responses
+            const responses = buffer.getAll();
+
+            if (responses.length > 0) {
+              sendJSON(res, 200, responses.length === 1 ? responses[0] : responses);
+            } else {
+              // No response (notification or async processing)
+              sendAccepted(res);
+            }
+          } finally {
+            currentBuffer = null;
+          }
+        } catch (error: any) {
+          debugLog('[HTTP Transport] Request error:', error);
+
+          if (!res.headersSent) {
+            sendError(res, 500, -32603, error.message || 'Internal server error');
+          }
+        }
+      });
+
+    req.on('error', (error: Error) => {
+      debugLog('[HTTP Transport] Request error:', error);
+      if (!res.headersSent) {
+        sendError(res, 400, -32600, 'Invalid request');
+      }
+    });
+  }
+}
+
+export default StreamableHTTPServerTransport;
