feat: add MCP multi-transport (stdio + SSE + Streamable HTTP)

- Created httpTransport.ts with singleton MCP server and WebStandard Streamable HTTP transport running inside Next.js process - Added /api/mcp/sse route (GET+POST) for SSE transport - Added /api/mcp/stream route (GET+POST+DELETE) for Streamable HTTP - Added mcpTransport enum to settingsSchemas (stdio|sse|streamable-http) - Updated /api/mcp/status to report HTTP transport state - Added TransportSelector UI with mode buttons and connection URL display - Routes guard against disabled MCP or wrong transport mode
2026-07-26 09:52:11 +03:00 · 2026-03-05 17:45:44 -03:00
parent 3de8b4371a
commit e09d4a02a2
7 changed files with 408 additions and 28 deletions
--- a/open-sse/mcp-server/httpTransport.ts
+++ b/open-sse/mcp-server/httpTransport.ts
@@ -0,0 +1,120 @@
+/**
+ * MCP HTTP Transport Layer — Singleton server + SSE/Streamable HTTP handlers.
+ *
+ * Runs the MCP server **inside** the Next.js process so it can be toggled
+ * from the dashboard without requiring `omniroute --mcp`.
+ *
+ * Transport modes:
+ *   - SSE:             GET /api/mcp/sse (event stream)  +  POST /api/mcp/sse (messages)
+ *   - Streamable HTTP: POST /api/mcp/stream (messages)  +  GET /api/mcp/stream (SSE stream)  +  DELETE /api/mcp/stream (session end)
+ */
+
+import { randomUUID } from "node:crypto";
+import { createMcpServer } from "./server.ts";
+import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js";
+import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
+
+// ────── Singleton ──────────────────────────────────────────
+
+let _server: McpServer | null = null;
+let _transport: WebStandardStreamableHTTPServerTransport | null = null;
+let _startedAt: number | null = null;
+let _activeTransportMode: "sse" | "streamable-http" | null = null;
+
+function ensureServer(mode: "sse" | "streamable-http"): {
+  server: McpServer;
+  transport: WebStandardStreamableHTTPServerTransport;
+} {
+  if (_server && _transport && _activeTransportMode === mode) {
+    return { server: _server, transport: _transport };
+  }
+
+  // Shutdown previous if switching modes
+  if (_transport) {
+    try { _transport.close(); } catch { /* ignore */ }
+  }
+
+  _server = createMcpServer();
+  _transport = new WebStandardStreamableHTTPServerTransport({
+    sessionIdGenerator: () => randomUUID(),
+  });
+  _activeTransportMode = mode;
+  _startedAt = Date.now();
+
+  // Connect server to transport (fire-and-forget, will be ready by first request)
+  void _server.connect(_transport);
+
+  console.log(`[MCP] HTTP transport started (${mode})`);
+  return { server: _server, transport: _transport };
+}
+
+// ────── Streamable HTTP Handler ────────────────────────────
+
+/**
+ * Handle Streamable HTTP requests (POST / GET / DELETE).
+ * Used by the Next.js route at /api/mcp/stream.
+ */
+export async function handleMcpStreamableHTTP(request: Request): Promise<Response> {
+  const { transport } = ensureServer("streamable-http");
+
+  try {
+    return await transport.handleRequest(request);
+  } catch (err) {
+    console.error("[MCP] Streamable HTTP error:", err);
+    return new Response(
+      JSON.stringify({ error: "MCP transport error" }),
+      { status: 500, headers: { "Content-Type": "application/json" } },
+    );
+  }
+}
+
+/**
+ * Handle SSE requests.
+ * SSE transport is implemented via Streamable HTTP transport with GET for SSE stream
+ * and POST for messages (the Streamable HTTP transport supports both patterns).
+ */
+export async function handleMcpSSE(request: Request): Promise<Response> {
+  const { transport } = ensureServer("sse");
+
+  try {
+    return await transport.handleRequest(request);
+  } catch (err) {
+    console.error("[MCP] SSE error:", err);
+    return new Response(
+      JSON.stringify({ error: "MCP SSE transport error" }),
+      { status: 500, headers: { "Content-Type": "application/json" } },
+    );
+  }
+}
+
+// ────── Status & Lifecycle ─────────────────────────────────
+
+export function getMcpHttpStatus(): {
+  online: boolean;
+  transport: string | null;
+  startedAt: number | null;
+  uptime: string | null;
+} {
+  const online = _transport !== null && _activeTransportMode !== null;
+  return {
+    online,
+    transport: _activeTransportMode,
+    startedAt: _startedAt,
+    uptime: _startedAt ? `${Math.floor((Date.now() - _startedAt) / 1000)}s` : null,
+  };
+}
+
+export function shutdownMcpHttp(): void {
+  if (_transport) {
+    try { _transport.close(); } catch { /* ignore */ }
+  }
+  _server = null;
+  _transport = null;
+  _activeTransportMode = null;
+  _startedAt = null;
+  console.log("[MCP] HTTP transport shutdown");
+}
+
+export function isMcpHttpActive(): boolean {
+  return _transport !== null;
+}
--- a/open-sse/mcp-server/index.ts
+++ b/open-sse/mcp-server/index.ts
@@ -9,4 +9,12 @@ export {
  isMcpHeartbeatOnline,
  isProcessAlive,
 } from "./runtimeHeartbeat.ts";
