feat: Streaming http support [DEV-5079] (#43)

sownak · dependabot[bot] · web-flow · commit 820245b459ef · 2025-06-11T19:19:42.000+05:30
* updated packages Signed-off-by: Sownak Roy <sownak@cheqd.io> * fix lockfile Signed-off-by: Sownak Roy <sownak@cheqd.io> * fix: support for streamable http Signed-off-by: Sownak Roy <sownak@cheqd.io> * chore(deps-dev): bump @types/node from 20.17.47 to 20.19.0 (#41) Bumps [@types/node](https://github.com/DefinitelyTyped/DefinitelyTyped/tree/HEAD/types/node) from 20.17.47 to 20.19.0. - [Release notes](https://github.com/DefinitelyTyped/DefinitelyTyped/releases) - [Commits](https://github.com/DefinitelyTyped/DefinitelyTyped/commits/HEAD/types/node) --- updated-dependencies: - dependency-name: "@types/node" dependency-version: 20.19.0 dependency-type: direct:development update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> --------- Signed-off-by: Sownak Roy <sownak@cheqd.io> Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
diff --git a/package.json b/package.json
@@ -34,7 +34,7 @@
     "@semantic-release/npm": "^12.0.1",
     "@semantic-release/release-notes-generator": "^14.0.3",
     "@tsconfig/strictest": "^2.0.5",
-    "@types/node": "^20.19.0",
+    "@types/node": "^24.0.0",
     "conventional-changelog-conventionalcommits": "^9.0.0",
     "husky": "^9.1.7",
     "lerna": "^8.2.2",
diff --git a/packages/credo/package.json b/packages/credo/package.json
@@ -40,7 +40,7 @@
     "@modelcontextprotocol/sdk": "^1.12.1",
     "zod": "^3.25.56"  },
   "devDependencies": {
-    "@types/node": "^20.19.0",
+    "@types/node": "^24.0.0",
     "@types/qrcode": "^1.5.5",
     "typescript": "^5.8.3"
   },
diff --git a/packages/remote-server/package.json b/packages/remote-server/package.json
@@ -37,7 +37,7 @@
     "@types/cors": "^2.8.17",
     "@types/express": "^5.0.1",
     "@types/helmet": "^4.0.0",
-    "@types/node": "^20.19.0",
+    "@types/node": "^24.0.0",
     "dotenv": "^16.5.0",
     "typescript": "^5.8.3"
   },
diff --git a/packages/remote-server/src/app.ts b/packages/remote-server/src/app.ts
@@ -4,8 +4,10 @@ import cors from 'cors';
 import cookieParser from 'cookie-parser';
 import { StatusCodes } from 'http-status-codes';
 import { config } from 'dotenv';
-import { SSEServerTransport } from '@modelcontextprotocol/sdk/server/sse.js';
 import { AgentMcpServer, normalizeEnvVar } from '@cheqd/mcp-toolkit-server';
+import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js';
+import { isInitializeRequest } from '@modelcontextprotocol/sdk/types.js';
+import { randomUUID } from 'node:crypto';
 
 config();
 
