dev-mcps/coding-agent-mcp/test/http-transport.test.js

import test from "node:test";
import assert from "node:assert/strict";
import http from "node:http";
import { startHttpToolServer } from "../src/mcp-http.js";

async function startServer(t, options) {
  try { return await startHttpToolServer(options); }
  catch (error) {
    if (error?.code === "EPERM") { t.skip("Local TCP listeners are not permitted in this sandbox"); return null; }
    throw error;
  }
}

function closeServer(server) {
  return new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}

function openSse(url, headers) {
  return new Promise((resolve, reject) => {
    const request = http.get(url, { headers }, (response) => {
      response.setEncoding("utf8");
      let body = "";
      response.on("data", (chunk) => { body += chunk; });
      resolve({ status: response.statusCode, sessionId: response.headers["x-session-id"], body: () => body, close: () => request.destroy() });
    });
    request.on("error", reject);
  });
}

function rpc(url, token, sessionId, payload) {
  return fetch(url, {
    method: "POST",
    headers: { "content-type": "application/json", Authorization: `Bearer ${token}`, ...(sessionId ? { "x-session-id": sessionId } : {}) },
    body: JSON.stringify(payload)
  });
}

function streamableRpc(url, token, sessionId, payload) {
  return fetch(url, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      Authorization: `Bearer ${token}`,
      ...(sessionId ? { "mcp-session-id": sessionId } : {})
    },
    body: JSON.stringify(payload)
  });
}

async function waitForBody(sse, expected) {
  for (let attempt = 0; attempt < 20; attempt += 1) {
    if (sse.body().includes(expected)) return;
    await new Promise((resolve) => setTimeout(resolve, 10));
  }
  assert.fail(`Timed out waiting for SSE payload: ${expected}`);
}

const tools = [{ name: "echo", description: "test tool", inputSchema: { type: "object" } }];

test("HTTP transport fails closed without credentials", async () => {
  assert.throws(() => startHttpToolServer({ serverName: "x", serverVersion: "1", tools, callTool: async () => ({}) }), /requires at least one API key/);
});

test("HTTP transport binds sessions to an authenticated identity", async (t) => {
  const listener = await startServer(t, {
    serverName: "secure-server",
    serverVersion: "1.0.0",
    tools,
    apiKeys: ["agent-a=token-a", "agent-b=token-b"],
    callTool: async (_name, _arguments, context) => ({ subject: context.subject, sessionId: context.sessionId, workspaceSubpath: context.workspaceSubpath }),
    host: "127.0.0.1",
    port: 0
  });
  if (!listener) return;
  const baseUrl = `http://${listener.host}:${listener.port}`;
  try {
    const noAuth = await fetch(`${baseUrl}/health`);
    assert.equal(noAuth.status, 401);
    const sse = await openSse(`${baseUrl}/events?workspaceSubpath=project-a`, { Authorization: "Bearer token-a" });
    try {
      assert.equal(sse.status, 200);
      assert.ok(sse.sessionId);
      const absentSession = await rpc(`${baseUrl}/mcp`, "token-a", null, { jsonrpc: "2.0", id: 1, method: "tools/call", params: { name: "echo" } });
      assert.equal(absentSession.status, 401);
      const stolenSession = await rpc(`${baseUrl}/mcp`, "token-b", sse.sessionId, { jsonrpc: "2.0", id: 2, method: "tools/call", params: { name: "echo" } });
      assert.equal(stolenSession.status, 401);
      const response = await rpc(`${baseUrl}/mcp`, "token-a", sse.sessionId, { jsonrpc: "2.0", id: 3, method: "tools/call", params: { name: "echo" } });
      const body = await response.json();
      assert.equal(response.status, 200);
      assert.equal(body.result.structuredContent.subject, "agent-a");
      assert.equal(body.result.structuredContent.workspaceSubpath, "project-a");
      const changedSubpath = await fetch(`${baseUrl}/mcp?workspaceSubpath=other`, { method: "POST", headers: { "content-type": "application/json", Authorization: "Bearer token-a", "x-session-id": sse.sessionId }, body: JSON.stringify({ jsonrpc: "2.0", id: 4, method: "ping" }) });
      assert.equal(changedSubpath.status, 400);
    } finally { sse.close(); }
  } finally { await closeServer(listener.server); }
});

