diff --git a/src/server/streamableHttp.test.ts b/src/server/streamableHttp.test.ts index 8fb55e79..129b499e 100644 --- a/src/server/streamableHttp.test.ts +++ b/src/server/streamableHttp.test.ts @@ -73,7 +73,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "application/json", + "accept": "application/json, text/event-stream", }, body: JSON.stringify(initializeMessage), }); @@ -93,7 +93,7 @@ describe("StreamableHTTPServerTransport", () => { method: "GET", headers: { "mcp-session-id": "invalid-session-id", - "accept": "text/event-stream" + "accept": "application/json, text/event-stream" }, }); @@ -109,7 +109,7 @@ describe("StreamableHTTPServerTransport", () => { const req = createMockRequest({ method: "GET", headers: { - accept: "text/event-stream", + accept: "application/json, text/event-stream", // No mcp-session-id header }, }); @@ -143,7 +143,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "application/json", + "accept": "application/json, text/event-stream", }, body: JSON.stringify(message), }); @@ -161,7 +161,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "application/json", + "accept": "application/json, text/event-stream", "mcp-session-id": "invalid-session-id", // This would cause a 404 in stateful mode }, body: JSON.stringify({ @@ -195,7 +195,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "application/json", + "accept": "application/json, text/event-stream", "mcp-session-id": "non-existent-session-id", // This would be rejected in stateful mode }, body: JSON.stringify(message), @@ -216,7 +216,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - accept: "text/event-stream", + accept: "application/json, text/event-stream", }, body: JSON.stringify({ jsonrpc: "2.0", @@ -242,7 +242,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - accept: "text/event-stream", + accept: "application/json, text/event-stream", "mcp-session-id": "some-random-session-id", }, body: JSON.stringify({ @@ -264,7 +264,7 @@ describe("StreamableHTTPServerTransport", () => { ); }); - it("should handle initialization requests properly in statefull mode", async () => { + it("should handle initialization requests properly in stateful mode", async () => { // Initialize message that would typically be sent during initialization const initializeMessage: JSONRPCMessage = { jsonrpc: "2.0", @@ -281,7 +281,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "application/json", + "accept": "application/json, text/event-stream", }, body: JSON.stringify(initializeMessage), }); @@ -314,7 +314,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "application/json", + "accept": "application/json, text/event-stream", }, body: JSON.stringify(initializeMessage), }); @@ -333,7 +333,7 @@ describe("StreamableHTTPServerTransport", () => { const req = createMockRequest({ method: "GET", headers: { - "accept": "text/event-stream", + "accept": "application/json, text/event-stream", "mcp-session-id": transport.sessionId, }, }); @@ -380,7 +380,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "text/event-stream", + "accept": "application/json, text/event-stream", }, body: JSON.stringify(message), }); @@ -394,7 +394,7 @@ describe("StreamableHTTPServerTransport", () => { expect(mockResponse.writeHead).toHaveBeenCalledWith( 200, expect.objectContaining({ - "Content-Type": "text/event-stream", + "mcp-session-id": transport.sessionId, }) ); }); @@ -435,7 +435,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "application/json", + "accept": "application/json, text/event-stream", "mcp-session-id": transport.sessionId, }, body: JSON.stringify(batchMessages), @@ -488,7 +488,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "text/plain", - "accept": "application/json", + "accept": "application/json, text/event-stream", "mcp-session-id": transport.sessionId, }, body: "test", @@ -532,7 +532,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "text/event-stream", + "accept": "application/json, text/event-stream", "mcp-session-id": transport.sessionId }, body: JSON.stringify(requestMessage) @@ -571,7 +571,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "text/event-stream", + "accept": "application/json, text/event-stream", "mcp-session-id": transport.sessionId }, body: JSON.stringify(requestMessage) @@ -620,7 +620,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "text/event-stream", + "accept": "application/json, text/event-stream", "mcp-session-id": transport.sessionId }, body: JSON.stringify(requestMessage1), @@ -639,7 +639,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "text/event-stream", + "accept": "application/json, text/event-stream", "mcp-session-id": transport.sessionId }, body: JSON.stringify(requestMessage2), @@ -691,7 +691,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "application/json", + "accept": "application/json, text/event-stream", }, body: "invalid json", }); @@ -712,7 +712,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "application/json", + "accept": "application/json, text/event-stream", }, body: JSON.stringify({ invalid: "message" }), }); @@ -745,7 +745,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "application/json", + "accept": "application/json, text/event-stream", }, // No body provided here - it will be passed as parsedBody }); @@ -761,7 +761,7 @@ describe("StreamableHTTPServerTransport", () => { expect(mockResponse.writeHead).toHaveBeenCalledWith( 200, expect.objectContaining({ - "Content-Type": "application/json", + "mcp-session-id": transport.sessionId, }) ); }); @@ -787,7 +787,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "text/event-stream", + "accept": "application/json, text/event-stream", "mcp-session-id": transport.sessionId, }, // No body provided here - it will be passed as parsedBody @@ -825,7 +825,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - "accept": "application/json", + "accept": "application/json, text/event-stream", "mcp-session-id": transport.sessionId, }, body: JSON.stringify(requestBodyMessage), @@ -863,7 +863,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - accept: "text/event-stream", + accept: "application/json, text/event-stream", "mcp-session-id": transportWithHeaders.sessionId }, body: JSON.stringify({ @@ -888,41 +888,11 @@ describe("StreamableHTTPServerTransport", () => { ); }); - it("should include custom headers in JSON response", async () => { - const message: JSONRPCMessage = { - jsonrpc: "2.0", - method: "test", - params: {}, - id: 1, - }; - - const req = createMockRequest({ - method: "POST", - headers: { - "content-type": "application/json", - "accept": "application/json", - "mcp-session-id": transportWithHeaders.sessionId - }, - body: JSON.stringify(message), - }); - - await transportWithHeaders.handleRequest(req, mockResponse); - - expect(mockResponse.writeHead).toHaveBeenCalledWith( - 200, - expect.objectContaining({ - ...customHeaders, - "Content-Type": "application/json", - "mcp-session-id": transportWithHeaders.sessionId - }) - ); - }); - it("should include custom headers in error responses", async () => { const req = createMockRequest({ method: "GET", headers: { - accept: "text/event-stream", + accept: "application/json, text/event-stream", "mcp-session-id": "invalid-session-id" }, }); @@ -948,7 +918,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - accept: "text/event-stream", + accept: "application/json, text/event-stream", "mcp-session-id": transportWithConflictingHeaders.sessionId }, body: JSON.stringify({ @@ -979,7 +949,7 @@ describe("StreamableHTTPServerTransport", () => { method: "POST", headers: { "content-type": "application/json", - accept: "text/event-stream", + accept: "application/json, text/event-stream", "mcp-session-id": transportWithoutHeaders.sessionId }, body: JSON.stringify({ diff --git a/src/server/streamableHttp.ts b/src/server/streamableHttp.ts index 6afe826a..984b2c19 100644 --- a/src/server/streamableHttp.ts +++ b/src/server/streamableHttp.ts @@ -37,7 +37,7 @@ export interface StreamableHTTPServerTransportOptions { * sessionId: randomUUID(), * }); * - * // Stateless mode - explisitly set session ID to undefined + * // Stateless mode - explicitly set session ID to undefined * const statelessTransport = new StreamableHTTPServerTransport({ * sessionId: undefined, * }); @@ -60,7 +60,7 @@ export interface StreamableHTTPServerTransportOptions { * - No session validation is performed */ export class StreamableHTTPServerTransport implements Transport { - // when sessionID is not set, it means the transport is in stateless mode + // when sessionId is not set (undefined), it means the transport is in stateless mode private _sessionId: string | undefined; private _started: boolean = false; private _customHeaders: Record; @@ -71,8 +71,8 @@ export class StreamableHTTPServerTransport implements Transport { onmessage?: (message: JSONRPCMessage) => void; constructor(options: StreamableHTTPServerTransportOptions) { - this._sessionId = options?.sessionId; - this._customHeaders = options?.customHeaders || {}; + this._sessionId = options.sessionId; + this._customHeaders = options.customHeaders || {}; } /** @@ -90,16 +90,6 @@ export class StreamableHTTPServerTransport implements Transport { * Handles an incoming HTTP request, whether GET or POST */ async handleRequest(req: IncomingMessage, res: ServerResponse, parsedBody?: unknown): Promise { - // Only validate session ID for non-initialization requests when session management is enabled - if (this._sessionId !== undefined) { - const isInitializationRequest = req.method === "POST" && - req.headers["content-type"]?.includes("application/json"); - - if (!isInitializationRequest && !this.validateSession(req, res)) { - return; - } - } - if (req.method === "GET") { await this.handleGetRequest(req, res); } else if (req.method === "POST") { @@ -124,7 +114,11 @@ export class StreamableHTTPServerTransport implements Transport { * We choose to return 405 Method Not Allowed as we don't support GET SSE connections yet. */ private async handleGetRequest(req: IncomingMessage, res: ServerResponse): Promise { - // Return 405 Method Not Allowed + // Check session validity first for GET requests when session management is enabled + if (this._sessionId !== undefined && !this.validateSession(req, res)) { + return; + } + res.writeHead(405, { ...this._customHeaders, "Allow": "POST, DELETE" @@ -143,15 +137,16 @@ export class StreamableHTTPServerTransport implements Transport { */ private async handlePostRequest(req: IncomingMessage, res: ServerResponse, parsedBody?: unknown): Promise { try { - // validate the Accept header + // Validate the Accept header const acceptHeader = req.headers.accept; + // The client MUST include an Accept header, listing both application/json and text/event-stream as supported content types. if (!acceptHeader || - (!acceptHeader.includes("application/json") && !acceptHeader.includes("text/event-stream"))) { + !acceptHeader.includes("application/json") || !acceptHeader.includes("text/event-stream")) { res.writeHead(406).end(JSON.stringify({ jsonrpc: "2.0", error: { code: -32000, - message: "Not Acceptable: Client must accept application/json and/or text/event-stream" + message: "Not Acceptable: Client must accept both application/json and text/event-stream" }, id: null })); @@ -192,16 +187,33 @@ export class StreamableHTTPServerTransport implements Transport { messages = [JSONRPCMessageSchema.parse(rawMessage)]; } - if (this._sessionId !== undefined) { - // Check if this is an initialization request - // https://spec.modelcontextprotocol.io/specification/2025-03-26/basic/lifecycle/ - const isInitializationRequest = messages.some( - msg => 'method' in msg && msg.method === 'initialize' && 'id' in msg - ); + // Check if this is an initialization request + // https://spec.modelcontextprotocol.io/specification/2025-03-26/basic/lifecycle/ + const isInitializationRequest = messages.some( + msg => 'method' in msg && msg.method === 'initialize' + ); + if (isInitializationRequest) { + const headers: Record = { + ...this._customHeaders + }; - if (!isInitializationRequest && !this.validateSession(req, res)) { - return; + if (this._sessionId !== undefined) { + headers["mcp-session-id"] = this._sessionId; } + + // Process initialization messages before responding + for (const message of messages) { + this.onmessage?.(message); + } + + res.writeHead(200, headers).end(); + return; + } + // If an Mcp-Session-Id is returned by the server during initialization, + // clients using the Streamable HTTP transport MUST include it + // in the Mcp-Session-Id header on all of their subsequent HTTP requests. + if (this._sessionId !== undefined && !isInitializationRequest && !this.validateSession(req, res)) { + return; } @@ -219,62 +231,35 @@ export class StreamableHTTPServerTransport implements Transport { this.onmessage?.(message); } } else if (hasRequests) { - // if it contains requests, you can choose to return an SSE stream or a JSON response - const useSSE = acceptHeader.includes("text/event-stream"); + const headers: Record = { + ...this._customHeaders, + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + }; - if (useSSE) { - const headers: Record = { - ...this._customHeaders, - "Content-Type": "text/event-stream", - "Cache-Control": "no-cache", - Connection: "keep-alive", - }; - - // For initialization requests, always include the session ID if we have one - // even if we're in stateless mode - if (this._sessionId !== undefined) { - headers["mcp-session-id"] = this._sessionId; - } - - res.writeHead(200, headers); - - // Store the response for this request to send messages back through this connection - // We need to track by request ID to maintain the connection - for (const message of messages) { - if ('method' in message && 'id' in message) { - this._sseResponseMapping.set(message.id, res); - } - } - - // handle each message - for (const message of messages) { - this.onmessage?.(message); - } - - // The server SHOULD NOT close the SSE stream before sending all JSON-RPC responses - // This will be handled by the send() method when responses are ready - } else { - // use direct JSON response - const headers: Record = { - ...this._customHeaders, - "Content-Type": "application/json", - }; - - // For initialization requests, always include the session ID if we have one - // even if we're in stateless mode - if (this._sessionId !== undefined) { - headers["mcp-session-id"] = this._sessionId; - } - - res.writeHead(200, headers); - - // handle each message - for (const message of messages) { - this.onmessage?.(message); - } - - res.end(); + // For initialization requests, always include the session ID if we have one + // even if we're in stateless mode + if (this._sessionId !== undefined) { + headers["mcp-session-id"] = this._sessionId; } + + res.writeHead(200, headers); + + // Store the response for this request to send messages back through this connection + // We need to track by request ID to maintain the connection + for (const message of messages) { + if ('method' in message && 'id' in message) { + this._sseResponseMapping.set(message.id, res); + } + } + + // handle each message + for (const message of messages) { + this.onmessage?.(message); + } + // The server SHOULD NOT close the SSE stream before sending all JSON-RPC responses + // This will be handled by the send() method when responses are ready } } catch (error) { // return JSON-RPC formatted error @@ -295,6 +280,9 @@ export class StreamableHTTPServerTransport implements Transport { * Handles DELETE requests to terminate sessions */ private async handleDeleteRequest(req: IncomingMessage, res: ServerResponse): Promise { + if (!this.validateSession(req, res)) { + return; + } await this.close(); res.writeHead(200).end(); } @@ -346,12 +334,12 @@ export class StreamableHTTPServerTransport implements Transport { async send(message: JSONRPCMessage, options?: { relatedRequestId?: RequestId }): Promise { const relatedRequestId = options?.relatedRequestId; if (relatedRequestId === undefined) { - throw new Error("relatedRequestId is required"); + throw new Error("relatedRequestId is required for Streamable HTTP transport"); } const sseResponse = this._sseResponseMapping.get(relatedRequestId); if (!sseResponse) { - throw new Error("No SSE connection established"); + throw new Error(`No SSE connection established for request ID: ${String(relatedRequestId)}`); } // Send the message as an SSE event