import { Context, Data, Effect, Layer, Option, Ref, Schema, Scope } from "effect";import * as http from "node:http";import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js";import { OAuthTestServer } from "@executor-js/sdk/testing";import z from "zod"; export type McpTestServer = { readonly url: string; readonly endpoint: string; /** Number of MCP sessions created (each connect = 1 session) */ readonly sessionCount: () => number; /** Requests the server has accepted and not yet finished answering. */ readonly inFlightRequests: () => number; readonly requests: Effect.Effect<readonly McpTestRequest[]>; readonly clearRequests: Effect.Effect<void>; /** Drops all server-side session registrations without notifying clients. */ readonly forgetSessions: Effect.Effect<void>; /** Rejects the next request carrying an MCP session id with this status. */ readonly rejectNextSessionRequest: (status: number) => Effect.Effect<void>; /** From now on, rejects every session-carrying POST whose JSON-RPC body * names this method with the given status (401 answers with the same * WWW-Authenticate shape as the auth gate). Models a bearer revoked right * after the handshake: `initialize` succeeds, the named request meets the * auth wall. */ readonly rejectSessionMethod: (method: string, status: number) => Effect.Effect<void>;}; export type McpTestRequest = { readonly method: string; readonly url: string; readonly authorization: string | undefined; readonly sessionId: string | undefined;}; export type McpTestServerOptions = { readonly path?: string; /** Hold authenticated requests at the transport boundary until the test * releases them, so callback ordering does not depend on elapsed time. */ readonly beforeAuthenticatedRequest?: () => Promise<void>; readonly auth?: { readonly validateAuthorization: (authorization: string | undefined) => Effect.Effect<boolean>; readonly authorizationServerUrls?: readonly string[]; readonly scopes?: readonly string[]; readonly wwwAuthenticate?: string; };}; export class McpTestServerError extends Data.TaggedError("McpTestServerError")<{ readonly cause: unknown;}> {} const writeJson = ( response: http.ServerResponse, status: number, body: Readonly<Record<string, unknown>>, headers: Readonly<Record<string, string>> = {},) => { response.writeHead(status, { "content-type": "application/json", ...headers, }); response.end(JSON.stringify(body));}; const writeText = (response: http.ServerResponse, status: number, body: string) => { response.writeHead(status, { "content-type": "text/plain; charset=utf-8" }); response.end(body);}; const readRequestBody = ( request: http.IncomingMessage,): Effect.Effect<string, McpTestServerError> => Effect.tryPromise({ try: () => new Promise<string>((resolve, reject) => { const chunks: Buffer[] = []; request.on("data", (chunk: Buffer) => chunks.push(chunk)); request.on("end", () => resolve(Buffer.concat(chunks).toString("utf8"))); request.on("error", reject); }), catch: (cause) => new McpTestServerError({ cause }), }); const decodeJsonBody = Schema.decodeUnknownOption(Schema.fromJsonString(Schema.Unknown)); const isMcpPath = (url: string, path: string): boolean => { const parsed = new URL(url, "http://executor.test"); return parsed.pathname === path;}; const protectedResourcePath = "/.well-known/oauth-protected-resource"; export const serveMcpServer = (factory: () => McpServer, options: McpTestServerOptions = {}) => Effect.acquireRelease( Effect.gen(function* () { const transports = new Map<string, StreamableHTTPServerTransport>(); const allTransports = new Set<StreamableHTTPServerTransport>(); const requests = yield* Ref.make<readonly McpTestRequest[]>([]); const path = options.path ?? "/"; let sessions = 0; let nextSessionRequestStatus: number | undefined; let sessionMethodRejection: { readonly method: string; readonly status: number } | undefined; const writeUnauthorized = (response: http.ServerResponse, origin: string) => writeJson( response, 401, { error: "invalid_token" }, { "www-authenticate": options.auth?.wwwAuthenticate ?? `Bearer resource_metadata="${origin}${protectedResourcePath}${path}", error="invalid_token"`, }, ); const namesJsonRpcMethod = (parsedBody: unknown, method: string): boolean => { const messages = Array.isArray(parsedBody) ? parsedBody : [parsedBody]; return messages.some( (message) => typeof message === "object" && message !== null && (message as { readonly method?: unknown }).method === method, ); }; const handleMcpRequest = ( request: http.IncomingMessage, response: http.ServerResponse, ): Effect.Effect<void> => Effect.gen(function* () { const requestUrl = request.url ?? "/"; const sessionId = Array.isArray(request.headers["mcp-session-id"]) ? request.headers["mcp-session-id"][0] : request.headers["mcp-session-id"]; const authorization = Array.isArray(request.headers.authorization) ? request.headers.authorization[0] : request.headers.authorization; const origin = request.headers.host ? `http://${request.headers.host}` : "http://127.0.0.1"; yield* Ref.update(requests, (all) => [ ...all, { method: request.method ?? "GET", url: requestUrl, authorization, sessionId, }, ]); if ( options.auth?.authorizationServerUrls && requestUrl.startsWith(protectedResourcePath) ) { const resourcePath = requestUrl.slice(protectedResourcePath.length); writeJson(response, 200, { resource: `${origin}${resourcePath}`, authorization_servers: options.auth.authorizationServerUrls, bearer_methods_supported: ["header"], scopes_supported: options.auth.scopes ?? ["read"], }); return; } if (!isMcpPath(requestUrl, path)) { writeJson(response, 404, { error: "not_found" }); return; } if (options.auth) { const accepted = yield* options.auth.validateAuthorization(authorization); if (!accepted) { writeUnauthorized(response, origin); return; } if (options.beforeAuthenticatedRequest !== undefined) { yield* Effect.promise(options.beforeAuthenticatedRequest); } } if (sessionId && request.method === "POST" && nextSessionRequestStatus !== undefined) { const status = nextSessionRequestStatus; nextSessionRequestStatus = undefined; writeText(response, status, `Forced HTTP ${status}`); return; } const existingTransport = sessionId ? transports.get(sessionId) : undefined; if (sessionId && !existingTransport) { writeText(response, 404, "Session not found"); return; } if (existingTransport) { const rejection = sessionMethodRejection; if (rejection !== undefined && request.method === "POST") { const body = yield* readRequestBody(request); const parsedBody = Option.getOrUndefined(decodeJsonBody(body)); if (namesJsonRpcMethod(parsedBody, rejection.method)) { if (rejection.status === 401 && options.auth) { writeUnauthorized(response, origin); } else { writeText(response, rejection.status, `Forced HTTP ${rejection.status}`); } return; } yield* Effect.tryPromise({ try: () => existingTransport.handleRequest(request, response, parsedBody), catch: (cause) => new McpTestServerError({ cause }), }); return; } yield* Effect.tryPromise({ try: () => existingTransport.handleRequest(request, response), catch: (cause) => new McpTestServerError({ cause }), }); return; } // Mirror the real v1 transport's stateful contract: only an // `initialize` POST opens a session; any other sessionless POST // (e.g. a v2 client's `server/discover` era probe) is rejected // with 400 + a JSON-RPC error and no session is minted. let parsedBody: unknown; if (request.method === "POST") { const body = yield* readRequestBody(request); parsedBody = Option.getOrUndefined(decodeJsonBody(body)); if (!isInitializeRequest(parsedBody)) { writeJson(response, 400, { jsonrpc: "2.0", error: { code: -32000, message: "Bad Request: Server not initialized" }, id: null, }); return; } } const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: () => crypto.randomUUID(), onsessioninitialized: (sid) => { transports.set(sid, transport); }, }); allTransports.add(transport); sessions += 1; const mcpServer = factory(); yield* Effect.tryPromise({ try: () => mcpServer.connect(transport), catch: (cause) => new McpTestServerError({ cause }), }); yield* Effect.tryPromise({ try: () => transport.handleRequest(request, response, parsedBody), catch: (cause) => new McpTestServerError({ cause }), }); }).pipe( Effect.catch(() => Effect.sync(() => { if (!response.headersSent) { writeJson(response, 500, { error: "mcp_test_server_failed" }); } else if (!response.writableEnded) { response.end(); } }), ), ); // An abandoned SSE `GET` leaves the session gone but the request open, // which `sessionCount` cannot see and socket counting cannot either // (keep-alive holds idle sockets open regardless). let inFlight = 0; const nodeServer = http.createServer((request, response) => { inFlight += 1; response.once("close", () => { inFlight -= 1; }); void Effect.runPromise(handleMcpRequest(request, response)); }); const port = yield* Effect.callback<number, McpTestServerError>((resume) => { const onError = (cause: Error) => { nodeServer.off("error", onError); resume(Effect.fail(new McpTestServerError({ cause }))); }; nodeServer.once("error", onError); nodeServer.listen(0, () => { nodeServer.off("error", onError); const address = nodeServer.address(); if (typeof address === "object" && address) { resume(Effect.succeed(address.port)); return; } resume(Effect.fail(new McpTestServerError({ cause: address }))); }); }); const baseUrl = `http://127.0.0.1:${port}`; const endpoint = path === "/" ? baseUrl : new URL(path, baseUrl).toString(); return { url: endpoint, endpoint, sessionCount: () => sessions, inFlightRequests: () => inFlight, requests: Ref.get(requests), clearRequests: Ref.set(requests, []), forgetSessions: Effect.sync(() => transports.clear()), rejectNextSessionRequest: (status: number) => Effect.sync(() => { nextSessionRequestStatus = status; }), rejectSessionMethod: (method: string, status: number) => Effect.sync(() => { sessionMethodRejection = { method, status }; }), close: Effect.gen(function* () { for (const transport of allTransports) { yield* Effect.tryPromise({ try: () => transport.close(), catch: (cause) => new McpTestServerError({ cause }), }).pipe(Effect.ignore); } yield* Effect.callback<void>((resume) => { nodeServer.close(() => resume(Effect.void)); nodeServer.closeAllConnections?.(); }); // closeAllConnections destroys sockets out from under in-flight SDK // reads; give those rejections a beat to settle before the scope // ends so they surface inside the test, not as unhandled rejections. // A raw timer, not Effect.sleep: this runs inside it.effect tests // whose TestClock never advances wall-clock sleeps. yield* Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, 25))); }), }; }), (server) => server.close.pipe(Effect.ignore), ).pipe(Effect.map(({ close: _close, ...server }) => server)); export const serveMcpServerWithOAuth = ( factory: () => McpServer, options: Omit<McpTestServerOptions, "auth"> & { readonly scopes?: readonly string[]; readonly wwwAuthenticate?: string; } = {},) => Effect.gen(function* () { const oauth = yield* OAuthTestServer; return yield* serveMcpServer(factory, { path: options.path, beforeAuthenticatedRequest: options.beforeAuthenticatedRequest, auth: { validateAuthorization: oauth.acceptsAuthorizationHeader, authorizationServerUrls: [oauth.issuerUrl], scopes: options.scopes ?? ["read"], wwwAuthenticate: options.wwwAuthenticate, }, }); }); export class McpTestServerLayer extends Context.Service<McpTestServerLayer, McpTestServer>()( "@executor-js/plugin-mcp/testing/McpTestServer",) { static readonly layer = ( factory: () => McpServer, options?: McpTestServerOptions, ): Layer.Layer<McpTestServerLayer, McpTestServerError, Scope.Scope> => Layer.effect(McpTestServerLayer, serveMcpServer(factory, options)); static readonly layerWithOAuth = ( factory: () => McpServer, options?: Omit<McpTestServerOptions, "auth"> & { readonly scopes?: readonly string[]; readonly wwwAuthenticate?: string; }, ): Layer.Layer<McpTestServerLayer, McpTestServerError, Scope.Scope | OAuthTestServer> => Layer.effect(McpTestServerLayer, serveMcpServerWithOAuth(factory, options));} export const makeGreetingMcpServer = ( options: { readonly name?: string; readonly version?: string; readonly toolName?: string; readonly toolDescription?: string; readonly text?: string; } = {},) => { const server = new McpServer( { name: options.name ?? "executor-test-mcp", version: options.version ?? "1.0.0", }, { capabilities: {} }, ); server.registerTool( options.toolName ?? "simple_echo", { description: options.toolDescription ?? "Echoes from the executor MCP test server", inputSchema: {}, }, async () => ({ content: [{ type: "text" as const, text: options.text ?? "mcp-ok" }], }), ); return server;}; export const TEST_IMAGE_MIME_TYPE = "image/png";export const TEST_IMAGE_PNG_BASE64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAFgwJ/l8N1wwAAAABJRU5ErkJggg=="; const testImageMetadata = () => ({ name: "mcp-image-fixture.png", mimeType: TEST_IMAGE_MIME_TYPE, byteLength: Buffer.from(TEST_IMAGE_PNG_BASE64, "base64").byteLength,}); export const makeImageMcpServer = () => { const server = new McpServer( { name: "image-test-server", version: "1.0.0" }, { capabilities: {} }, ); server.registerTool( "image_fixture", { description: "Returns a deterministic PNG as MCP image content", inputSchema: {}, }, async () => ({ content: [ { type: "image" as const, data: TEST_IMAGE_PNG_BASE64, mimeType: TEST_IMAGE_MIME_TYPE, }, ], structuredContent: testImageMetadata(), }), ); server.registerTool( "image_fixture_with_metadata", { description: "Returns text metadata followed by MCP image content", inputSchema: {}, }, async () => ({ content: [ { type: "text" as const, text: "Deterministic image fixture: mcp-image-fixture.png (image/png, 70 bytes)", }, { type: "image" as const, data: TEST_IMAGE_PNG_BASE64, mimeType: TEST_IMAGE_MIME_TYPE, }, ], structuredContent: testImageMetadata(), }), ); return server;}; export const makeEchoMcpServer = ( options: { readonly name?: string; readonly version?: string; readonly toolName?: string; readonly toolDescription?: string; readonly inputName?: "name" | "value" | "marker"; readonly text?: (value: string) => string; } = {},) => { const inputName = options.inputName ?? "value"; const server = new McpServer( { name: options.name ?? "executor-echo-mcp", version: options.version ?? "1.0.0", }, { capabilities: {} }, ); server.registerTool( options.toolName ?? "echo", { description: options.toolDescription ?? "Echoes a string value", inputSchema: { [inputName]: z.string() }, }, async (input) => ({ content: [ { type: "text" as const,Lines 1–500 of 730. Load more
import { Context, Data, Effect, Layer, Option, Ref, Schema, Scope } from "effect"; import * as http from "node:http"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"; import { OAuthTestServer } from "@executor-js/sdk/testing"; import z from "zod"; export type McpTestServer = { readonly url: string; readonly endpoint: string; /** Number of MCP sessions created (each connect = 1 session) */ readonly sessionCount: () => number; /** Requests the server has accepted and not yet finished answering. */ readonly inFlightRequests: () => number; readonly requests: Effect.Effect<readonly McpTestRequest[]>; readonly clearRequests: Effect.Effect<void>; /** Drops all server-side session registrations without notifying clients. */ readonly forgetSessions: Effect.Effect<void>; /** Rejects the next request carrying an MCP session id with this status. */ readonly rejectNextSessionRequest: (status: number) => Effect.Effect<void>; /** From now on, rejects every session-carrying POST whose JSON-RPC body * names this method with the given status (401 answers with the same * WWW-Authenticate shape as the auth gate). Models a bearer revoked right * after the handshake: `initialize` succeeds, the named request meets the * auth wall. */ readonly rejectSessionMethod: (method: string, status: number) => Effect.Effect<void>; }; export type McpTestRequest = { readonly method: string; readonly url: string; readonly authorization: string | undefined; readonly sessionId: string | undefined; }; export type McpTestServerOptions = { readonly path?: string; /** Hold authenticated requests at the transport boundary until the test * releases them, so callback ordering does not depend on elapsed time. */ readonly beforeAuthenticatedRequest?: () => Promise<void>; readonly auth?: { readonly validateAuthorization: (authorization: string | undefined) => Effect.Effect<boolean>; readonly authorizationServerUrls?: readonly string[]; readonly scopes?: readonly string[]; readonly wwwAuthenticate?: string; }; }; export class McpTestServerError extends Data.TaggedError("McpTestServerError")<{ readonly cause: unknown; }> {} const writeJson = ( response: http.ServerResponse, status: number, body: Readonly<Record<string, unknown>>, headers: Readonly<Record<string, string>> = {}, ) => { response.writeHead(status, { "content-type": "application/json", ...headers, }); response.end(JSON.stringify(body)); }; const writeText = (response: http.ServerResponse, status: number, body: string) => { response.writeHead(status, { "content-type": "text/plain; charset=utf-8" }); response.end(body); }; const readRequestBody = ( request: http.IncomingMessage, ): Effect.Effect<string, McpTestServerError> => Effect.tryPromise({ try: () => new Promise<string>((resolve, reject) => { const chunks: Buffer[] = []; request.on("data", (chunk: Buffer) => chunks.push(chunk)); request.on("end", () => resolve(Buffer.concat(chunks).toString("utf8"))); request.on("error", reject); }), catch: (cause) => new McpTestServerError({ cause }), }); const decodeJsonBody = Schema.decodeUnknownOption(Schema.fromJsonString(Schema.Unknown)); const isMcpPath = (url: string, path: string): boolean => { const parsed = new URL(url, "http://executor.test"); return parsed.pathname === path; }; const protectedResourcePath = "/.well-known/oauth-protected-resource"; export const serveMcpServer = (factory: () => McpServer, options: McpTestServerOptions = {}) => Effect.acquireRelease( Effect.gen(function* () { const transports = new Map<string, StreamableHTTPServerTransport>(); const allTransports = new Set<StreamableHTTPServerTransport>(); const requests = yield* Ref.make<readonly McpTestRequest[]>([]); const path = options.path ?? "/"; let sessions = 0; let nextSessionRequestStatus: number | undefined; let sessionMethodRejection: { readonly method: string; readonly status: number } | undefined; const writeUnauthorized = (response: http.ServerResponse, origin: string) => writeJson( response, 401, { error: "invalid_token" }, { "www-authenticate": options.auth?.wwwAuthenticate ?? `Bearer resource_metadata="${origin}${protectedResourcePath}${path}", error="invalid_token"`, }, ); const namesJsonRpcMethod = (parsedBody: unknown, method: string): boolean => { const messages = Array.isArray(parsedBody) ? parsedBody : [parsedBody]; return messages.some( (message) => typeof message === "object" && message !== null && (message as { readonly method?: unknown }).method === method, ); }; const handleMcpRequest = ( request: http.IncomingMessage, response: http.ServerResponse, ): Effect.Effect<void> => Effect.gen(function* () { const requestUrl = request.url ?? "/"; const sessionId = Array.isArray(request.headers["mcp-session-id"]) ? request.headers["mcp-session-id"][0] : request.headers["mcp-session-id"]; const authorization = Array.isArray(request.headers.authorization) ? request.headers.authorization[0] : request.headers.authorization; const origin = request.headers.host ? `http://${request.headers.host}` : "http://127.0.0.1"; yield* Ref.update(requests, (all) => [ ...all, { method: request.method ?? "GET", url: requestUrl, authorization, sessionId, }, ]); if ( options.auth?.authorizationServerUrls && requestUrl.startsWith(protectedResourcePath) ) { const resourcePath = requestUrl.slice(protectedResourcePath.length); writeJson(response, 200, { resource: `${origin}${resourcePath}`, authorization_servers: options.auth.authorizationServerUrls, bearer_methods_supported: ["header"], scopes_supported: options.auth.scopes ?? ["read"], }); return; } if (!isMcpPath(requestUrl, path)) { writeJson(response, 404, { error: "not_found" }); return; } if (options.auth) { const accepted = yield* options.auth.validateAuthorization(authorization); if (!accepted) { writeUnauthorized(response, origin); return; } if (options.beforeAuthenticatedRequest !== undefined) { yield* Effect.promise(options.beforeAuthenticatedRequest); } } if (sessionId && request.method === "POST" && nextSessionRequestStatus !== undefined) { const status = nextSessionRequestStatus; nextSessionRequestStatus = undefined; writeText(response, status, `Forced HTTP ${status}`); return; } const existingTransport = sessionId ? transports.get(sessionId) : undefined; if (sessionId && !existingTransport) { writeText(response, 404, "Session not found"); return; } if (existingTransport) { const rejection = sessionMethodRejection; if (rejection !== undefined && request.method === "POST") { const body = yield* readRequestBody(request); const parsedBody = Option.getOrUndefined(decodeJsonBody(body)); if (namesJsonRpcMethod(parsedBody, rejection.method)) { if (rejection.status === 401 && options.auth) { writeUnauthorized(response, origin); } else { writeText(response, rejection.status, `Forced HTTP ${rejection.status}`); } return; } yield* Effect.tryPromise({ try: () => existingTransport.handleRequest(request, response, parsedBody), catch: (cause) => new McpTestServerError({ cause }), }); return; } yield* Effect.tryPromise({ try: () => existingTransport.handleRequest(request, response), catch: (cause) => new McpTestServerError({ cause }), }); return; } // Mirror the real v1 transport's stateful contract: only an // `initialize` POST opens a session; any other sessionless POST // (e.g. a v2 client's `server/discover` era probe) is rejected // with 400 + a JSON-RPC error and no session is minted. let parsedBody: unknown; if (request.method === "POST") { const body = yield* readRequestBody(request); parsedBody = Option.getOrUndefined(decodeJsonBody(body)); if (!isInitializeRequest(parsedBody)) { writeJson(response, 400, { jsonrpc: "2.0", error: { code: -32000, message: "Bad Request: Server not initialized" }, id: null, }); return; } } const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: () => crypto.randomUUID(), onsessioninitialized: (sid) => { transports.set(sid, transport); }, }); allTransports.add(transport); sessions += 1; const mcpServer = factory(); yield* Effect.tryPromise({ try: () => mcpServer.connect(transport), catch: (cause) => new McpTestServerError({ cause }), }); yield* Effect.tryPromise({ try: () => transport.handleRequest(request, response, parsedBody), catch: (cause) => new McpTestServerError({ cause }), }); }).pipe( Effect.catch(() => Effect.sync(() => { if (!response.headersSent) { writeJson(response, 500, { error: "mcp_test_server_failed" }); } else if (!response.writableEnded) { response.end(); } }), ), ); // An abandoned SSE `GET` leaves the session gone but the request open, // which `sessionCount` cannot see and socket counting cannot either // (keep-alive holds idle sockets open regardless). let inFlight = 0; const nodeServer = http.createServer((request, response) => { inFlight += 1; response.once("close", () => { inFlight -= 1; }); void Effect.runPromise(handleMcpRequest(request, response)); }); const port = yield* Effect.callback<number, McpTestServerError>((resume) => { const onError = (cause: Error) => { nodeServer.off("error", onError); resume(Effect.fail(new McpTestServerError({ cause }))); }; nodeServer.once("error", onError); nodeServer.listen(0, () => { nodeServer.off("error", onError); const address = nodeServer.address(); if (typeof address === "object" && address) { resume(Effect.succeed(address.port)); return; } resume(Effect.fail(new McpTestServerError({ cause: address }))); }); }); const baseUrl = `http://127.0.0.1:${port}`; const endpoint = path === "/" ? baseUrl : new URL(path, baseUrl).toString(); return { url: endpoint, endpoint, sessionCount: () => sessions, inFlightRequests: () => inFlight, requests: Ref.get(requests), clearRequests: Ref.set(requests, []), forgetSessions: Effect.sync(() => transports.clear()), rejectNextSessionRequest: (status: number) => Effect.sync(() => { nextSessionRequestStatus = status; }), rejectSessionMethod: (method: string, status: number) => Effect.sync(() => { sessionMethodRejection = { method, status }; }), close: Effect.gen(function* () { for (const transport of allTransports) { yield* Effect.tryPromise({ try: () => transport.close(), catch: (cause) => new McpTestServerError({ cause }), }).pipe(Effect.ignore); } yield* Effect.callback<void>((resume) => { nodeServer.close(() => resume(Effect.void)); nodeServer.closeAllConnections?.(); }); // closeAllConnections destroys sockets out from under in-flight SDK // reads; give those rejections a beat to settle before the scope // ends so they surface inside the test, not as unhandled rejections. // A raw timer, not Effect.sleep: this runs inside it.effect tests // whose TestClock never advances wall-clock sleeps. yield* Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, 25))); }), }; }), (server) => server.close.pipe(Effect.ignore), ).pipe(Effect.map(({ close: _close, ...server }) => server)); export const serveMcpServerWithOAuth = ( factory: () => McpServer, options: Omit<McpTestServerOptions, "auth"> & { readonly scopes?: readonly string[]; readonly wwwAuthenticate?: string; } = {}, ) => Effect.gen(function* () { const oauth = yield* OAuthTestServer; return yield* serveMcpServer(factory, { path: options.path, beforeAuthenticatedRequest: options.beforeAuthenticatedRequest, auth: { validateAuthorization: oauth.acceptsAuthorizationHeader, authorizationServerUrls: [oauth.issuerUrl], scopes: options.scopes ?? ["read"], wwwAuthenticate: options.wwwAuthenticate, }, }); }); export class McpTestServerLayer extends Context.Service<McpTestServerLayer, McpTestServer>()( "@executor-js/plugin-mcp/testing/McpTestServer", ) { static readonly layer = ( factory: () => McpServer, options?: McpTestServerOptions, ): Layer.Layer<McpTestServerLayer, McpTestServerError, Scope.Scope> => Layer.effect(McpTestServerLayer, serveMcpServer(factory, options)); static readonly layerWithOAuth = ( factory: () => McpServer, options?: Omit<McpTestServerOptions, "auth"> & { readonly scopes?: readonly string[]; readonly wwwAuthenticate?: string; }, ): Layer.Layer<McpTestServerLayer, McpTestServerError, Scope.Scope | OAuthTestServer> => Layer.effect(McpTestServerLayer, serveMcpServerWithOAuth(factory, options)); } export const makeGreetingMcpServer = ( options: { readonly name?: string; readonly version?: string; readonly toolName?: string; readonly toolDescription?: string; readonly text?: string; } = {}, ) => { const server = new McpServer( { name: options.name ?? "executor-test-mcp", version: options.version ?? "1.0.0", }, { capabilities: {} }, ); server.registerTool( options.toolName ?? "simple_echo", { description: options.toolDescription ?? "Echoes from the executor MCP test server", inputSchema: {}, }, async () => ({ content: [{ type: "text" as const, text: options.text ?? "mcp-ok" }], }), ); return server; }; export const TEST_IMAGE_MIME_TYPE = "image/png"; export const TEST_IMAGE_PNG_BASE64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAFgwJ/l8N1wwAAAABJRU5ErkJggg=="; const testImageMetadata = () => ({ name: "mcp-image-fixture.png", mimeType: TEST_IMAGE_MIME_TYPE, byteLength: Buffer.from(TEST_IMAGE_PNG_BASE64, "base64").byteLength, }); export const makeImageMcpServer = () => { const server = new McpServer( { name: "image-test-server", version: "1.0.0" }, { capabilities: {} }, ); server.registerTool( "image_fixture", { description: "Returns a deterministic PNG as MCP image content", inputSchema: {}, }, async () => ({ content: [ { type: "image" as const, data: TEST_IMAGE_PNG_BASE64, mimeType: TEST_IMAGE_MIME_TYPE, }, ], structuredContent: testImageMetadata(), }), ); server.registerTool( "image_fixture_with_metadata", { description: "Returns text metadata followed by MCP image content", inputSchema: {}, }, async () => ({ content: [ { type: "text" as const, text: "Deterministic image fixture: mcp-image-fixture.png (image/png, 70 bytes)", }, { type: "image" as const, data: TEST_IMAGE_PNG_BASE64, mimeType: TEST_IMAGE_MIME_TYPE, }, ], structuredContent: testImageMetadata(), }), ); return server; }; export const makeEchoMcpServer = ( options: { readonly name?: string; readonly version?: string; readonly toolName?: string; readonly toolDescription?: string; readonly inputName?: "name" | "value" | "marker"; readonly text?: (value: string) => string; } = {}, ) => { const inputName = options.inputName ?? "value"; const server = new McpServer( { name: options.name ?? "executor-echo-mcp", version: options.version ?? "1.0.0", }, { capabilities: {} }, ); server.registerTool( options.toolName ?? "echo", { description: options.toolDescription ?? "Echoes a string value", inputSchema: { [inputName]: z.string() }, }, async (input) => ({ content: [ { type: "text" as const,