diff --git a/open-sse/mcp-server/httpTransport.ts b/open-sse/mcp-server/httpTransport.ts new file mode 100644 index 0000000000..cd467acdda --- /dev/null +++ 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 { + 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 { + 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; +} diff --git a/open-sse/mcp-server/index.ts b/open-sse/mcp-server/index.ts index a497cfc3c4..cd9670cb2c 100644 --- 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"; + diff --git a/src/app/(dashboard)/dashboard/endpoint/page.tsx b/src/app/(dashboard)/dashboard/endpoint/page.tsx index 7d311cdf87..898f76bd1b 100644 --- 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 (
- {/* Status indicator */}
- {/* Toggle switch */} + ))} +
+ + {/* Connection info */} +
+ + {value === "stdio" ? "terminal" : "link"} + + + {urlMap[value]} + + {value !== "stdio" && ( + + )} +
+
+ ); +} + +/* ────── 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("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) => { + 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() { )} + {/* Transport selector for MCP */} + {activeTab === "mcp" && mcpEnabled && ( + void changeTransport(t)} + disabled={transportSaving} + baseUrl={baseUrl} + /> + )} + {activeTab === "endpoint-proxy" && } {activeTab === "mcp" && } {activeTab === "a2a" && } diff --git a/src/app/api/mcp/sse/route.ts b/src/app/api/mcp/sse/route.ts new file mode 100644 index 0000000000..a4cff28923 --- /dev/null +++ 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 { + 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); +} diff --git a/src/app/api/mcp/status/route.ts b/src/app/api/mcp/status/route.ts index d6bb5f59c6..3ca52107cd 100644 --- 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, diff --git a/src/app/api/mcp/stream/route.ts b/src/app/api/mcp/stream/route.ts new file mode 100644 index 0000000000..2a5bb57620 --- /dev/null +++ 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 { + 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); +} diff --git a/src/shared/validation/settingsSchemas.ts b/src/shared/validation/settingsSchemas.ts index 0e1085a3c1..6c6cc78e4a 100644 --- 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(), });