From c053c2e97dcb0a1cf8be0583bd17a8b3d6af35ec Mon Sep 17 00:00:00 2001 From: ihrpr Date: Wed, 16 Apr 2025 11:56:28 +0100 Subject: [PATCH] handle session termination --- src/client/streamableHttp.test.ts | 75 +++++++++++++++++++++ src/client/streamableHttp.ts | 47 +++++++++++++ src/examples/client/simpleStreamableHttp.ts | 48 ++++++++++++- src/examples/server/simpleStreamableHttp.ts | 42 ++++++++++++ 4 files changed, 211 insertions(+), 1 deletion(-) diff --git a/src/client/streamableHttp.test.ts b/src/client/streamableHttp.test.ts index 93e44150..ac129e29 100644 --- a/src/client/streamableHttp.test.ts +++ b/src/client/streamableHttp.test.ts @@ -101,6 +101,81 @@ describe("StreamableHTTPClientTransport", () => { expect(lastCall[1].headers.get("mcp-session-id")).toBe("test-session-id"); }); + it("should terminate session with DELETE request", async () => { + // First, simulate getting a session ID + const message: JSONRPCMessage = { + jsonrpc: "2.0", + method: "initialize", + params: { + clientInfo: { name: "test-client", version: "1.0" }, + protocolVersion: "2025-03-26" + }, + id: "init-id" + }; + + (global.fetch as jest.Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ "content-type": "text/event-stream", "mcp-session-id": "test-session-id" }), + }); + + await transport.send(message); + expect(transport.sessionId).toBe("test-session-id"); + + // Now terminate the session + (global.fetch as jest.Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers() + }); + + await transport.terminateSession(); + + // Verify the DELETE request was sent with the session ID + const calls = (global.fetch as jest.Mock).mock.calls; + const lastCall = calls[calls.length - 1]; + expect(lastCall[1].method).toBe("DELETE"); + expect(lastCall[1].headers.get("mcp-session-id")).toBe("test-session-id"); + + // The session ID should be cleared after successful termination + expect(transport.sessionId).toBeUndefined(); + }); + + it("should handle 405 response when server doesn't support session termination", async () => { + // First, simulate getting a session ID + const message: JSONRPCMessage = { + jsonrpc: "2.0", + method: "initialize", + params: { + clientInfo: { name: "test-client", version: "1.0" }, + protocolVersion: "2025-03-26" + }, + id: "init-id" + }; + + (global.fetch as jest.Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ "content-type": "text/event-stream", "mcp-session-id": "test-session-id" }), + }); + + await transport.send(message); + + // Now terminate the session, but server responds with 405 + (global.fetch as jest.Mock).mockResolvedValueOnce({ + ok: false, + status: 405, + statusText: "Method Not Allowed", + headers: new Headers() + }); + + // This should not throw an error + await transport.terminateSession(); + + // The session ID should still be preserved since termination wasn't accepted + expect(transport.sessionId).toBe("test-session-id"); + }); + it("should handle 404 response when session expires", async () => { const message: JSONRPCMessage = { jsonrpc: "2.0", diff --git a/src/client/streamableHttp.ts b/src/client/streamableHttp.ts index 9efd8536..72559f35 100644 --- a/src/client/streamableHttp.ts +++ b/src/client/streamableHttp.ts @@ -467,4 +467,51 @@ export class StreamableHTTPClientTransport implements Transport { get sessionId(): string | undefined { return this._sessionId; } + + /** + * Terminates the current session by sending a DELETE request to the server. + * + * Clients that no longer need a particular session + * (e.g., because the user is leaving the client application) SHOULD send an + * HTTP DELETE to the MCP endpoint with the Mcp-Session-Id header to explicitly + * terminate the session. + * + * The server MAY respond with HTTP 405 Method Not Allowed, indicating that + * the server does not allow clients to terminate sessions. + */ + async terminateSession(): Promise { + if (!this._sessionId) { + return; // No session to terminate + } + + try { + const headers = await this._commonHeaders(); + + const init = { + ...this._requestInit, + method: "DELETE", + headers, + signal: this._abortController?.signal, + }; + + const response = await fetch(this._url, init); + + // We specifically handle 405 as a valid response according to the spec, + // meaning the server does not support explicit session termination + if (!response.ok && response.status !== 405) { + throw new StreamableHTTPError( + response.status, + `Failed to terminate session: ${response.statusText}` + ); + } + + // If session termination was successful, clear our session ID + if (response.ok) { + this._sessionId = undefined; + } + } catch (error) { + this.onerror?.(error as Error); + throw error; + } + } } diff --git a/src/examples/client/simpleStreamableHttp.ts b/src/examples/client/simpleStreamableHttp.ts index d0d5408b..c1501a57 100644 --- a/src/examples/client/simpleStreamableHttp.ts +++ b/src/examples/client/simpleStreamableHttp.ts @@ -48,6 +48,7 @@ function printHelp(): void { console.log('\nAvailable commands:'); console.log(' connect [url] - Connect to MCP server (default: http://localhost:3000/mcp)'); console.log(' disconnect - Disconnect from server'); + console.log(' terminate-session - Terminate the current session'); console.log(' reconnect - Reconnect to the server'); console.log(' list-tools - List available tools'); console.log(' call-tool [args] - Call a tool with optional JSON arguments'); @@ -76,6 +77,10 @@ function commandLoop(): void { await disconnect(); break; + case 'terminate-session': + await terminateSession(); + break; + case 'reconnect': await reconnect(); break; @@ -249,6 +254,36 @@ async function disconnect(): Promise { } } +async function terminateSession(): Promise { + if (!client || !transport) { + console.log('Not connected.'); + return; + } + + try { + console.log('Terminating session with ID:', transport.sessionId); + await transport.terminateSession(); + console.log('Session terminated successfully'); + + // Check if sessionId was cleared after termination + if (!transport.sessionId) { + console.log('Session ID has been cleared'); + sessionId = undefined; + + // Also close the transport and clear client objects + await transport.close(); + console.log('Transport closed after session termination'); + client = null; + transport = null; + } else { + console.log('Server responded with 405 Method Not Allowed (session termination not supported)'); + console.log('Session ID is still active:', transport.sessionId); + } + } catch (error) { + console.error('Error terminating session:', error); + } +} + async function reconnect(): Promise { if (client) { await disconnect(); @@ -411,13 +446,24 @@ async function listResources(): Promise { async function cleanup(): Promise { if (client && transport) { try { + // First try to terminate the session gracefully + if (transport.sessionId) { + try { + console.log('Terminating session before exit...'); + await transport.terminateSession(); + console.log('Session terminated successfully'); + } catch (error) { + console.error('Error terminating session:', error); + } + } + + // Then close the transport await transport.close(); } catch (error) { console.error('Error closing transport:', error); } } - process.stdin.setRawMode(false); readline.close(); console.log('\nGoodbye!'); diff --git a/src/examples/server/simpleStreamableHttp.ts b/src/examples/server/simpleStreamableHttp.ts index a318b738..0ae0f910 100644 --- a/src/examples/server/simpleStreamableHttp.ts +++ b/src/examples/server/simpleStreamableHttp.ts @@ -255,6 +255,15 @@ app.post('/mcp', async (req: Request, res: Response) => { } }); + // Set up onclose handler to clean up transport when closed + transport.onclose = () => { + const sid = transport.sessionId; + if (sid && transports[sid]) { + console.log(`Transport closed for session ${sid}, removing from transports map`); + delete transports[sid]; + } + }; + // Connect the transport to the MCP server BEFORE handling the request // so responses can flow back through the same transport await server.connect(transport); @@ -312,6 +321,27 @@ app.get('/mcp', async (req: Request, res: Response) => { await transport.handleRequest(req, res); }); +// Handle DELETE requests for session termination (according to MCP spec) +app.delete('/mcp', async (req: Request, res: Response) => { + const sessionId = req.headers['mcp-session-id'] as string | undefined; + if (!sessionId || !transports[sessionId]) { + res.status(400).send('Invalid or missing session ID'); + return; + } + + console.log(`Received session termination request for session ${sessionId}`); + + try { + const transport = transports[sessionId]; + await transport.handleRequest(req, res); + } catch (error) { + console.error('Error handling session termination:', error); + if (!res.headersSent) { + res.status(500).send('Error processing session termination'); + } + } +}); + // Start the server const PORT = 3000; app.listen(PORT, () => { @@ -343,6 +373,18 @@ app.listen(PORT, () => { // Handle server shutdown process.on('SIGINT', async () => { console.log('Shutting down server...'); + + // Close all active transports to properly clean up resources + for (const sessionId in transports) { + try { + console.log(`Closing transport for session ${sessionId}`); + await transports[sessionId].close(); + delete transports[sessionId]; + } catch (error) { + console.error(`Error closing transport for session ${sessionId}:`, error); + } + } await server.close(); + console.log('Server shutdown complete'); process.exit(0); }); \ No newline at end of file