Implement support for streamable-http transport type mcp servers

Author: taylorwilsdon

src/services/mcp/McpHub.ts

@@ -1,6 +1,7 @@
 import { Client } from "@modelcontextprotocol/sdk/client/index.js"
 import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"
 import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js"
+import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"
 import ReconnectingEventSource from "reconnecting-eventsource"
 import {
 	CallToolResultSchema,
@@ -35,7 +36,7 @@ import { injectEnv } from "../../utils/config"
 export type McpConnection = {
 	server: McpServer
 	client: Client
-	transport: StdioClientTransport | SSEClientTransport
+	transport: StdioClientTransport | SSEClientTransport | StreamableHTTPClientTransport
 }
 
 // Base configuration schema for common settings
@@ -47,14 +48,17 @@ const BaseConfigSchema = z.object({
 })
 
 // Custom error messages for better user feedback
-const typeErrorMessage = "Server type must be either 'stdio' or 'sse'"
+const typeErrorMessage = "Server type must be 'stdio', 'sse', or 'streamable-http'"
 const stdioFieldsErrorMessage =
 	"For 'stdio' type servers, you must provide a 'command' field and can optionally include 'args' and 'env'"
 const sseFieldsErrorMessage =
 	"For 'sse' type servers, you must provide a 'url' field and can optionally include 'headers'"
+const streamableHttpFieldsErrorMessage =
+	"For 'streamable-http' type servers, you must provide a 'url' field and can optionally include 'headers'"
 const mixedFieldsErrorMessage =
-	"Cannot mix 'stdio' and 'sse' fields. For 'stdio' use 'command', 'args', and 'env'. For 'sse' use 'url' and 'headers'"
-const missingFieldsErrorMessage = "Server configuration must include either 'command' (for stdio) or 'url' (for sse)"
+	"Cannot mix 'stdio' and ('sse' or 'streamable-http') fields. For 'stdio' use 'command', 'args', and 'env'. For 'sse'/'streamable-http' use 'url' and 'headers'"
+const missingFieldsErrorMessage =
+	"Server configuration must include either 'command' (for stdio) or 'url' (for sse/streamable-http) and a corresponding 'type' if 'url' is used."
 
 // Helper function to create a refined schema with better error messages
 const createServerTypeSchema = () => {
@@ -90,6 +94,23 @@ const createServerTypeSchema = () => {
 				type: "sse" as const,
 			}))
 			.refine((data) => data.type === undefined || data.type === "sse", { message: typeErrorMessage }),
+		// StreamableHTTP config (has url field)
+		BaseConfigSchema.extend({
+			type: z.enum(["streamable-http"]).optional(),
+			url: z.string().url("URL must be a valid URL format"),
+			headers: z.record(z.string()).optional(),
+			// Ensure no stdio fields are present
+			command: z.undefined().optional(),
+			args: z.undefined().optional(),
+			env: z.undefined().optional(),
+		})
+			.transform((data) => ({
+				...data,
+				type: "streamable-http" as const,
+			}))
+			.refine((data) => data.type === undefined || data.type === "streamable-http", {
+				message: typeErrorMessage,
+			}),
 	])
 }
 
@@ -152,33 +173,43 @@ export class McpHub {
 	private validateServerConfig(config: any, serverName?: string): z.infer<typeof ServerConfigSchema> {
 		// Detect configuration issues before validation
 		const hasStdioFields = config.command !== undefined
-		const hasSseFields = config.url !== undefined
+		const hasUrlFields = config.url !== undefined // Covers sse and streamable-http
 
-		// Check for mixed fields
-		if (hasStdioFields && hasSseFields) {
+		// Check for mixed fields (stdio vs url-based)
+		if (hasStdioFields && hasUrlFields) {
 			throw new Error(mixedFieldsErrorMessage)
 		}
 
-		// Check if it's a stdio or SSE config and add type if missing
-		if (!config.type) {
-			if (hasStdioFields) {
-				config.type = "stdio"
-			} else if (hasSseFields) {
-				config.type = "sse"
-			} else {
-				throw new Error(missingFieldsErrorMessage)
-			}
-		} else if (config.type !== "stdio" && config.type !== "sse") {
+		// Infer type for stdio if not provided
+		if (!config.type && hasStdioFields) {
+			config.type = "stdio"
+		}
+
+		// For url-based configs, type must be provided by the user
+		if (hasUrlFields && !config.type) {
+			throw new Error("Configuration with 'url' must explicitly specify 'type' as 'sse' or 'streamable-http'.")
+		}
+
+		// Validate type if provided
+		if (config.type && !["stdio", "sse", "streamable-http"].includes(config.type)) {
 			throw new Error(typeErrorMessage)
 		}
 
 		// Check for type/field mismatch
 		if (config.type === "stdio" && !hasStdioFields) {
 			throw new Error(stdioFieldsErrorMessage)
 		}
-		if (config.type === "sse" && !hasSseFields) {
+		if (config.type === "sse" && !hasUrlFields) {
 			throw new Error(sseFieldsErrorMessage)
 		}
+		if (config.type === "streamable-http" && !hasUrlFields) {
+			throw new Error(streamableHttpFieldsErrorMessage)
+		}
+
+		// If neither command nor url is present (type alone is not enough)
+		if (!hasStdioFields && !hasUrlFields) {
+			throw new Error(missingFieldsErrorMessage)
+		}
 
 		// Validate the config against the schema
 		try {
@@ -441,7 +472,7 @@ export class McpHub {
 				},
 			)
 
-			let transport: StdioClientTransport | SSEClientTransport
+			let transport: StdioClientTransport | SSEClientTransport | StreamableHTTPClientTransport
 
 			// Inject environment variables to the config
 			const configInjected = (await injectEnv(config)) as typeof config
@@ -506,8 +537,33 @@ export class McpHub {
 				} else {
 					console.error(`No stderr stream for ${name}`)
 				}
-				transport.start = async () => {} // No-op now, .connect() won't fail
-			} else {
+			} else if (configInjected.type === "streamable-http") {
+				// Streamable HTTP connection
+				transport = new StreamableHTTPClientTransport(new URL(configInjected.url), {
+					requestInit: {
+						headers: configInjected.headers,
+					},
+				})
+
+				// Set up Streamable HTTP specific error handling
+				transport.onerror = async (error) => {
+					console.error(`Transport error for "${name}" (streamable-http):`, error)
+					const connection = this.findConnection(name, source)
+					if (connection) {
+						connection.server.status = "disconnected"
+						this.appendErrorMessage(connection, error instanceof Error ? error.message : `${error}`)
+					}
+					await this.notifyWebviewOfServerChanges()
+				}
+
+				transport.onclose = async () => {
+					const connection = this.findConnection(name, source)
+					if (connection) {
+						connection.server.status = "disconnected"
+					}
+					await this.notifyWebviewOfServerChanges()
+				}
+			} else if (configInjected.type === "sse") {
 				// SSE connection
 				const sseOptions = {
 					requestInit: {
@@ -542,7 +598,13 @@ export class McpHub {
 					}
 					await this.notifyWebviewOfServerChanges()
 				}
+			} else {
+				// Correctly placed "unsupported type" else block
+				// Should not happen if validateServerConfig is correct
+				throw new Error(`Unsupported MCP server type: ${(configInjected as any).type}`)
 			}
+			// transport.start assignment moved after all type-specific initializations
+			transport.start = async () => {} // No-op now, .connect() won't fail
 
 			const connection: McpConnection = {
 				server: {