@@ -47,39 +49,75 @@ class App {
 
 	private routes() {
 		const app = this.express;
-		const transports: { [sessionId: string]: SSEServerTransport } = {};
+		app.use(express.json());
 
-		app.get('/sse', async (_req: Request, res: Response) => {
-			const transport = new SSEServerTransport('/messages', res);
-			transports[transport.sessionId] = transport;
+		// Map to store transports by session ID
+		const transports: { [sessionId: string]: StreamableHTTPServerTransport } = {};
 
-			console.log('SSE session started:', transport.sessionId);
-
-			res.on('close', () => {
-				console.log('SSE session closed:', transport.sessionId);
-				delete transports[transport.sessionId];
-			});
-
-			console.log('Connecting to server...');
-			await this.server.getStatus();
-			if (this.server) {
-				await this.server.start(transport).catch((err) => {
-					console.error('Unhandled error in server startup:', err);
+		// Handle POST requests for client-to-server communication
+		app.post('/mcp', async (_req: Request, res: Response) => {
+			// Check for existing session ID
+			const sessionId = _req.headers['mcp-session-id'] as string | undefined;
+			let transport: StreamableHTTPServerTransport;
+			if (sessionId && transports[sessionId]) {
+				transport = transports[sessionId];
+			} else if (!sessionId && isInitializeRequest(_req.body)) {
+				transport = new StreamableHTTPServerTransport({
+					sessionIdGenerator: () => randomUUID(),
+					onsessioninitialized: (sessionId) => {
+						// Store the transport by session ID
+						transports[sessionId] = transport;
+					},
+				});
+				// Clean up transport when closed
+				transport.onclose = () => {
+					if (transport.sessionId) {
+						console.log('Session closed:', transport.sessionId);
+						delete transports[transport.sessionId];
+					}
+				};
+				console.log('Connecting to server...');
+				// Connect to the MCP server
+				await this.server.getStatus();
+				if (this.server) {
+					await this.server.start(transport).catch((err) => {
+						console.error('Unhandled error in server startup:', err);
+					});
+					console.log('Connected.');
+				}
+				console.log('MCP session started');
+			} else {
+				// Invalid request
+				res.status(400).json({
+					jsonrpc: '2.0',
+					error: {
+						code: -32000,
+						message: 'Bad Request: No valid session ID provided',
+					},
+					id: null,
 				});
-				console.log('Connected.');
+				return;
 			}
+
+			// Handle the request
+			await transport.handleRequest(_req, res, _req.body);
 		});
+		// Reusable handler for GET and DELETE requests
+		const handleSessionRequest = async (req: express.Request, res: express.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;
+			}
 
-		app.post('/messages', async (req, res) => {
-			const sessionId = req.query.sessionId as string;
 			const transport = transports[sessionId];
+			await transport.handleRequest(req, res);
+		};
+		// Handle GET requests for server-to-client notifications via SSE
+		app.get('/mcp', handleSessionRequest);
 
-			if (transport) {
-				await transport.handlePostMessage(req, res);
-			} else {
-				res.status(400).send('No transport found for sessionId');
-			}
-		});
+		// Handle DELETE requests for session termination
+		app.delete('/mcp', handleSessionRequest);
 
 		app.get('/status', async (_req: Request, res: Response) => {
 			// Basic server information
diff --git a/packages/server/package.json b/packages/server/package.json
@@ -35,7 +35,7 @@
     "zod": "^3.25.56"
   },
   "devDependencies": {
-    "@types/node": "^20.19.0",
+    "@types/node": "^24.0.0",
     "typescript": "^5.8.3"
   },
   "publishConfig": {
diff --git a/packages/server/src/server.ts b/packages/server/src/server.ts
@@ -4,6 +4,7 @@ import { SSEServerTransport } from '@modelcontextprotocol/sdk/server/sse.js';
 import { CredoToolKit } from '@cheqd/mcp-toolkit-credo';
 import { ToolDefinition } from '@cheqd/mcp-toolkit-credo/build/types.js';
 import { IAgentMCPServerOptions } from './types/index.js';
+import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js';
 import { readFile } from 'fs/promises';
 import path from 'path';
 import { fileURLToPath } from 'url';
@@ -17,7 +18,7 @@ const packageJson = JSON.parse(await readFile(packageJsonPath, 'utf-8'));
  * for the cheqd agent, including tool setup, signal handling, and proper cleanup.
  */
 export class AgentMcpServer extends McpServer {
-	private transport: StdioServerTransport | SSEServerTransport | null = null;
+	private transport: StdioServerTransport | SSEServerTransport | StreamableHTTPServerTransport | null = null;
 	private credoToolkit: CredoToolKit | null = null;
 	private options: IAgentMCPServerOptions;
 
@@ -161,7 +162,7 @@ export class AgentMcpServer extends McpServer {
 	/*
 	 * Start the server and connect to the specified transport
 	 */
-	async start(transport?: StdioServerTransport | SSEServerTransport): Promise<void> {
+	async start(transport?: StdioServerTransport | SSEServerTransport | StreamableHTTPServerTransport): Promise<void> {
 		try {
 			this.transport = transport || new StdioServerTransport();
 			await this.connect(this.transport);
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
