feat(transport): Add new transport modes (Streamable HTTP & SSE)

package.json

@@ -27,6 +27,10 @@
     "dev": "concurrently \"tsc --watch\" \"tsc-alias --watch\"",
     "test:ts": "tsc --noEmit",
     "start": "node dist/index.js --env-file .env",
+    "start:http": "node dist/index.js --env-file .env --transport http",
+    "start:sse": "node dist/index.js --env-file .env --transport sse",
+    "start:all": "node dist/index.js --env-file .env --transport http,sse,stdio",
+    "help": "node dist/index.js -h",
     "lint": "eslint .",
     "format": "prettier --write \"**/*.+(js|ts|json)\"",
     "build": "tsc && tsc-alias",
@@ -57,9 +61,12 @@
     "@commander-js/extra-typings": "^13.1.0",
     "@confluentinc/kafka-javascript": "^1.3.0",
     "@confluentinc/schemaregistry": "^1.3.1",
+    "@fastify/swagger": "^9.5.1",
+    "@fastify/swagger-ui": "^5.2.2",
     "@modelcontextprotocol/sdk": "^1.11.0",
     "commander": "^13.1.0",
     "dotenv": "^16.5.0",
+    "fastify": "^5.3.2",
     "openapi-fetch": "^0.13.7",
     "pino": "^9.6.0",
     "zod": "^3.24.4"

src/cli.ts

@@ -1,12 +1,37 @@
-import { Command } from "@commander-js/extra-typings";
+import { Command, Option } from "@commander-js/extra-typings";
 import { logger } from "@src/logger.js";
+import { TransportType } from "@src/mcp/transports/types.js";
 import * as dotenv from "dotenv";
 import fs from "fs";
 import path from "path";
 
 // Define the interface for our CLI options
 export interface CLIOptions {
   envFile?: string;
+  transports: TransportType[];
+}
+
+function parseTransportList(value: string): TransportType[] {
+  // Split, trim, and filter out empty strings
+  const types = value
+    .split(",")
+    .map((type) => type.trim())
+    .filter(Boolean);
+
+  // Validate each transport type
+  const validTypes = new Set(Object.values(TransportType));
+  const invalidTypes = types.filter(
+    (type) => !validTypes.has(type as TransportType),
+  );
+
+  if (invalidTypes.length > 0) {
+    throw new Error(
+      `Invalid transport type(s): ${invalidTypes.join(", ")}. Valid options: ${Array.from(validTypes).join(", ")}`,
+    );
+  }
+
+  // Deduplicate using Set
+  return Array.from(new Set(types)) as TransportType[];
 }
 
 /**
@@ -21,17 +46,31 @@ export function parseCliArgs(): CLIOptions {
     )
     .version(process.env.npm_package_version ?? "dev")
     .option("-e, --env-file <path>", "Load environment variables from file")
+    .addOption(
+      new Option(
+        "-t, --transport <types>",
+        "Transport types (comma-separated list)",
+      )
+        .choices(Object.values(TransportType))
+        .default(TransportType.STDIO)
+        .argParser((value) => parseTransportList(value)),
+    )
     .action((options) => {
       if (options.envFile) {
-        loadEnvironmentVariables(options);
+        loadEnvironmentVariables(options.envFile);
       }
     })
     .allowExcessArguments(false)
     .exitOverride();
 
   try {
-    // Parse arguments and get options (no need for generic type parameter with extra-typings)
-    return program.parse().opts();
+    const opts = program.parse().opts();
+    return {
+      envFile: opts.envFile,
+      transports: Array.isArray(opts.transport)
+        ? opts.transport
+        : [opts.transport],
+    };
   } catch {
     // This block is reached when --help or --version is called
     // as these will throw an error due to exitOverride()
@@ -40,30 +79,28 @@ export function parseCliArgs(): CLIOptions {
 }
 
 /**
- * Load environment variables from file if specified in options
- * @param options CLI options containing envFile path
+ * Load environment variables from file
+ * @param envFile Path to the environment file
  */
-export function loadEnvironmentVariables(options: CLIOptions): void {
-  if (options.envFile) {
-    const envPath = path.resolve(options.envFile);
-
-    // Check if file exists
-    if (!fs.existsSync(envPath)) {
-      logger.error(`Environment file not found: ${envPath}`);
-      return;
-    }
+export function loadEnvironmentVariables(envFile: string): void {
+  const envPath = path.resolve(envFile);
 
-    // Load environment variables from file
-    const result = dotenv.config({ path: envPath });
+  // Check if file exists
+  if (!fs.existsSync(envPath)) {
+    logger.error(`Environment file not found: ${envPath}`);
+    process.exit(1);
+  }
 
-    if (result.error) {
-      logger.error(
-        { error: result.error },
-        "Error loading environment variables",
-      );
-      return;
-    }
+  // Load environment variables from file
+  const result = dotenv.config({ path: envPath });
 
-    logger.info(`Loaded environment variables from ${envPath}`);
+  if (result.error) {
+    logger.error(
+      { error: result.error },
+      "Error loading environment variables",
+    );
+    process.exit(1);
   }
+
+  logger.info(`Loaded environment variables from ${envPath}`);
 }

src/confluent/helpers.ts

@@ -5,17 +5,17 @@ import env, { Environment } from "@src/env.js";
  * @param envVarName - The name of the environment variable to check
  * @param errorMessage - The error message to throw if neither parameter nor environment variable exists
  * @param param - Optional parameter value to use instead of environment variable
- * @returns The parameter value or environment variable value
+ * @returns The parameter value or environment variable value with its original type
  * @throws {Error} When neither parameter nor environment variable exists
  */
-export const getEnsuredParam = (
+export const getEnsuredParam = <T extends Environment[keyof Environment]>(
   envVarName: keyof Environment,
   errorMessage: string,
-  param?: string,
-) => {
+  param?: T,
+): T => {
   const finalParam = param || env[envVarName];
   if (!finalParam) {
     throw new Error(`${errorMessage}`);
   }
-  return finalParam;
+  return finalParam as T;
 };