Claude-Code-Monitor/mcp/build/transports/http-server.js

/**
 * @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ĩ <vinnt@smartgift.vn>
 */
/* =============================================================================
 * 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.
 *
 * ## Observability
 * Prometheus scrapes `GET /api/metrics` (see `monitoring/`). Grafana ships four
 * provisioned boards (overview, sessions, tools, alerts). Native npm scripts and
 * Docker Compose profiles are documented in `monitoring/README.md`.
 *
 * ## 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 };
}