+export {
+  handleMcpSSE,
+  handleMcpStreamableHTTP,
+  getMcpHttpStatus,
+  shutdownMcpHttp,
+  isMcpHttpActive,
+} from "./httpTransport.ts";
 export * from "./schemas/index.ts";
+
--- a/src/app/(dashboard)/dashboard/endpoint/page.tsx
+++ b/src/app/(dashboard)/dashboard/endpoint/page.tsx
@@ -2,7 +2,6 @@

 import { useState, useEffect, useCallback } from "react";
 import { SegmentedControl } from "@/shared/components";
-import { getMachineId } from "@/shared/utils/machine";
 import EndpointPageClient from "./EndpointPageClient";
 import McpDashboardPage from "../mcp/page";
 import A2ADashboardPage from "../a2a/page";
@@ -14,6 +13,9 @@ type ServiceStatus = {
  loading: boolean;
 };

+type McpTransport = "stdio" | "sse" | "streamable-http";
+
+/* ────── Toggle Switch ────── */
 function ServiceToggle({
  label,
  status,
@@ -29,7 +31,6 @@ function ServiceToggle({
 }) {
  return (
    <div className="flex items-center gap-3 ml-auto">
-      {/* Status indicator */}
      <div
        className="flex items-center gap-1.5 px-2.5 py-1 rounded-full text-xs font-medium border"
        style={{
@@ -64,18 +65,13 @@ function ServiceToggle({
        {status.loading ? "..." : status.online ? "Online" : "Offline"}
      </div>

-      {/* Toggle switch */}
      <button
        onClick={onToggle}
        disabled={toggling}
-        className="relative inline-flex items-center h-7 w-[52px] rounded-full transition-all duration-300 focus:outline-none focus:ring-2 focus:ring-offset-2 border"
+        className="relative inline-flex items-center h-7 w-[52px] rounded-full transition-all duration-300 focus:outline-none border"
        style={{
-          background: enabled
-            ? "rgb(34,197,94)"
-            : "var(--color-bg-tertiary)",
-          borderColor: enabled
-            ? "rgba(34,197,94,0.5)"
-            : "var(--color-border)",
+          background: enabled ? "rgb(34,197,94)" : "var(--color-bg-tertiary)",
+          borderColor: enabled ? "rgba(34,197,94,0.5)" : "var(--color-border)",
          opacity: toggling ? 0.6 : 1,
          cursor: toggling ? "wait" : "pointer",
        }}
@@ -92,9 +88,7 @@ function ServiceToggle({

      <span
        className="text-xs font-medium min-w-[24px]"
-        style={{
-          color: enabled ? "rgb(34,197,94)" : "var(--color-text-muted)",
-        }}
+        style={{ color: enabled ? "rgb(34,197,94)" : "var(--color-text-muted)" }}
      >
        {toggling ? "..." : enabled ? "ON" : "OFF"}
      </span>
@@ -102,6 +96,119 @@ function ServiceToggle({
  );
 }

+/* ────── Transport Selector ────── */
+function TransportSelector({
+  value,
+  onChange,
+  disabled,
+  baseUrl,
+}: {
+  value: McpTransport;
+  onChange: (t: McpTransport) => void;
+  disabled: boolean;
+  baseUrl: string;
+}) {
+  const options: { value: McpTransport; label: string; desc: string }[] = [
+    { value: "stdio", label: "stdio", desc: "Local — IDE spawns process via omniroute --mcp" },
+    { value: "sse", label: "SSE", desc: "Remote — Server-Sent Events over HTTP" },
+    { value: "streamable-http", label: "Streamable HTTP", desc: "Remote — Modern bidirectional HTTP" },
+  ];
+
+  const urlMap: Record<McpTransport, string> = {
+    stdio: "omniroute --mcp",
+    sse: `${baseUrl}/api/mcp/sse`,
+    "streamable-http": `${baseUrl}/api/mcp/stream`,
+  };
+
+  return (
+    <div
+      className="rounded-lg border p-4 mt-3"
+      style={{ borderColor: "var(--color-border)", background: "var(--color-bg-secondary)" }}
+    >
+      <div className="flex items-center gap-2 mb-3">
+        <span
+          className="material-symbols-rounded text-base"
+          style={{ color: "var(--color-primary)" }}
+        >
+          swap_horiz
+        </span>
+        <span className="text-sm font-medium" style={{ color: "var(--color-text)" }}>
+          Transport Mode
+        </span>
+      </div>
+
+      <div className="flex gap-2 flex-wrap">
+        {options.map((opt) => (
+          <button
+            key={opt.value}
+            onClick={() => onChange(opt.value)}
+            disabled={disabled}
+            className="flex flex-col items-start px-4 py-2.5 rounded-lg border transition-all duration-200 text-left"
+            style={{
+              borderColor:
+                value === opt.value ? "var(--color-primary)" : "var(--color-border)",
+              background:
+                value === opt.value
+                  ? "rgba(var(--color-primary-rgb, 99,102,241), 0.1)"
+                  : "transparent",
+              opacity: disabled ? 0.5 : 1,
+              cursor: disabled ? "wait" : "pointer",
+            }}
+          >
+            <span
+              className="text-sm font-semibold"
+              style={{
+                color: value === opt.value ? "var(--color-primary)" : "var(--color-text)",
+              }}
+            >
+              {opt.label}
+            </span>
+            <span
+              className="text-xs mt-0.5"
+              style={{ color: "var(--color-text-muted)" }}
+            >
+              {opt.desc}
+            </span>
+          </button>
+        ))}
+      </div>
+
+      {/* Connection info */}
+      <div
+        className="mt-3 rounded-md px-3 py-2 flex items-center gap-2"
+        style={{ background: "var(--color-bg-tertiary)" }}
+      >
+        <span
+          className="material-symbols-rounded text-sm"
+          style={{ color: "var(--color-text-muted)" }}
+        >
+          {value === "stdio" ? "terminal" : "link"}
+        </span>
+        <code
+          className="text-xs break-all"
+          style={{ color: "var(--color-text-muted)" }}
+        >
+          {urlMap[value]}
+        </code>
+        {value !== "stdio" && (
+          <button
+            className="ml-auto text-xs px-2 py-0.5 rounded border hover:opacity-80 transition-opacity"
+            style={{
+              borderColor: "var(--color-border)",
+              color: "var(--color-text-muted)",
+            }}
+            onClick={() => void navigator.clipboard.writeText(urlMap[value])}
+            title="Copy URL"
+          >
+            Copy
+          </button>
+        )}
+      </div>
+    </div>
+  );
+}
+
+/* ────── Main Page ────── */
 export default function EndpointPage() {
  const [activeTab, setActiveTab] = useState("endpoint-proxy");
  const t = useTranslations("endpoints");
@@ -112,8 +219,19 @@ export default function EndpointPage() {
  const [a2aEnabled, setA2aEnabled] = useState(false);
  const [mcpToggling, setMcpToggling] = useState(false);
  const [a2aToggling, setA2aToggling] = useState(false);
+  const [mcpTransport, setMcpTransport] = useState<McpTransport>("stdio");
+  const [transportSaving, setTransportSaving] = useState(false);

-  // Fetch initial enabled state from settings
+  const [baseUrl, setBaseUrl] = useState("");
+
+  // Detect base URL from browser
+  useEffect(() => {
+    if (typeof window !== "undefined") {
+      setBaseUrl(`${window.location.protocol}//${window.location.host}`);
+    }
+  }, []);
+
+  // Fetch initial settings
  useEffect(() => {
    const fetchSettings = async () => {
      try {
@@ -122,14 +240,23 @@ export default function EndpointPage() {
          const data = await res.json();
          setMcpEnabled(!!data.mcpEnabled);
          setA2aEnabled(!!data.a2aEnabled);
+          setMcpTransport((data.mcpTransport as McpTransport) || "stdio");
        }
      } catch {
-        // defaults stay false
+        // defaults stay
      }
    };
    void fetchSettings();
  }, []);

+  const patchSetting = useCallback(async (body: Record<string, unknown>) => {
+    return fetch("/api/settings", {
+      method: "PATCH",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify(body),
+    });
+  }, []);
+
  const toggleService = useCallback(
    async (service: "mcp" | "a2a") => {
      const setToggling = service === "mcp" ? setMcpToggling : setA2aToggling;
@@ -139,23 +266,32 @@ export default function EndpointPage() {

      setToggling(true);
      try {
-        const res = await fetch("/api/settings", {
-          method: "PATCH",
-          headers: { "Content-Type": "application/json" },
-          body: JSON.stringify({
-            [service === "mcp" ? "mcpEnabled" : "a2aEnabled"]: newValue,
-          }),
+        const res = await patchSetting({
+          [service === "mcp" ? "mcpEnabled" : "a2aEnabled"]: newValue,
        });
-        if (res.ok) {
-          setEnabled(newValue);
-        }
+        if (res.ok) setEnabled(newValue);
      } catch {
-        // toggle failed — keep current state
+        // keep current state
      } finally {
        setToggling(false);
      }
    },
-    [mcpEnabled, a2aEnabled],
+    [mcpEnabled, a2aEnabled, patchSetting],
+  );
+
+  const changeTransport = useCallback(
+    async (newTransport: McpTransport) => {
+      setTransportSaving(true);
+      try {
+        const res = await patchSetting({ mcpTransport: newTransport });
+        if (res.ok) setMcpTransport(newTransport);
+      } catch {
+        // keep current
+      } finally {
+        setTransportSaving(false);
+      }
+    },
+    [patchSetting],
  );

  const refreshMcpStatus = useCallback(async () => {
@@ -232,6 +368,16 @@ export default function EndpointPage() {
        )}
      </div>

+      {/* Transport selector for MCP */}
+      {activeTab === "mcp" && mcpEnabled && (
+        <TransportSelector
+          value={mcpTransport}
+          onChange={(t) => void changeTransport(t)}
+          disabled={transportSaving}
+          baseUrl={baseUrl}
+        />
+      )}
+
      {activeTab === "endpoint-proxy" && <EndpointPageClient machineId="" />}
      {activeTab === "mcp" && <McpDashboardPage />}
      {activeTab === "a2a" && <A2ADashboardPage />}
--- a/src/app/api/mcp/sse/route.ts
+++ b/src/app/api/mcp/sse/route.ts
@@ -0,0 +1,41 @@
+/**
+ * MCP SSE Transport — /api/mcp/sse
+ *
+ * Endpoints:
+ *   GET    — open SSE stream for bidirectional communication
+ *   POST   — send JSON-RPC messages to the MCP server
+ */
+
+import { NextRequest, NextResponse } from "next/server";
+import { getSettings } from "@/lib/db/settings";
+import { handleMcpSSE } from "../../../../../open-sse/mcp-server/httpTransport";
+
+async function guardEnabled(): Promise<NextResponse | null> {
+  const settings = await getSettings();
+  if (!settings.mcpEnabled) {
+    return NextResponse.json(
+      { error: "MCP server is disabled. Enable it from the Endpoints page." },
+      { status: 503 },
+    );
+  }
+  const transport = (settings.mcpTransport as string) || "stdio";
+  if (transport !== "sse") {
+    return NextResponse.json(
+      { error: `MCP transport is set to "${transport}", not "sse". Change it from Settings.` },
+      { status: 400 },
+    );
+  }
+  return null;
+}
+
+export async function GET(request: NextRequest) {
+  const blocked = await guardEnabled();
+  if (blocked) return blocked;
+  return handleMcpSSE(request);
+}
+
+export async function POST(request: NextRequest) {
+  const blocked = await guardEnabled();
+  if (blocked) return blocked;
+  return handleMcpSSE(request);
+}
--- a/src/app/api/mcp/status/route.ts
+++ b/src/app/api/mcp/status/route.ts
@@ -6,16 +6,29 @@ import {
  readMcpHeartbeat,
  resolveMcpHeartbeatPath,
 } from "@omniroute/open-sse/mcp-server/runtimeHeartbeat";
+import { getMcpHttpStatus } from "../../../../../open-sse/mcp-server/httpTransport";
+import { getSettings } from "@/lib/db/settings";

 export async function GET() {
  try {
-    const [heartbeat, stats, lastCallPage] = await Promise.all([
+    const [heartbeat, stats, lastCallPage, settings] = await Promise.all([
      readMcpHeartbeat(),
      getAuditStats(),
      queryAuditEntries({ limit: 1, offset: 0 }),
+      getSettings(),
    ]);

-    const online = isMcpHeartbeatOnline(heartbeat, { requireLivePid: true });
+    const mcpEnabled = !!settings.mcpEnabled;
+    const mcpTransport = (settings.mcpTransport as string) || "stdio";
+
+    // Check HTTP transport (SSE / Streamable HTTP) if active
+    const httpStatus = getMcpHttpStatus();
+
+    // stdio uses heartbeat file; HTTP transports use in-process state
+    const stdioOnline = isMcpHeartbeatOnline(heartbeat, { requireLivePid: true });
+    const online =
+      mcpTransport === "stdio" ? stdioOnline : httpStatus.online;
+
    const lastCall = lastCallPage.entries[0] || null;
    const now = Date.now();
    const lastHeartbeatAtMs = heartbeat ? new Date(heartbeat.lastHeartbeatAt).getTime() : null;
@@ -32,6 +45,8 @@ export async function GET() {
    return NextResponse.json({
      status: online ? "online" : "offline",
      online,
+      enabled: mcpEnabled,
+      transport: mcpTransport,
      heartbeatPath: resolveMcpHeartbeatPath(),
      heartbeat: heartbeat
        ? {
@@ -41,6 +56,7 @@ export async function GET() {
            uptimeMs,
          }
        : null,
+      httpTransport: httpStatus,
      activity: {
        totalCalls24h: stats.totalCalls,
        successRate: stats.successRate,
--- a/src/app/api/mcp/stream/route.ts
+++ b/src/app/api/mcp/stream/route.ts
@@ -0,0 +1,48 @@
+/**
+ * MCP Streamable HTTP Transport — /api/mcp/stream
+ *
+ * Endpoints:
+ *   POST   — send JSON-RPC messages to the MCP server
+ *   GET    — open SSE stream for server-initiated messages
+ *   DELETE — end session
+ */
+
+import { NextRequest, NextResponse } from "next/server";
+import { getSettings } from "@/lib/db/settings";
+import { handleMcpStreamableHTTP } from "../../../../../open-sse/mcp-server/httpTransport";
+
+async function guardEnabled(): Promise<NextResponse | null> {
+  const settings = await getSettings();
+  if (!settings.mcpEnabled) {
+    return NextResponse.json(
+      { error: "MCP server is disabled. Enable it from the Endpoints page." },
+      { status: 503 },
+    );
+  }
+  const transport = (settings.mcpTransport as string) || "stdio";
+  if (transport !== "streamable-http") {
+    return NextResponse.json(
+      { error: `MCP transport is set to "${transport}", not "streamable-http". Change it from Settings.` },
+      { status: 400 },
+    );
+  }
+  return null;
+}
+
+export async function POST(request: NextRequest) {
+  const blocked = await guardEnabled();
+  if (blocked) return blocked;
+  return handleMcpStreamableHTTP(request);
+}
+
+export async function GET(request: NextRequest) {
+  const blocked = await guardEnabled();
+  if (blocked) return blocked;
+  return handleMcpStreamableHTTP(request);
+}
+
+export async function DELETE(request: NextRequest) {
+  const blocked = await guardEnabled();
+  if (blocked) return blocked;
+  return handleMcpStreamableHTTP(request);
+}
--- a/src/shared/validation/settingsSchemas.ts
+++ b/src/shared/validation/settingsSchemas.ts
@@ -32,5 +32,6 @@ export const updateSettingsSchema = z.object({
  stickyRoundRobinLimit: z.number().int().min(0).max(1000).optional(),
  // Protocol toggles (default: disabled)
  mcpEnabled: z.boolean().optional(),
+  mcpTransport: z.enum(["stdio", "sse", "streamable-http"]).optional(),
  a2aEnabled: z.boolean().optional(),
 });