diff --git a/src/http-server.ts b/src/http-server.ts new file mode 100644 index 0000000..46f1fe8 --- /dev/null +++ b/src/http-server.ts @@ -0,0 +1,411 @@ +/** + * Hosted Remote MCP Server implementation over HTTP/SSE. + * + * Provides a multi-tenant HTTP/SSE MCP server using native Bun.serve. + * Incoming SSE connections supply their personal OpenProject API keys + * via Authorization: Bearer, X-OpenProject-Api-Key, or ?apiKey= query parameter. + * Each connection runs in an isolated session with its own OpenProjectClient + * and McpServer instance. + */ + +import type { Server } from "bun"; +import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; +import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; +import { + JSONRPCMessageSchema, + type JSONRPCMessage, + type MessageExtraInfo, +} from "@modelcontextprotocol/sdk/types.js"; +import type { AppConfig } from "./config/index.ts"; +import { OpenProjectClient } from "./client/api-client.ts"; +import { runWithContext } from "./context.ts"; +import { registerAllTools } from "./tools/index.ts"; +import { formatToolError } from "./tools/common.ts"; +import { OpenProjectError } from "./client/errors.ts"; +import { SERVER_NAME, SERVER_VERSION } from "./server.ts"; + +export interface HttpServerInstance { + server: Server; + port: number; + stop(): Promise; + getActiveSessionsCount(): number; +} + +/** + * Extracts OpenProject API key from incoming request according to precedence: + * 1. Authorization: Bearer + * 2. X-OpenProject-Api-Key: + * 3. ?apiKey= query parameter + */ +export function extractApiKey(req: Request, url: URL): string | undefined { + const authHeader = req.headers.get("authorization"); + if (authHeader) { + const match = authHeader.match(/^Bearer\s+(.+)$/i); + if (match && match[1] && match[1].trim().length > 0) { + return match[1].trim(); + } + } + + const headerKey = req.headers.get("x-openproject-api-key"); + if (headerKey && headerKey.trim().length > 0) { + return headerKey.trim(); + } + + const queryKey = url.searchParams.get("apiKey"); + if (queryKey && queryKey.trim().length > 0) { + return queryKey.trim(); + } + + return undefined; +} + +/** + * Custom SSE Transport for native Bun Web Streams. + */ +export class BunSseTransport implements Transport { + readonly sessionId: string; + private controller: ReadableStreamDefaultController | null = null; + private closed = false; + private readonly encoder = new TextEncoder(); + + onclose?: () => void; + onerror?: (error: Error) => void; + onmessage?: (message: T, extra?: MessageExtraInfo) => void; + + constructor(sessionId: string) { + this.sessionId = sessionId; + } + + setController(controller: ReadableStreamDefaultController): void { + this.controller = controller; + } + + async start(): Promise { + if (!this.controller) { + throw new Error("SSE stream controller not initialized"); + } + const endpoint = `/messages?sessionId=${this.sessionId}`; + this.sendRawEvent("endpoint", endpoint); + } + + private sendRawEvent(event: string, data: string): void { + if (this.closed || !this.controller) { + return; + } + try { + this.controller.enqueue(this.encoder.encode(`event: ${event}\ndata: ${data}\n\n`)); + } catch (err) { + const error = err instanceof Error ? err : new Error(String(err)); + this.onerror?.(error); + } + } + + async send(message: JSONRPCMessage): Promise { + if (this.closed) { + throw new Error("Transport is closed"); + } + this.sendRawEvent("message", JSON.stringify(message)); + } + + async close(): Promise { + if (this.closed) return; + this.closed = true; + try { + this.controller?.close(); + } catch { + // Ignore if already closed or cancelled + } + this.controller = null; + this.onclose?.(); + } + + handleMessage(message: JSONRPCMessage, extra?: MessageExtraInfo): void { + if (this.closed) { + throw new Error("Session transport is closed"); + } + this.onmessage?.(message, extra); + } +} + +interface SessionRecord { + sessionId: string; + transport: BunSseTransport; + server: McpServer; + client: OpenProjectClient; + cleanup: () => Promise; +} + +const CORS_HEADERS = { + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "GET, POST, OPTIONS", + "Access-Control-Allow-Headers": "Content-Type, Authorization, X-OpenProject-Api-Key", +}; + +/** + * Starts the hosted HTTP/SSE MCP server with Bun.serve. + */ +export async function startHttpServer(config: AppConfig): Promise { + const activeSessions = new Map(); + + const server = Bun.serve({ + port: config.port ?? 3000, + hostname: config.host || "0.0.0.0", + async fetch(req: Request): Promise { + const url = new URL(req.url); + + // Handle CORS preflight + if (req.method === "OPTIONS") { + return new Response(null, { + status: 204, + headers: CORS_HEADERS, + }); + } + + // 1. Health check endpoint + if (url.pathname === "/health") { + if (req.method !== "GET") { + return new Response(JSON.stringify({ error: "Method not allowed" }), { + status: 405, + headers: { "Content-Type": "application/json", ...CORS_HEADERS }, + }); + } + return new Response( + JSON.stringify({ + status: "ok", + mode: "remote-mcp", + openproject: config.baseUrl, + readOnly: config.readOnly, + }), + { + status: 200, + headers: { "Content-Type": "application/json", ...CORS_HEADERS }, + } + ); + } + + // 2. SSE connection initiation endpoint + if (url.pathname === "/sse") { + if (req.method !== "GET") { + return new Response(JSON.stringify({ error: "Method not allowed" }), { + status: 405, + headers: { "Content-Type": "application/json", ...CORS_HEADERS }, + }); + } + + const apiKey = extractApiKey(req, url); + if (!apiKey) { + return new Response( + JSON.stringify({ + error: + "Missing OpenProject API key. Provide via Authorization header, X-OpenProject-Api-Key header, or ?apiKey= query parameter.", + }), + { + status: 401, + headers: { "Content-Type": "application/json", ...CORS_HEADERS }, + } + ); + } + + const sessionId = crypto.randomUUID(); + const transport = new BunSseTransport(sessionId); + + const sessionClient = new OpenProjectClient({ + baseUrl: config.baseUrl, + apiKey, + }); + + const sessionServer = new McpServer({ + name: SERVER_NAME, + version: SERVER_VERSION, + }); + + registerAllTools(sessionServer, { + readOnly: config.readOnly, + wrapExecute: async (fn, tool) => { + if (config.readOnly && !tool.readOnly) { + return formatToolError( + new OpenProjectError( + "Operation rejected. OpenProject MCP server is running in read-only mode.", + { code: "SERVER_READ_ONLY" } + ) + ); + } + return runWithContext( + { client: sessionClient, isReadOnly: config.readOnly }, + fn + ); + }, + }); + + let cleanedUp = false; + const cleanup = async () => { + if (cleanedUp) return; + cleanedUp = true; + activeSessions.delete(sessionId); + try { + await transport.close(); + } catch {} + try { + await sessionServer.close(); + } catch {} + }; + + const sessionRecord: SessionRecord = { + sessionId, + transport, + server: sessionServer, + client: sessionClient, + cleanup, + }; + activeSessions.set(sessionId, sessionRecord); + + if (req.signal.aborted) { + cleanup().catch(() => {}); + } else { + req.signal.addEventListener("abort", () => { + cleanup().catch(() => {}); + }); + } + + const stream = new ReadableStream({ + async start(controller) { + transport.setController(controller); + try { + await sessionServer.connect(transport); + } catch (err) { + await cleanup(); + try { + controller.error(err); + } catch {} + } + }, + cancel() { + cleanup().catch(() => {}); + }, + }); + + return new Response(stream, { + status: 200, + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache, no-transform", + "Connection": "keep-alive", + ...CORS_HEADERS, + }, + }); + } + + // 3. JSON-RPC message posting endpoint + if (url.pathname === "/messages") { + if (req.method !== "POST") { + return new Response(JSON.stringify({ error: "Method not allowed" }), { + status: 405, + headers: { "Content-Type": "application/json", ...CORS_HEADERS }, + }); + } + + const sessionId = url.searchParams.get("sessionId"); + if (!sessionId) { + return new Response( + JSON.stringify({ error: "Missing sessionId query parameter" }), + { + status: 400, + headers: { "Content-Type": "application/json", ...CORS_HEADERS }, + } + ); + } + + const session = activeSessions.get(sessionId); + if (!session) { + return new Response(JSON.stringify({ error: "Session not found" }), { + status: 404, + headers: { "Content-Type": "application/json", ...CORS_HEADERS }, + }); + } + + let body: unknown; + try { + body = await req.json(); + } catch { + return new Response(JSON.stringify({ error: "Invalid JSON" }), { + status: 400, + headers: { "Content-Type": "application/json", ...CORS_HEADERS }, + }); + } + + if (Array.isArray(body)) { + for (const item of body) { + const parseResult = JSONRPCMessageSchema.safeParse(item); + if (!parseResult.success) { + return new Response( + JSON.stringify({ + error: "Invalid JSON-RPC message in batch", + details: parseResult.error.format(), + }), + { + status: 400, + headers: { "Content-Type": "application/json", ...CORS_HEADERS }, + } + ); + } + } + await runWithContext( + { client: session.client, isReadOnly: config.readOnly }, + async () => { + for (const item of body) { + session.transport.handleMessage(item as JSONRPCMessage); + } + } + ); + } else { + const parseResult = JSONRPCMessageSchema.safeParse(body); + if (!parseResult.success) { + return new Response( + JSON.stringify({ + error: "Invalid JSON-RPC message", + details: parseResult.error.format(), + }), + { + status: 400, + headers: { "Content-Type": "application/json", ...CORS_HEADERS }, + } + ); + } + + await runWithContext( + { client: session.client, isReadOnly: config.readOnly }, + async () => { + session.transport.handleMessage(parseResult.data); + } + ); + } + + return new Response("Accepted", { + status: 202, + headers: { "Content-Type": "text/plain", ...CORS_HEADERS }, + }); + } + + // 4. Default Not Found + return new Response(JSON.stringify({ error: "Not Found" }), { + status: 404, + headers: { "Content-Type": "application/json", ...CORS_HEADERS }, + }); + }, + }); + + return { + server, + port: server.port ?? (config.port ?? 3000), + async stop() { + for (const session of activeSessions.values()) { + await session.cleanup(); + } + activeSessions.clear(); + server.stop(true); + }, + getActiveSessionsCount() { + return activeSessions.size; + }, + }; +} diff --git a/src/server.ts b/src/server.ts index 65c9912..6d92d81 100644 --- a/src/server.ts +++ b/src/server.ts @@ -32,7 +32,7 @@ export function createServer(config: AppConfig): OpenProjectMcpServer { const client = new OpenProjectClient({ baseUrl: config.baseUrl, - apiKey: config.apiKey, + apiKey: config.apiKey ?? "", }); // Register all tools with ambient context execution and read-only guards diff --git a/tests/http-server.test.ts b/tests/http-server.test.ts new file mode 100644 index 0000000..745845f --- /dev/null +++ b/tests/http-server.test.ts @@ -0,0 +1,318 @@ +import { describe, test, expect, beforeAll, afterAll } from "bun:test"; +import { startHttpServer, type HttpServerInstance } from "../src/http-server.ts"; + +describe("Hosted Remote MCP Server (HTTP/SSE)", () => { + let instance: HttpServerInstance; + let serverUrl: string; + + beforeAll(async () => { + instance = await startHttpServer({ + baseUrl: "https://mock.openproject.example.com", + readOnly: true, + port: 0, // OS assigns random available port + host: "127.0.0.1", + }); + serverUrl = `http://127.0.0.1:${instance.port}`; + }); + + afterAll(async () => { + await instance.stop(); + }); + + test("GET /health returns HTTP 200 with server status", async () => { + const res = await fetch(`${serverUrl}/health`); + expect(res.status).toBe(200); + const data = (await res.json()) as { + status: string; + mode: string; + openproject: string; + readOnly: boolean; + }; + expect(data.status).toBe("ok"); + expect(data.mode).toBe("remote-mcp"); + expect(data.openproject).toBe("https://mock.openproject.example.com"); + expect(data.readOnly).toBe(true); + }); + + test("GET /sse without API key returns HTTP 401 Unauthorized", async () => { + const res = await fetch(`${serverUrl}/sse`); + expect(res.status).toBe(401); + const data = (await res.json()) as { error: string }; + expect(data.error).toContain("Missing OpenProject API key"); + }); + + test("GET /sse with X-OpenProject-Api-Key header initiates SSE stream", async () => { + const res = await fetch(`${serverUrl}/sse`, { + headers: { + "X-OpenProject-Api-Key": "test-key-header", + Accept: "text/event-stream", + }, + }); + + expect(res.status).toBe(200); + expect(res.headers.get("content-type")).toContain("text/event-stream"); + + const reader = res.body?.getReader(); + expect(reader).toBeDefined(); + + const { value } = await reader!.read(); + const text = new TextDecoder().decode(value); + expect(text).toContain("event: endpoint"); + expect(text).toContain("/messages?sessionId="); + + await reader!.cancel(); + }); + + test("GET /sse with Authorization: Bearer header initiates SSE stream", async () => { + const res = await fetch(`${serverUrl}/sse`, { + headers: { + Authorization: "Bearer test-key-bearer", + Accept: "text/event-stream", + }, + }); + + expect(res.status).toBe(200); + expect(res.headers.get("content-type")).toContain("text/event-stream"); + + const reader = res.body?.getReader(); + const { value } = await reader!.read(); + const text = new TextDecoder().decode(value); + expect(text).toContain("event: endpoint"); + + await reader!.cancel(); + }); + + test("GET /sse with ?apiKey= query parameter initiates SSE stream", async () => { + const res = await fetch(`${serverUrl}/sse?apiKey=test-key-query`, { + headers: { + Accept: "text/event-stream", + }, + }); + + expect(res.status).toBe(200); + const reader = res.body?.getReader(); + const { value } = await reader!.read(); + const text = new TextDecoder().decode(value); + expect(text).toContain("event: endpoint"); + + await reader!.cancel(); + }); + + test("POST /messages with invalid sessionId returns HTTP 404", async () => { + const res = await fetch(`${serverUrl}/messages?sessionId=non-existent-id`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ jsonrpc: "2.0", method: "ping", id: 1 }), + }); + + expect(res.status).toBe(404); + }); + + test("POST /messages with missing sessionId returns HTTP 400", async () => { + const res = await fetch(`${serverUrl}/messages`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ jsonrpc: "2.0", method: "ping", id: 1 }), + }); + + expect(res.status).toBe(400); + }); + + test("POST /messages with invalid JSON returns HTTP 400", async () => { + const sseRes = await fetch(`${serverUrl}/sse?apiKey=test-key-json`, { + headers: { Accept: "text/event-stream" }, + }); + const reader = sseRes.body?.getReader(); + const { value } = await reader!.read(); + const text = new TextDecoder().decode(value); + const match = text.match(/\/messages\?sessionId=([a-zA-Z0-9_-]+)/); + expect(match).toBeTruthy(); + const sessionId = match![1]; + + const res = await fetch(`${serverUrl}/messages?sessionId=${sessionId}`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: "not-json{", + }); + expect(res.status).toBe(400); + + await reader!.cancel(); + }); + + test("full JSON-RPC flow over SSE and POST /messages", async () => { + const sseRes = await fetch(`${serverUrl}/sse?apiKey=test-session-key`, { + headers: { Accept: "text/event-stream" }, + }); + expect(sseRes.status).toBe(200); + const reader = sseRes.body?.getReader(); + expect(reader).toBeDefined(); + + // 1. First event is endpoint + const { value: epValue } = await reader!.read(); + const epText = new TextDecoder().decode(epValue); + expect(epText).toContain("event: endpoint"); + const match = epText.match(/\/messages\?sessionId=([a-zA-Z0-9_-]+)/); + expect(match).toBeTruthy(); + const sessionId = match![1]; + + // 2. Client sends initialize request via POST /messages + const postRes = await fetch(`${serverUrl}/messages?sessionId=${sessionId}`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2024-11-05", + capabilities: {}, + clientInfo: { name: "test-client", version: "1.0.0" }, + }, + }), + }); + expect(postRes.status).toBe(202); + + // 3. Receive initialize response via SSE + const { value: msgValue } = await reader!.read(); + const msgText = new TextDecoder().decode(msgValue); + expect(msgText).toContain("event: message"); + const jsonMatch = msgText.match(/data:\s*(\{.*\})/); + expect(jsonMatch).toBeTruthy(); + const rawData = jsonMatch?.[1]; + expect(rawData).toBeDefined(); + const rpcResponse = JSON.parse(rawData!); + expect(rpcResponse.id).toBe(1); + expect(rpcResponse.result).toBeDefined(); + expect(rpcResponse.result.serverInfo.name).toBe("openproject-mcp"); + + // 4. Client sends tools/list request + const toolsPostRes = await fetch(`${serverUrl}/messages?sessionId=${sessionId}`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 2, + method: "tools/list", + params: {}, + }), + }); + expect(toolsPostRes.status).toBe(202); + + const { value: toolsValue } = await reader!.read(); + const toolsText = new TextDecoder().decode(toolsValue); + expect(toolsText).toContain("event: message"); + const toolsMatch = toolsText.match(/data:\s*(\{.*\})/); + expect(toolsMatch).toBeTruthy(); + const toolsRawData = toolsMatch?.[1]; + expect(toolsRawData).toBeDefined(); + const toolsRpcResponse = JSON.parse(toolsRawData!); + expect(toolsRpcResponse.id).toBe(2); + expect(toolsRpcResponse.result.tools).toBeDefined(); + expect(toolsRpcResponse.result.tools.length).toBeGreaterThan(0); + + await reader!.cancel(); + }); + + test("client disconnect evicts session from active sessions", async () => { + const ac = new AbortController(); + const sseRes = await fetch(`${serverUrl}/sse?apiKey=test-evict-key`, { + headers: { Accept: "text/event-stream" }, + signal: ac.signal, + }); + const reader = sseRes.body?.getReader(); + expect(reader).toBeDefined(); + + const initialCount = instance.getActiveSessionsCount(); + expect(initialCount).toBeGreaterThan(0); + + // Abort client request to simulate disconnect + ac.abort(); + + // Small delay to allow abort callback to execute + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(instance.getActiveSessionsCount()).toBeLessThan(initialCount); + }); + + test("OPTIONS /sse and /messages return 204 with CORS headers", async () => { + const res = await fetch(`${serverUrl}/sse`, { + method: "OPTIONS", + }); + expect(res.status).toBe(204); + expect(res.headers.get("access-control-allow-origin")).toBe("*"); + expect(res.headers.get("access-control-allow-methods")).toContain("GET"); + }); + + test("concurrent clients with different API keys maintain isolated sessions", async () => { + // Client A + const resA = await fetch(`${serverUrl}/sse?apiKey=user-a-secret`, { + headers: { Accept: "text/event-stream" }, + }); + const readerA = resA.body?.getReader(); + const { value: valA } = await readerA!.read(); + const textA = new TextDecoder().decode(valA); + const matchA = textA.match(/\/messages\?sessionId=([a-zA-Z0-9_-]+)/); + expect(matchA).toBeTruthy(); + const sessionIdA = matchA![1]; + + // Client B + const resB = await fetch(`${serverUrl}/sse?apiKey=user-b-secret`, { + headers: { Accept: "text/event-stream" }, + }); + const readerB = resB.body?.getReader(); + const { value: valB } = await readerB!.read(); + const textB = new TextDecoder().decode(valB); + const matchB = textB.match(/\/messages\?sessionId=([a-zA-Z0-9_-]+)/); + expect(matchB).toBeTruthy(); + const sessionIdB = matchB![1]; + + // Session IDs must be distinct + expect(sessionIdA).not.toBe(sessionIdB); + + // Both can initialize independently + const initA = await fetch(`${serverUrl}/messages?sessionId=${sessionIdA}`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 101, + method: "initialize", + params: { + protocolVersion: "2024-11-05", + capabilities: {}, + clientInfo: { name: "client-a", version: "1.0.0" }, + }, + }), + }); + expect(initA.status).toBe(202); + + const initB = await fetch(`${serverUrl}/messages?sessionId=${sessionIdB}`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 201, + method: "initialize", + params: { + protocolVersion: "2024-11-05", + capabilities: {}, + clientInfo: { name: "client-b", version: "1.0.0" }, + }, + }), + }); + expect(initB.status).toBe(202); + + // Read response for A + const { value: respValA } = await readerA!.read(); + const respTextA = new TextDecoder().decode(respValA); + expect(respTextA).toContain('"id":101'); + + // Read response for B + const { value: respValB } = await readerB!.read(); + const respTextB = new TextDecoder().decode(respValB); + expect(respTextB).toContain('"id":201'); + + await readerA!.cancel(); + await readerB!.cancel(); + }); +});