test("Streamable HTTP initializes, scopes requests, and deletes MCP sessions", async (t) => {
  const initializedSessions = [];
  const listener = await startServer(t, {
    serverName: "streamable-server", serverVersion: "1.0.0", tools, apiKeys: ["token-a"],
    callTool: async (_name, _arguments, context) => ({ subject: context.subject, sessionId: context.sessionId }),
    onSessionInitialize: (context) => initializedSessions.push(context),
    host: "127.0.0.1", port: 0
  });
  if (!listener) return;
  const baseUrl = `http://${listener.host}:${listener.port}`;
  try {
    const initialize = await streamableRpc(`${baseUrl}/mcp`, "token-a", null, {
      jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1" } }
    });
    const sessionId = initialize.headers.get("mcp-session-id");
    assert.equal(initialize.status, 200);
    assert.ok(sessionId);
    assert.deepEqual(initializedSessions, [{
      sessionId,
      subject: "shared",
      workspaceSubpath: null,
      protocolVersion: "2025-03-26"
    }]);
    assert.equal(initialize.headers.get("mcp-protocol-version"), "2025-03-26");
    assert.equal((await initialize.json()).result.protocolVersion, "2025-03-26");

    const toolList = await streamableRpc(`${baseUrl}/mcp`, "token-a", sessionId, { jsonrpc: "2.0", id: 2, method: "tools/list" });
    assert.equal(toolList.status, 200);
    assert.equal((await toolList.json()).result.tools[0].name, "echo");

    const removed = await fetch(`${baseUrl}/mcp`, { method: "DELETE", headers: { Authorization: "Bearer token-a", "mcp-session-id": sessionId } });
    assert.equal(removed.status, 204);
    const afterDelete = await streamableRpc(`${baseUrl}/mcp`, "token-a", sessionId, { jsonrpc: "2.0", id: 3, method: "ping" });
    assert.equal(afterDelete.status, 404);
  } finally { await closeServer(listener.server); }
});

test("Legacy SSE fallback on /mcp exposes a message endpoint and returns JSON-RPC messages", async (t) => {
  const listener = await startServer(t, {
    serverName: "legacy-server", serverVersion: "1.0.0", tools, apiKeys: ["token-a"], callTool: async () => ({}), host: "127.0.0.1", port: 0
  });
  if (!listener) return;
  const baseUrl = `http://${listener.host}:${listener.port}`;
  const sse = await openSse(`${baseUrl}/mcp`, { Authorization: "Bearer token-a" });
  try {
    assert.equal(sse.status, 200);
    await waitForBody(sse, "event: endpoint");
    const endpoint = /data: "([^"]+)"/.exec(sse.body())?.[1];
    assert.ok(endpoint);
    const response = await fetch(new URL(endpoint, `${baseUrl}/mcp`), {
      method: "POST",
      headers: { "content-type": "application/json", Authorization: "Bearer token-a" },
      body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/list" })
    });
    assert.equal(response.status, 202);
    await waitForBody(sse, '"id":1');
  } finally {
    sse.close();
    await closeServer(listener.server);
  }
});

test("CORS is denied by default and can be explicitly allowlisted", async (t) => {
  const listener = await startServer(t, {
    serverName: "cors-server", serverVersion: "1.0.0", tools, apiKeys: ["token"], corsOrigins: ["https://console.example"], callTool: async () => ({}), host: "127.0.0.1", port: 0
  });
  if (!listener) return;
  const url = `http://${listener.host}:${listener.port}/health`;
  try {
    const denied = await fetch(url, { headers: { Authorization: "Bearer token", Origin: "https://evil.example" } });
    assert.equal(denied.headers.get("access-control-allow-origin"), null);
    const allowed = await fetch(url, { headers: { Authorization: "Bearer token", Origin: "https://console.example" } });
    assert.equal(allowed.headers.get("access-control-allow-origin"), "https://console.example");
  } finally { await closeServer(listener.server); }
});