/** * @file http-server.ts * @description Implements the HTTP server transport for the MCP server, supporting both the newer Streamable HTTP protocol and the legacy SSE-based protocol. The server handles incoming requests, manages active sessions, and routes messages to the appropriate transport handlers. It also includes a health check endpoint and integrates with the MCP server instance to facilitate communication with connected clients. The module provides a shutdown function to gracefully close all active transports and the HTTP server itself. * @author Nguyễn Ngọc Trí Vĩ */ /* ============================================================================= * MODULE_GUIDE — extended in-file reference (comments only; safe to read, never executed) * ============================================================================= * **Purpose:** Dashboard module consumed by the React client, MCP tools, or desktop shell depending on deployment mode. * * ## Design constraints * - Local-first: no telemetry leaves the machine unless the user configures webhooks. * - Fail-safe hooks path on the server must never block Claude Code; UI mirrors that * philosophy by degrading gracefully (empty states, stale badges, reconnect loops). * - Destructive flows stay behind explicit confirmation modals and server-side gates. * - Internationalization: user-visible strings belong in i18n JSON, not literals here. * * ## Remote data & SSH * Remote Data Sources let operators aggregate multiple machines. SSH entries describe * how to reach a peer dashboard; the global data scope (`dataScope.ts`) narrows every * scoped GET via `?sources=`. Health checks and import history surface in Settings. * * ## Internal dependencies * - `../config/app-config.js` * - `../core/logger.js` * - `../ui/banner.js` * * ## Public surface * - `startHttpServer` — exported API; see TSDoc on the symbol for behavior. * * ## Testing pointers * - Prefer colocated `__tests__` with Vitest + Testing Library for UI. * - Server contract changes require `npm run test:server` and OpenAPI sync. * - MCP edits: `npm run mcp:typecheck` and `npm run mcp:build`. * * ## Related docs * - `ARCHITECTURE.md` — hooks → API → SQLite → WebSocket → UI pipeline. * - `docs/API.md` — REST reference. * - `.claude/skills/file-headers/` — mandatory `@author` header policy. * ============================================================================= */ /* ----------------------------------------------------------------------------- * EXPORT CATALOG — quick index of symbols defined below (documentation only). * ----------------------------------------------------------------------------- * **startHttpServer** * Part of this module's public contract. Downstream imports should treat * the signature and return type as stable unless release notes say otherwise. * When behavior changes, update the `@file` overview and relevant tests. * * ----------------------------------------------------------------------------- */ import { randomUUID } from "node:crypto"; import { createMcpExpressApp } from "@modelcontextprotocol/sdk/server/express.js"; import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js"; import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"; import { printBanner, printServerInfo, printReady, printShutdown } from "../ui/banner.js"; import * as c from "../ui/colors.js"; /** * Starts the HTTP transport, exposing the current Streamable HTTP protocol * (2025-11-25) and the legacy HTTP+SSE protocol (2024-11-05) side by side on * one Express app. Unlike stdio (one `McpServer` for the whole process), * **every new client session gets its own freshly-built `McpServer`** via * `buildServerFn`, isolated from other sessions but sharing the same * {@link AppConfig}/`DashboardApiClient`. * * Endpoints: * - `GET /health` — liveness/uptime/session-count probe for this MCP * process, distinct from `dashboard_health_check` (which checks the * dashboard itself). * - `ALL /mcp` — Streamable HTTP: a POST `initialize` with no * `mcp-session-id` starts a new session; later requests must carry that * header and route to the matching transport, rejected with a JSON-RPC * `-32000` error on a protocol mismatch. * - `GET /sse` — legacy SSE: a long-lived stream, one `SSEServerTransport` + * `McpServer` pair per connection. * - `POST /messages?sessionId=...` — legacy SSE's client-to-server companion * endpoint (SSE itself is server-to-client only). * * On successful bind, prints the banner/info panel/endpoint table to * stdout — this transport owns stdout, unlike stdio's protocol stream. * @returns The Express `app` and a `shutdown` closing every tracked * transport before the HTTP server itself. */ export async function startHttpServer(config, buildServerFn, logger, toolCount) { const app = createMcpExpressApp({ host: config.httpHost }); const transports = new Map(); // ── Health endpoint ─────────────────────────────────────────── app.get("/health", (_req, res) => { res.json({ status: "ok", server: config.serverName, version: config.serverVersion, transport: "http", uptime: process.uptime(), activeSessions: transports.size, }); }); // ── Streamable HTTP (protocol version 2025-11-25) ───────────── app.all("/mcp", async (req, res) => { const sessionId = req.headers["mcp-session-id"]; try { if (sessionId && transports.has(sessionId)) { const entry = transports.get(sessionId); if (entry.type !== "streamable") { res.status(400).json({ jsonrpc: "2.0", error: { code: -32000, message: "Session uses a different transport protocol" }, id: null, }); return; } await entry.transport.handleRequest(req, res, req.body); return; } if (req.method === "POST" && isInitializeRequest(req.body)) { logger.info("New Streamable HTTP session"); const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), }); transport.onclose = () => { const sid = transport.sessionId; if (sid) transports.delete(sid); logger.debug("Streamable HTTP session closed", { sessionId: sid }); }; const server = buildServerFn(); await server.connect(transport); const sid = transport.sessionId ?? randomUUID(); transports.set(sid, { transport, type: "streamable" }); await transport.handleRequest(req, res, req.body); return; } res.status(400).json({ jsonrpc: "2.0", error: { code: -32000, message: "Bad Request: No valid session or initialization" }, id: null, }); } catch (err) { logger.error("Streamable HTTP error", { error: err instanceof Error ? err.message : String(err), }); if (!res.headersSent) { res.status(500).json({ jsonrpc: "2.0", error: { code: -32603, message: "Internal server error" }, id: null, }); } } }); // ── Legacy SSE transport (protocol version 2024-11-05) ──────── app.get("/sse", async (_req, res) => { logger.info("New SSE session"); const transport = new SSEServerTransport("/messages", res); transports.set(transport.sessionId, { transport, type: "sse" }); res.on("close", () => { transports.delete(transport.sessionId); logger.debug("SSE session closed", { sessionId: transport.sessionId }); }); const server = buildServerFn(); await server.connect(transport); }); app.post("/messages", async (req, res) => { const sessionId = req.query.sessionId; if (!sessionId || !transports.has(sessionId)) { res.status(400).json({ jsonrpc: "2.0", error: { code: -32000, message: "No transport found for session" }, id: null, }); return; } const entry = transports.get(sessionId); if (entry.type !== "sse") { res.status(400).json({ jsonrpc: "2.0", error: { code: -32000, message: "Session uses a different transport protocol" }, id: null, }); return; } await entry.transport.handlePostMessage(req, res, req.body); }); // ── Start listening ─────────────────────────────────────────── printBanner(); printServerInfo({ transport: "http (sse + streamable)", version: config.serverVersion, dashboard: config.dashboardBaseUrl.toString(), port: config.httpPort, mutations: config.allowMutations, destructive: config.allowDestructive, tools: toolCount, }); const httpServer = await new Promise((resolve, reject) => { const srv = app.listen(config.httpPort, config.httpHost, () => resolve(srv)); srv.on("error", reject); }); const endpoints = [ ["Streamable HTTP", `http://${config.httpHost}:${config.httpPort}/mcp`, "POST/GET/DELETE"], ["Legacy SSE", `http://${config.httpHost}:${config.httpPort}/sse`, "GET"], ["Legacy Messages", `http://${config.httpHost}:${config.httpPort}/messages`, "POST"], ["Health", `http://${config.httpHost}:${config.httpPort}/health`, "GET"], ]; process.stdout.write(` ${c.bold(c.brightCyan("◆"))} ${c.bold(c.brightWhite("Endpoints"))}\n`); for (const [name, url, methods] of endpoints) { process.stdout.write(` ${c.dim(c.cyan("→"))} ${c.label(name.padEnd(20))} ${c.green(url)} ${c.muted(`[${methods}]`)}\n`); } process.stdout.write("\n"); printReady("http"); // ── Shutdown ────────────────────────────────────────────────── const shutdown = async () => { printShutdown(); const closePromises = []; for (const [sid, entry] of transports) { logger.debug("Closing transport", { sessionId: sid }); closePromises.push(entry.transport.close?.().catch((err) => { logger.error("Error closing transport", { sessionId: sid, error: err instanceof Error ? err.message : String(err), }); }) ?? Promise.resolve()); } await Promise.allSettled(closePromises); transports.clear(); await new Promise((resolve) => { httpServer.close(() => resolve()); }); logger.info("HTTP server stopped"); }; return { app, shutdown }; }