|
| 1 | +/** |
| 2 | + * StreamableHTTPServerTransport - MCP transport over HTTP |
| 3 | + * |
| 4 | + * Implements the MCP Server transport interface using HTTP instead of stdio. |
| 5 | + * Properly handles the async nature of MCP over synchronous HTTP. |
| 6 | + */ |
| 7 | + |
| 8 | +import { createServer, IncomingMessage, ServerResponse } from 'http'; |
| 9 | +import { JSONRPCMessage } from '@modelcontextprotocol/sdk/types.js'; |
| 10 | +import { debugLog } from '../utils/debug.js'; |
| 11 | + |
| 12 | +interface HTTPTransportOptions { |
| 13 | + port?: number; |
| 14 | + host?: string; |
| 15 | + timeout?: number; // Request timeout in ms |
| 16 | +} |
| 17 | + |
| 18 | +export class StreamableHTTPServerTransport { |
| 19 | + private port: number; |
| 20 | + private host: string; |
| 21 | + private timeout: number; |
| 22 | + private httpServer: any = null; |
| 23 | + |
| 24 | + // Map request IDs to their response handlers |
| 25 | + private pendingResponses = new Map<string | number, { |
| 26 | + resolve: (message: JSONRPCMessage) => void; |
| 27 | + timer: NodeJS.Timeout; |
| 28 | + }>(); |
| 29 | + |
| 30 | + // Transport interface properties |
| 31 | + onmessage?: (message: JSONRPCMessage) => void; |
| 32 | + onclose?: () => void; |
| 33 | + onerror?: (error: Error) => void; |
| 34 | + |
| 35 | + constructor(options: HTTPTransportOptions = {}) { |
| 36 | + this.port = options.port || 3000; |
| 37 | + this.host = options.host || 'localhost'; |
| 38 | + this.timeout = options.timeout || 30000; // 30 second default |
| 39 | + } |
| 40 | + |
| 41 | + async start(): Promise<void> { |
| 42 | + if (this.httpServer) { |
| 43 | + throw new Error('Server already started'); |
| 44 | + } |
| 45 | + |
| 46 | + return new Promise((resolve, reject) => { |
| 47 | + this.httpServer = createServer(this.handleRequest.bind(this)); |
| 48 | + |
| 49 | + this.httpServer.once('error', (error: any) => { |
| 50 | + this.httpServer = null; |
| 51 | + if (error.code === 'EADDRINUSE') { |
| 52 | + reject(new Error(`Port ${this.port} is already in use`)); |
| 53 | + } else { |
| 54 | + reject(error); |
| 55 | + } |
| 56 | + }); |
| 57 | + |
| 58 | + this.httpServer.once('listening', () => { |
| 59 | + console.log(`✅ HTTP MCP Transport listening on http://${this.host}:${this.port}/mcp`); |
| 60 | + resolve(); |
| 61 | + }); |
| 62 | + |
| 63 | + this.httpServer.listen(this.port, this.host); |
| 64 | + }); |
| 65 | + } |
| 66 | + |
| 67 | + async close(): Promise<void> { |
| 68 | + // Clear all pending responses |
| 69 | + for (const [id, pending] of this.pendingResponses) { |
| 70 | + clearTimeout(pending.timer); |
| 71 | + pending.resolve({ |
| 72 | + jsonrpc: '2.0', |
| 73 | + error: { code: -32603, message: 'Server shutting down' }, |
| 74 | + id |
| 75 | + }); |
| 76 | + } |
| 77 | + this.pendingResponses.clear(); |
| 78 | + |
| 79 | + return new Promise((resolve) => { |
| 80 | + if (this.httpServer) { |
| 81 | + this.httpServer.close(() => { |
| 82 | + console.log('✓ HTTP MCP Transport closed'); |
| 83 | + resolve(); |
| 84 | + }); |
| 85 | + } else { |
| 86 | + resolve(); |
| 87 | + } |
| 88 | + }); |
| 89 | + } |
| 90 | + |
| 91 | + /** |
| 92 | + * Send a message (response) back to the waiting HTTP request |
| 93 | + */ |
| 94 | + async send(message: JSONRPCMessage): Promise<void> { |
| 95 | + debugLog('[HTTP Transport] Sending message:', message); |
| 96 | + |
| 97 | + // Check if this is a response to a pending request |
| 98 | + if ('id' in message && message.id !== null) { |
| 99 | + const pending = this.pendingResponses.get(message.id); |
| 100 | + if (pending) { |
| 101 | + clearTimeout(pending.timer); |
| 102 | + this.pendingResponses.delete(message.id); |
| 103 | + pending.resolve(message); |
| 104 | + } else { |
| 105 | + debugLog('[HTTP Transport] No pending request for response:', message.id); |
| 106 | + } |
| 107 | + } |
| 108 | + // Notifications don't need handling in HTTP context |
| 109 | + } |
| 110 | + |
| 111 | + private async handleRequest(req: IncomingMessage, res: ServerResponse): Promise<void> { |
| 112 | + // Add CORS headers |
| 113 | + res.setHeader('Access-Control-Allow-Origin', '*'); |
| 114 | + res.setHeader('Access-Control-Allow-Methods', 'POST, GET, OPTIONS'); |
| 115 | + res.setHeader('Access-Control-Allow-Headers', 'Content-Type'); |
| 116 | + |
| 117 | + // Handle OPTIONS requests |
| 118 | + if (req.method === 'OPTIONS') { |
| 119 | + res.writeHead(200); |
| 120 | + res.end(); |
| 121 | + return; |
| 122 | + } |
| 123 | + |
| 124 | + // Health check endpoint |
| 125 | + if (req.url === '/health' && req.method === 'GET') { |
| 126 | + this.sendJSON(res, 200, { status: 'ok', transport: 'http' }); |
| 127 | + return; |
| 128 | + } |
| 129 | + |
| 130 | + // MCP protocol endpoint |
| 131 | + if (req.url === '/mcp' && req.method === 'POST') { |
| 132 | + await this.handleMCPRequest(req, res); |
| 133 | + return; |
| 134 | + } |
| 135 | + |
| 136 | + // Unknown endpoint |
| 137 | + this.sendJSON(res, 404, { error: 'Not found' }); |
| 138 | + } |
| 139 | + |
| 140 | + private async handleMCPRequest(req: IncomingMessage, res: ServerResponse): Promise<void> { |
| 141 | + try { |
| 142 | + const body = await this.readBody(req); |
| 143 | + const request = this.parseRequest(body); |
| 144 | + |
| 145 | + if (!request) { |
| 146 | + this.sendError(res, 400, -32700, 'Parse error'); |
| 147 | + return; |
| 148 | + } |
| 149 | + |
| 150 | + debugLog('[HTTP Transport] Received request:', request); |
| 151 | + |
| 152 | + // Handle based on whether it's a request or notification |
| 153 | + if ('id' in request && request.id !== null) { |
| 154 | + // Request - expects a response |
| 155 | + const response = await this.waitForResponse(request); |
| 156 | + this.sendJSON(res, 200, response); |
| 157 | + } else { |
| 158 | + // Notification - no response expected |
| 159 | + if (this.onmessage) { |
| 160 | + this.onmessage(request); |
| 161 | + } |
| 162 | + res.writeHead(202); |
| 163 | + res.end(); |
| 164 | + } |
| 165 | + } catch (error: any) { |
| 166 | + debugLog('[HTTP Transport] Request error:', error); |
| 167 | + if (!res.headersSent) { |
| 168 | + this.sendError(res, 500, -32603, error.message || 'Internal server error'); |
| 169 | + } |
| 170 | + } |
| 171 | + } |
| 172 | + |
| 173 | + private async waitForResponse(request: JSONRPCMessage): Promise<JSONRPCMessage> { |
| 174 | + return new Promise((resolve) => { |
| 175 | + // Type narrowing - we know this request has an id because we checked before calling this method |
| 176 | + if (!('id' in request) || request.id === null || request.id === undefined) { |
| 177 | + resolve({ |
| 178 | + jsonrpc: '2.0', |
| 179 | + error: { code: -32600, message: 'Invalid request - missing id' } |
| 180 | + } as JSONRPCMessage); |
| 181 | + return; |
| 182 | + } |
| 183 | + |
| 184 | + const id = request.id; |
| 185 | + |
| 186 | + // Set up timeout |
| 187 | + const timer = setTimeout(() => { |
| 188 | + this.pendingResponses.delete(id); |
| 189 | + resolve({ |
| 190 | + jsonrpc: '2.0', |
| 191 | + error: { code: -32603, message: 'Request timeout' }, |
| 192 | + id |
| 193 | + } as JSONRPCMessage); |
| 194 | + }, this.timeout); |
| 195 | + |
| 196 | + // Store the response handler |
| 197 | + this.pendingResponses.set(id, { resolve, timer }); |
| 198 | + |
| 199 | + // Process the request |
| 200 | + if (this.onmessage) { |
| 201 | + this.onmessage(request); |
| 202 | + } |
| 203 | + }); |
| 204 | + } |
| 205 | + |
| 206 | + private async readBody(req: IncomingMessage): Promise<string> { |
| 207 | + return new Promise((resolve, reject) => { |
| 208 | + let body = ''; |
| 209 | + |
| 210 | + req.on('data', (chunk: Buffer) => { |
| 211 | + body += chunk.toString('utf-8'); |
| 212 | + |
| 213 | + // Prevent body from growing too large (1MB limit) |
| 214 | + if (body.length > 1024 * 1024) { |
| 215 | + req.removeAllListeners(); |
| 216 | + reject(new Error('Payload too large')); |
| 217 | + } |
| 218 | + }); |
| 219 | + |
| 220 | + req.on('end', () => resolve(body)); |
| 221 | + req.on('error', reject); |
| 222 | + }); |
| 223 | + } |
| 224 | + |
| 225 | + private parseRequest(body: string): JSONRPCMessage | null { |
| 226 | + try { |
| 227 | + return JSON.parse(body); |
| 228 | + } catch { |
| 229 | + return null; |
| 230 | + } |
| 231 | + } |
| 232 | + |
| 233 | + // Helper methods |
| 234 | + private sendJSON(res: ServerResponse, statusCode: number, data: any): void { |
| 235 | + res.writeHead(statusCode, { 'Content-Type': 'application/json' }); |
| 236 | + res.end(JSON.stringify(data)); |
| 237 | + } |
| 238 | + |
| 239 | + private sendError(res: ServerResponse, statusCode: number, code: number, message: string, id: any = null): void { |
| 240 | + this.sendJSON(res, statusCode, { |
| 241 | + jsonrpc: '2.0', |
| 242 | + error: { code, message }, |
| 243 | + id |
| 244 | + }); |
| 245 | + } |
| 246 | +} |
| 247 | + |
| 248 | +export default StreamableHTTPServerTransport; |
0 commit comments