From c6016c1e28e9fa9dc31034512109f125ecef07da Mon Sep 17 00:00:00 2001 From: Edward Bramanti Date: Sat, 25 Jul 2026 01:43:08 -0600 Subject: [PATCH 1/4] Fix MCP Streamable HTTP transport compliance --- .changeset/calm-servers-share.md | 5 + packages/effect/src/unstable/ai/McpServer.ts | 58 ++++++-- .../effect/test/unstable/ai/McpServer.test.ts | 129 +++++++++++++++++- 3 files changed, 177 insertions(+), 15 deletions(-) create mode 100644 .changeset/calm-servers-share.md diff --git a/.changeset/calm-servers-share.md b/.changeset/calm-servers-share.md new file mode 100644 index 00000000000..504a98ce55d --- /dev/null +++ b/.changeset/calm-servers-share.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +Update `McpServer.layerHttp` to return `405` for unsupported HTTP methods, reject unsupported `MCP-Protocol-Version` headers with `400`, and return an empty `202` for accepted notifications and responses. diff --git a/packages/effect/src/unstable/ai/McpServer.ts b/packages/effect/src/unstable/ai/McpServer.ts index f602c0227df..855c7adc53a 100644 --- a/packages/effect/src/unstable/ai/McpServer.ts +++ b/packages/effect/src/unstable/ai/McpServer.ts @@ -440,14 +440,41 @@ export const run: (options: { ...protocol, run: (f) => protocol.run((clientId, request_) => { - const request = request_ as any as + const request = request_ as unknown as | RpcMessage.FromServerEncoded | RpcMessage.FromClientEncoded + const httpRequest = isHttp + ? Context.getUnsafe(Fiber.getCurrent()!.context, HttpServerRequest.HttpServerRequest) + : undefined + if (httpRequest !== undefined) { + const protocolVersion = httpRequest.headers[mcpProtocolVersionHeader] + if ( + protocolVersion !== undefined && + !SUPPORTED_PROTOCOL_VERSIONS.includes(protocolVersion) + ) { + appendPreResponseHandlerUnsafe( + httpRequest, + () => Effect.succeed(HttpServerResponse.empty({ status: 400 })) + ) + return Effect.die(new Error(`Unsupported MCP-Protocol-Version`)) + } + } + if (httpRequest !== undefined && request._tag !== "Eof") { + appendPreResponseHandlerUnsafe(httpRequest, (_, response) => + Effect.succeed( + response.status === 200 && + response.body._tag === "Uint8Array" && + response.body.contentLength === 0 + ? HttpServerResponse.empty({ + headers: Headers.remove(response.headers, "content-type"), + status: 202 + }) + : response + )) + } switch (request._tag) { case "Request": { - if (isHttp) { - const fiber = Fiber.getCurrent()! - const httpRequest = Context.getUnsafe(fiber.context, HttpServerRequest.HttpServerRequest) + if (httpRequest !== undefined) { const client = getInitializedClient(clientSessions, clientId, httpRequest.headers) if (client) { appendPreResponseHandlerUnsafe(httpRequest, (_, res) => @@ -636,8 +663,7 @@ export const layerStdio = (options: { ) /** - * Registers an HTTP POST JSON-RPC route at `options.path` on the current - * `HttpRouter`. + * Registers a Streamable HTTP MCP endpoint at `options.path`. * * **When to use** * @@ -645,8 +671,9 @@ export const layerStdio = (options: { * * **Details** * - * This layer composes `layer(options)`, `RpcServer.layerProtocolHttp(options)`, - * and `RpcSerialization.layerJsonRpc()`. + * POST serves JSON-RPC and accepted notification-only requests return `202`. + * Unsupported protocol versions return `400`; methods without MCP handlers + * return `405`. * * @see {@link layerStdio} for exposing the server over stdio * @see {@link layer} for the base MCP server layer without a transport protocol @@ -659,11 +686,22 @@ export const layerHttp = (options: { readonly version: string readonly path: HttpRouter.PathInput readonly extensions?: Record<`${string}/${string}`, unknown> | undefined -}): Layer.Layer => - layer(options).pipe( +}): Layer.Layer => { + const methodNotAllowedResponse = HttpServerResponse.empty({ + status: 405, + headers: { allow: "POST" } + }) + const routes = Layer.mergeAll( + HttpRouter.add("GET", options.path, methodNotAllowedResponse), + HttpRouter.add("PUT", options.path, methodNotAllowedResponse), + HttpRouter.add("PATCH", options.path, methodNotAllowedResponse), + HttpRouter.add("DELETE", options.path, methodNotAllowedResponse) + ) + return Layer.merge(layer(options), routes).pipe( Layer.provide(RpcServer.layerProtocolHttp(options)), Layer.provide(RpcSerialization.layerJsonRpc()) ) +} const INTERNAL_TOOL_ERROR_MESSAGE = "Tool execution failed due to an internal server error." diff --git a/packages/effect/test/unstable/ai/McpServer.test.ts b/packages/effect/test/unstable/ai/McpServer.test.ts index a900cbae215..41d76b497a5 100644 --- a/packages/effect/test/unstable/ai/McpServer.test.ts +++ b/packages/effect/test/unstable/ai/McpServer.test.ts @@ -55,14 +55,37 @@ const TestServerLayer = McpServer.layerHttp({ version: "1.0.0", path: "/mcp" }).pipe( - Layer.provideMerge(Layer.succeed(References.CurrentLoggers, new Set([noopLogger]))) + Layer.provideMerge(Layer.succeed( + References.CurrentLoggers, + new Set([noopLogger]) + )) ) +const initializePayload = { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { + name: "TestClient", + version: "1.0.0" + } +} + +const pingBody = { + jsonrpc: "2.0", + method: "ping", + params: {}, + id: 0 +} + const makeTestClientWith = Effect.fnUntraced(function*( - serverLayer: Layer.Layer + serverLayer: Layer.Layer, + options?: { + readonly routerLayer?: Layer.Layer | undefined + } | undefined ) { const responses: Array = [] - const { handler, dispose } = HttpRouter.toWebHandler(serverLayer, { disableLogger: true }) + const appLayer = options?.routerLayer ? Layer.merge(serverLayer, options.routerLayer) : serverLayer + const { handler, dispose } = HttpRouter.toWebHandler(appLayer, { disableLogger: true }) yield* Effect.addFinalizer(() => Effect.promise(() => dispose())) let sessionId: string | null = null @@ -72,7 +95,7 @@ const makeTestClientWith = Effect.fnUntraced(function*( request.headers.set("Mcp-Session-Id", sessionId) } const response = await handler(request) - sessionId = response.headers.get("Mcp-Session-Id") + sessionId = response.headers.get("Mcp-Session-Id") ?? sessionId responses.push(response.clone()) return response } @@ -94,6 +117,10 @@ const makeTestClientWith = Effect.fnUntraced(function*( const makeTestClient = makeTestClientWith(TestServerLayer) +const makeRouterTestClient = ( + router: Layer.Layer +) => makeTestClientWith(TestServerLayer, { routerLayer: router }) + const makeToolkitTestClient = Effect.fnUntraced(function*(handlers: TestToolkitHandlers = testToolkitHandlers) { const serverLayer = McpServer.toolkit(TestToolkit).pipe( Layer.provideMerge(TestToolkit.toLayer(handlers)), @@ -135,13 +162,14 @@ describe("McpServer", () => { strictEqual(responses.length, 2) strictEqual(responses[0].headers.get("Mcp-Protocol-Version"), "2025-06-18") + strictEqual(responses[1].headers.get("Mcp-Protocol-Version"), "2025-06-18") })) it.effect("returns 404 when a non-initialize request omits the MCP session id", () => Effect.gen(function*() { const { httpClient } = yield* makeTestClient - const response = yield* HttpClientRequest.post("http://locahost/mcp").pipe( + const response = yield* HttpClientRequest.post("http://localhost/mcp").pipe( HttpClientRequest.bodyJsonUnsafe({ jsonrpc: "2.0", method: "ping", params: {}, id: 0 }), httpClient.execute ) @@ -261,4 +289,95 @@ describe("McpServer", () => { assert.strictEqual(error.message, "Tool 'UnknownTool' not found") })) }) + + it.effect("rejects unsupported HTTP methods without disturbing an initialized session", () => + Effect.gen(function*() { + const { client, httpClient } = yield* makeTestClient + + yield* client.initialize(initializePayload) + + for (const method of ["GET", "PUT", "PATCH", "DELETE", "HEAD"] as const) { + const response = yield* HttpClientRequest.make(method)("http://localhost/mcp").pipe( + httpClient.execute + ) + strictEqual(response.status, 405) + strictEqual(response.headers["allow"], "POST") + } + + yield* client.ping({}) + })) + + it.effect("returns an empty 202 for notifications and responses and remains successful for request POSTs", () => + Effect.gen(function*() { + const { client, httpClient } = yield* makeRouterTestClient(HttpRouter.cors()) + + yield* client.initialize(initializePayload) + + const notificationResponse = yield* HttpClientRequest.post("http://localhost/mcp").pipe( + HttpClientRequest.bodyJsonUnsafe({ + jsonrpc: "2.0", + method: "notifications/initialized", + params: {} + }), + httpClient.execute + ) + strictEqual(notificationResponse.status, 202) + strictEqual(yield* notificationResponse.text, "") + strictEqual(notificationResponse.headers["content-type"], undefined) + strictEqual(notificationResponse.headers["access-control-allow-origin"], "*") + strictEqual(notificationResponse.headers["mcp-protocol-version"], "2025-06-18") + + const responseOnly = yield* HttpClientRequest.post("http://localhost/mcp").pipe( + HttpClientRequest.bodyJsonUnsafe({ jsonrpc: "2.0", id: 1, result: {} }), + httpClient.execute + ) + strictEqual(responseOnly.status, 202) + strictEqual(yield* responseOnly.text, "") + + const pingResponse = yield* HttpClientRequest.post("http://localhost/mcp").pipe( + HttpClientRequest.bodyJsonUnsafe(pingBody), + httpClient.execute + ) + strictEqual(pingResponse.status, 200) + const pingResponseBody = yield* pingResponse.text + strictEqual(pingResponseBody.length > 0, true) + })) + + it.effect("validates supplied protocol versions on POST", () => + Effect.gen(function*() { + const { client, httpClient } = yield* makeTestClient + + yield* client.initialize(initializePayload) + + const unsupportedResponse = yield* HttpClientRequest.post("http://localhost/mcp").pipe( + HttpClientRequest.bodyJsonUnsafe(pingBody), + HttpClientRequest.setHeader("Mcp-Protocol-Version", "9999-01-01"), + httpClient.execute + ) + strictEqual(unsupportedResponse.status, 400) + strictEqual(yield* unsupportedResponse.text, "") + + const responseOnly = yield* HttpClientRequest.post("http://localhost/mcp").pipe( + HttpClientRequest.bodyJsonUnsafe({ jsonrpc: "2.0", id: 1, result: {} }), + HttpClientRequest.setHeader("Mcp-Protocol-Version", "9999-01-01"), + httpClient.execute + ) + strictEqual(responseOnly.status, 400) + strictEqual(yield* responseOnly.text, "") + + const absentVersionResponse = yield* HttpClientRequest.post("http://localhost/mcp").pipe( + HttpClientRequest.bodyJsonUnsafe(pingBody), + httpClient.execute + ) + strictEqual(absentVersionResponse.status, 200) + + for (const protocolVersion of ["2025-06-18", "2025-03-26", "2024-11-05", "2024-10-07"]) { + const response = yield* HttpClientRequest.post("http://localhost/mcp").pipe( + HttpClientRequest.bodyJsonUnsafe(pingBody), + HttpClientRequest.setHeader("Mcp-Protocol-Version", protocolVersion), + httpClient.execute + ) + strictEqual(response.status, 200) + } + })) }) From 64c882256d9d11565bc0f75f50fac70d1b972685 Mon Sep 17 00:00:00 2001 From: Maxwell Brown Date: Sat, 25 Jul 2026 13:12:15 -0400 Subject: [PATCH 2/4] test(ai): add MCP initialization conformance suite --- .../ai/McpServerInitialization.test.ts | 291 ++++++++++++++++++ 1 file changed, 291 insertions(+) create mode 100644 packages/effect/test/unstable/ai/McpServerInitialization.test.ts diff --git a/packages/effect/test/unstable/ai/McpServerInitialization.test.ts b/packages/effect/test/unstable/ai/McpServerInitialization.test.ts new file mode 100644 index 00000000000..bf5d5d0ca78 --- /dev/null +++ b/packages/effect/test/unstable/ai/McpServerInitialization.test.ts @@ -0,0 +1,291 @@ +import { assert, describe, it } from "@effect/vitest" +import * as Effect from "effect/Effect" +import { constVoid } from "effect/Function" +import * as Layer from "effect/Layer" +import * as Logger from "effect/Logger" +import * as References from "effect/References" +import * as Schema from "effect/Schema" +import * as McpSchema from "effect/unstable/ai/McpSchema" +import * as McpServer from "effect/unstable/ai/McpServer" +import * as Tool from "effect/unstable/ai/Tool" +import * as Toolkit from "effect/unstable/ai/Toolkit" +import * as HttpRouter from "effect/unstable/http/HttpRouter" + +const noopLogger = Logger.make(constVoid) + +const makeServerLayer = (options?: { + readonly extensions?: Record<`${string}/${string}`, unknown> | undefined +}) => + McpServer.layerHttp({ + name: "LifecycleServer", + version: "1.0.0", + path: "/mcp", + extensions: options?.extensions + }).pipe( + Layer.provideMerge(Layer.succeed( + References.CurrentLoggers, + new Set([noopLogger]) + )) + ) + +const makeHarness = Effect.fnUntraced(function*( + serverLayer: Layer.Layer = makeServerLayer() +) { + const { dispose, handler } = HttpRouter.toWebHandler(serverLayer, { disableLogger: true }) + yield* Effect.addFinalizer(() => Effect.promise(() => dispose())) + + const post = (body: unknown, headers?: HeadersInit) => + Effect.promise(() => + handler( + new Request("http://localhost/mcp", { + method: "POST", + headers: { + accept: "application/json, text/event-stream", + "content-type": "application/json", + ...headers + }, + body: JSON.stringify(body) + }) + ) + ) + + return { post } as const +}) + +const initializeRequest = (protocolVersion: string, id = 1) => ({ + jsonrpc: "2.0", + id, + method: "initialize", + params: { + protocolVersion, + capabilities: {}, + clientInfo: { + name: "LifecycleClient", + version: "1.0.0" + } + } +}) + +const initializedNotification = { + jsonrpc: "2.0", + method: "notifications/initialized" +} + +const pingRequest = { + jsonrpc: "2.0", + id: 2, + method: "ping", + params: {} +} + +const InitializeResponse = Schema.Struct({ + jsonrpc: Schema.Literal("2.0"), + id: Schema.Number, + result: McpSchema.InitializeResult +}) + +const ErrorResponse = Schema.Struct({ + jsonrpc: Schema.Literal("2.0"), + id: Schema.NullOr(Schema.Number), + error: McpSchema.McpError +}) + +const decodeInitializeResponse = Schema.decodeUnknownEffect(InitializeResponse) +const decodeErrorResponse = Schema.decodeUnknownEffect(ErrorResponse) + +type Post = (body: unknown, headers?: HeadersInit) => Effect.Effect + +const initialize = Effect.fnUntraced(function*( + post: Post, + protocolVersion: string, + id = 1 +) { + const response = yield* post(initializeRequest(protocolVersion, id)) + const body = yield* Effect.promise(() => response.json()) + return { + response, + message: yield* decodeInitializeResponse(body) + } as const +}) + +const TestTool = Tool.make("TestTool", { + success: Schema.String +}) +const TestToolkit = Toolkit.make(TestTool) +const TestToolkitLayer = McpServer.toolkit(TestToolkit).pipe( + Layer.provide(TestToolkit.toLayer({ + TestTool: () => Effect.succeed("ok") + })) +) +const FeaturesServerLayer = Layer.mergeAll( + TestToolkitLayer, + McpServer.resource({ + uri: "file:///test", + name: "TestResource", + content: Effect.succeed("test") + }), + McpServer.prompt({ + name: "TestPrompt", + content: () => Effect.succeed("test") + }) +).pipe( + Layer.provide(makeServerLayer({ + extensions: { "example/lifecycle": { enabled: true } } + })) +) + +describe("McpServer initialization", () => { + describe("2025-11-25", () => { + describe("Lifecycle", () => { + describe("1. Lifecycle Phases", () => { + describe("1.1 Initialization", () => { + it.effect("requires initialize to be the first request", () => + Effect.gen(function*() { + const { post } = yield* makeHarness() + const response = yield* post(pingRequest) + + assert.isAtLeast(response.status, 400) + })) + + it.effect("rejects initialized notifications before initialize", () => + Effect.gen(function*() { + const { post } = yield* makeHarness() + const response = yield* post(initializedNotification) + + assert.isAtLeast(response.status, 400) + })) + + it.effect("requires protocolVersion, capabilities, and clientInfo", () => + Effect.gen(function*() { + const { post } = yield* makeHarness() + const invalidParams = [ + { + capabilities: {}, + clientInfo: { name: "LifecycleClient", version: "1.0.0" } + }, + { + protocolVersion: "2025-11-25", + clientInfo: { name: "LifecycleClient", version: "1.0.0" } + }, + { + protocolVersion: "2025-11-25", + capabilities: {} + } + ] + + for (let i = 0; i < invalidParams.length; i++) { + const response = yield* post({ + jsonrpc: "2.0", + id: i + 1, + method: "initialize", + params: invalidParams[i] + }) + const body = yield* Effect.promise(() => response.json()) + const error = yield* decodeErrorResponse(body) + + assert.strictEqual(error.id, i + 1) + assert.isNumber(error.error.code) + assert.isNull(response.headers.get("Mcp-Session-Id")) + } + })) + + it.effect("returns server capabilities and implementation information", () => + Effect.gen(function*() { + const { post } = yield* makeHarness() + const { message, response } = yield* initialize(post, "2025-11-25") + + assert.strictEqual(response.status, 200) + assert.strictEqual(message.id, 1) + assert.deepStrictEqual(message.result.capabilities, { completions: {} }) + assert.deepStrictEqual(message.result.serverInfo, { + name: "LifecycleServer", + version: "1.0.0" + }) + const sessionId = response.headers.get("Mcp-Session-Id") + assert.isNotNull(sessionId) + assert.match(sessionId, /^[\x21-\x7e]+$/) + })) + + it.effect("accepts initialized after a successful initialize response", () => + Effect.gen(function*() { + const { post } = yield* makeHarness() + const initialized = yield* initialize(post, "2025-11-25") + const sessionId = initialized.response.headers.get("Mcp-Session-Id") + assert.isNotNull(sessionId) + + const response = yield* post(initializedNotification, { + "Mcp-Session-Id": sessionId, + "Mcp-Protocol-Version": initialized.message.result.protocolVersion + }) + + assert.strictEqual(response.status, 202) + assert.strictEqual(yield* Effect.promise(() => response.text()), "") + })) + }) + + describe("1.1.1 Version Negotiation", () => { + it.effect("echoes every requested version supported by the server", () => + Effect.gen(function*() { + const { post } = yield* makeHarness() + const supportedVersions = ["2025-06-18", "2025-03-26", "2024-11-05", "2024-10-07"] + + for (let i = 0; i < supportedVersions.length; i++) { + const protocolVersion = supportedVersions[i] + const { message } = yield* initialize(post, protocolVersion, i + 1) + assert.strictEqual(message.result.protocolVersion, protocolVersion) + } + })) + + it.effect("negotiates an unsupported requested version to the latest supported version", () => + Effect.gen(function*() { + const { post } = yield* makeHarness() + const { message } = yield* initialize(post, "2025-11-25") + + assert.strictEqual(message.result.protocolVersion, "2025-06-18") + })) + }) + + describe("1.1.2 Capability Negotiation", () => { + it.effect("advertises the capabilities provided by the server", () => + Effect.gen(function*() { + const { post } = yield* makeHarness(FeaturesServerLayer) + const { message } = yield* initialize(post, "2025-11-25") + + assert.deepStrictEqual(message.result.capabilities, { + completions: {}, + extensions: { "example/lifecycle": { enabled: true } }, + prompts: { listChanged: true }, + resources: { listChanged: true, subscribe: false }, + tools: { listChanged: true } + }) + })) + }) + + describe("1.2 Operation", () => { + it.effect("continues to use the version negotiated during initialization", () => + Effect.gen(function*() { + const { post } = yield* makeHarness() + const initialized = yield* initialize(post, "2025-03-26") + const sessionId = initialized.response.headers.get("Mcp-Session-Id") + assert.isNotNull(sessionId) + + const response = yield* post(pingRequest, { "Mcp-Session-Id": sessionId }) + + assert.strictEqual(response.status, 200) + assert.strictEqual(response.headers.get("Mcp-Protocol-Version"), "2025-03-26") + })) + }) + }) + + describe("3. Error Handling", () => { + it.effect("handles protocol version mismatch through version negotiation", () => + Effect.gen(function*() { + const { post } = yield* makeHarness() + const { message } = yield* initialize(post, "invalid-version") + + assert.strictEqual(message.result.protocolVersion, "2025-06-18") + })) + }) + }) + }) +}) From b979ee33fe62c8c3526f0f39fde933c0c8bed5f9 Mon Sep 17 00:00:00 2001 From: Maxwell Brown Date: Sat, 25 Jul 2026 13:15:46 -0400 Subject: [PATCH 3/4] Fix MCP protocol lifecycle handling --- packages/effect/src/unstable/ai/McpServer.ts | 129 +++++++++++++----- .../effect/test/unstable/ai/McpServer.test.ts | 9 +- 2 files changed, 98 insertions(+), 40 deletions(-) diff --git a/packages/effect/src/unstable/ai/McpServer.ts b/packages/effect/src/unstable/ai/McpServer.ts index 855c7adc53a..c30f0408a21 100644 --- a/packages/effect/src/unstable/ai/McpServer.ts +++ b/packages/effect/src/unstable/ai/McpServer.ts @@ -341,8 +341,22 @@ const SUPPORTED_PROTOCOL_VERSIONS = [ "2024-11-05", "2024-10-07" ] -const mcpSessionIdHeader = "mcp-session-id" -const mcpProtocolVersionHeader = "mcp-protocol-version" +const MCP_SESSION_ID_HEADER = "mcp-session-id" +const MCP_PROTOCOL_VERSION_HEADER = "mcp-protocol-version" + +class McpProtocolState extends Context.Service + readonly latestProtocolVersion: string + readonly supportedProtocolVersions: ReadonlySet +}>()("effect/ai/McpServer/McpProtocolState") {} + +const makeMcpProtocolState = (): McpProtocolState["Service"] => ({ + clientSessions: new Map(), + latestProtocolVersion: LATEST_PROTOCOL_VERSION, + supportedProtocolVersions: new Set(SUPPORTED_PROTOCOL_VERSIONS) +}) + +const layerMcpProtocolState = Layer.sync(McpProtocolState, makeMcpProtocolState) /** * Runs an MCP server over the current `RpcServer.Protocol`. @@ -367,12 +381,19 @@ export const run: (options: { > = Effect.fnUntraced(function*(options: { readonly name: string readonly version: string + readonly extensions?: Record<`${string}/${string}`, unknown> | undefined }) { + const protocolState = Option.getOrElse( + yield* Effect.serviceOption(McpProtocolState), + makeMcpProtocolState + ) const protocol = yield* RpcServer.Protocol const server = yield* McpServer const isHttp = Option.isSome(yield* Effect.serviceOption(HttpRouter.HttpRouter)) - const clientSessions = new Map() - const handlers = yield* Layer.build(layerHandlers(options, { clientSessions })) + const clientSessions = protocolState.clientSessions + const handlers = yield* Layer.build(layerHandlers(options)).pipe( + Effect.provideService(McpProtocolState, protocolState) + ) const clients = yield* RcMap.make({ lookup: Effect.fnUntraced(function*(clientId: number) { @@ -440,25 +461,13 @@ export const run: (options: { ...protocol, run: (f) => protocol.run((clientId, request_) => { + const fiber = Fiber.getCurrent()! const request = request_ as unknown as | RpcMessage.FromServerEncoded | RpcMessage.FromClientEncoded const httpRequest = isHttp - ? Context.getUnsafe(Fiber.getCurrent()!.context, HttpServerRequest.HttpServerRequest) + ? Context.getOrUndefined(fiber.context, HttpServerRequest.HttpServerRequest) : undefined - if (httpRequest !== undefined) { - const protocolVersion = httpRequest.headers[mcpProtocolVersionHeader] - if ( - protocolVersion !== undefined && - !SUPPORTED_PROTOCOL_VERSIONS.includes(protocolVersion) - ) { - appendPreResponseHandlerUnsafe( - httpRequest, - () => Effect.succeed(HttpServerResponse.empty({ status: 400 })) - ) - return Effect.die(new Error(`Unsupported MCP-Protocol-Version`)) - } - } if (httpRequest !== undefined && request._tag !== "Eof") { appendPreResponseHandlerUnsafe(httpRequest, (_, response) => Effect.succeed( @@ -479,12 +488,22 @@ export const run: (options: { if (client) { appendPreResponseHandlerUnsafe(httpRequest, (_, res) => Effect.succeed( - HttpServerResponse.setHeader(res, mcpProtocolVersionHeader, client.protocolVersion) + HttpServerResponse.setHeader(res, MCP_PROTOCOL_VERSION_HEADER, client.protocolVersion) )) } } const rpc = ClientNotificationRpcs.requests.get(request.tag) if (rpc) { + const headers = Headers.fromInput(request.headers) + if (!getInitializedClient(clientSessions, clientId, headers)) { + if (httpRequest !== undefined) { + appendPreResponseHandlerUnsafe( + httpRequest, + () => Effect.succeed(HttpServerResponse.empty({ status: 404 })) + ) + } + return Effect.void + } if (request.tag === "notifications/cancelled") { return f(clientId, { _tag: "Interrupt", @@ -497,7 +516,7 @@ export const run: (options: { rpc, requestId: RpcMessage.RequestId(request.id), client: new Rpc.ServerClient(clientId), - headers: Headers.fromInput(request.headers) + headers }) as any as Effect.Effect : Effect.void } @@ -590,6 +609,15 @@ export const layer = (options: { readonly version: string readonly extensions?: Record<`${string}/${string}`, unknown> | undefined }): Layer.Layer => + layerWithProtocolState(options).pipe( + Layer.provide(layerMcpProtocolState) + ) + +const layerWithProtocolState = (options: { + readonly name: string + readonly version: string + readonly extensions?: Record<`${string}/${string}`, unknown> | undefined +}): Layer.Layer => Layer.effectDiscard(Effect.forkScoped(run(options))).pipe( Layer.provideMerge(McpServer.layer) ) @@ -687,6 +715,7 @@ export const layerHttp = (options: { readonly path: HttpRouter.PathInput readonly extensions?: Record<`${string}/${string}`, unknown> | undefined }): Layer.Layer => { + const protocolState = layerMcpProtocolState const methodNotAllowedResponse = HttpServerResponse.empty({ status: 405, headers: { allow: "POST" } @@ -697,12 +726,37 @@ export const layerHttp = (options: { HttpRouter.add("PATCH", options.path, methodNotAllowedResponse), HttpRouter.add("DELETE", options.path, methodNotAllowedResponse) ) - return Layer.merge(layer(options), routes).pipe( - Layer.provide(RpcServer.layerProtocolHttp(options)), + return Layer.merge(layerWithProtocolState(options), routes).pipe( + Layer.provide(layerMcpProtocolHttp(options)), + Layer.provide(protocolState), Layer.provide(RpcSerialization.layerJsonRpc()) ) } +const layerMcpProtocolHttp = (options: { + readonly path: HttpRouter.PathInput +}): Layer.Layer< + RpcServer.Protocol, + never, + McpProtocolState | RpcSerialization.RpcSerialization | HttpRouter.HttpRouter +> => + Layer.effect(RpcServer.Protocol)(Effect.gen(function*() { + const state = yield* McpProtocolState + const { httpEffect, protocol } = yield* RpcServer.makeProtocolWithHttpEffect + const router = yield* HttpRouter.HttpRouter + yield* router.add("POST", options.path, (request) => { + const protocolVersion = request.headers[MCP_PROTOCOL_VERSION_HEADER] + if ( + protocolVersion !== undefined && + !state.supportedProtocolVersions.has(protocolVersion) + ) { + return Effect.succeed(HttpServerResponse.empty({ status: 400 })) + } + return httpEffect + }) + return protocol + })) + const INTERNAL_TOOL_ERROR_MESSAGE = "Tool execution failed due to an internal server error." const toolErrorResult = (message: string): CallToolResult => @@ -1332,21 +1386,21 @@ const layerHandlers = (serverInfo: { readonly name: string readonly version: string readonly extensions?: Record<`${string}/${string}`, unknown> | undefined -}, options: { - readonly clientSessions: Map }) => ClientRpcs.toLayer( Effect.gen(function*() { const server = yield* McpServer + const protocolState = yield* McpProtocolState + const clientSessions = protocolState.clientSessions let currentLogLevel = yield* CurrentLogLevel return ClientRpcs.of({ // Requests ping: () => Effect.succeed({}), initialize(params, { client }) { - const requestedVersion = SUPPORTED_PROTOCOL_VERSIONS.includes(params.protocolVersion) + const requestedVersion = protocolState.supportedProtocolVersions.has(params.protocolVersion) ? params.protocolVersion - : LATEST_PROTOCOL_VERSION + : protocolState.latestProtocolVersion if (requestedVersion !== params.protocolVersion) { params = { ...params, @@ -1375,14 +1429,14 @@ const layerHandlers = (serverInfo: { const httpRequest = Context.getOrUndefined(fiber.context, HttpServerRequest.HttpServerRequest) if (httpRequest) { const sessionId = crypto.randomUUID() - options.clientSessions.set(sessionId, params) + clientSessions.set(sessionId, params) appendPreResponseHandlerUnsafe(httpRequest, (_req, res) => Effect.succeed(HttpServerResponse.setHeaders(res, { - [mcpSessionIdHeader]: sessionId, - [mcpProtocolVersionHeader]: requestedVersion + [MCP_SESSION_ID_HEADER]: sessionId, + [MCP_PROTOCOL_VERSION_HEADER]: requestedVersion }))) } else { - options.clientSessions.set(String(client.id), params) + clientSessions.set(String(client.id), params) } return Effect.succeed({ capabilities, @@ -1424,12 +1478,12 @@ const layerHandlers = (serverInfo: { ), "prompts/list": (_, { client, headers }) => Effect.sync(() => { - const initialized = getInitializedClient(options.clientSessions, client.id, headers) + const initialized = getInitializedClient(clientSessions, client.id, headers) return new ListPromptsResult({ prompts: filterByClient(initialized, server.prompts, "prompt") }) }), "resources/list": (_, { client, headers }) => Effect.sync(() => { - const initialized = getInitializedClient(options.clientSessions, client.id, headers) + const initialized = getInitializedClient(clientSessions, client.id, headers) return new ListResourcesResult({ resources: filterByClient(initialized, server.resources, "resource") }) }), "resources/read": ({ uri }) => @@ -1442,7 +1496,7 @@ const layerHandlers = (serverInfo: { InternalError.notImplemented, "resources/templates/list": (_, { client, headers }) => Effect.sync(() => { - const initialized = getInitializedClient(options.clientSessions, client.id, headers) + const initialized = getInitializedClient(clientSessions, client.id, headers) return new ListResourceTemplatesResult({ resourceTemplates: filterByClient(initialized, server.resourceTemplates, "template") }) @@ -1453,7 +1507,7 @@ const layerHandlers = (serverInfo: { ), "tools/list": (_, { client, headers }) => Effect.sync(() => { - const initialized = getInitializedClient(options.clientSessions, client.id, headers) + const initialized = getInitializedClient(clientSessions, client.id, headers) return new ListToolsResult({ tools: filterByClient(initialized, server.tools, "tool") }) @@ -1461,7 +1515,10 @@ const layerHandlers = (serverInfo: { // Notifications "notifications/cancelled": (_) => Effect.void, - "notifications/initialized": (_) => Effect.void, + "notifications/initialized": (_, { client }) => + Effect.sync(() => { + server.initializedClients.add(client.id) + }), "notifications/progress": (_) => Effect.void, "notifications/roots/list_changed": (_) => Effect.void }) @@ -1519,7 +1576,7 @@ const getInitializedClient = ( clientId: number, headers: Headers.Headers ) => { - const sessionId = headers[mcpSessionIdHeader] + const sessionId = headers[MCP_SESSION_ID_HEADER] if (sessionId === undefined) { return sessions.get(String(clientId)) } diff --git a/packages/effect/test/unstable/ai/McpServer.test.ts b/packages/effect/test/unstable/ai/McpServer.test.ts index 41d76b497a5..10b5d9eead7 100644 --- a/packages/effect/test/unstable/ai/McpServer.test.ts +++ b/packages/effect/test/unstable/ai/McpServer.test.ts @@ -150,7 +150,7 @@ describe("McpServer", () => { const { client, responses } = yield* makeTestClient yield* client.initialize({ - protocolVersion: "9999-01-01", + protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "TestClient", @@ -161,8 +161,8 @@ describe("McpServer", () => { yield* client.ping({}) strictEqual(responses.length, 2) - strictEqual(responses[0].headers.get("Mcp-Protocol-Version"), "2025-06-18") - strictEqual(responses[1].headers.get("Mcp-Protocol-Version"), "2025-06-18") + strictEqual(responses[0].headers.get("Mcp-Protocol-Version"), "2025-03-26") + strictEqual(responses[1].headers.get("Mcp-Protocol-Version"), "2025-03-26") })) it.effect("returns 404 when a non-initialize request omits the MCP session id", () => @@ -345,7 +345,7 @@ describe("McpServer", () => { it.effect("validates supplied protocol versions on POST", () => Effect.gen(function*() { - const { client, httpClient } = yield* makeTestClient + const { client, httpClient } = yield* makeRouterTestClient(HttpRouter.cors()) yield* client.initialize(initializePayload) @@ -356,6 +356,7 @@ describe("McpServer", () => { ) strictEqual(unsupportedResponse.status, 400) strictEqual(yield* unsupportedResponse.text, "") + strictEqual(unsupportedResponse.headers["access-control-allow-origin"], "*") const responseOnly = yield* HttpClientRequest.post("http://localhost/mcp").pipe( HttpClientRequest.bodyJsonUnsafe({ jsonrpc: "2.0", id: 1, result: {} }), From 34ddf6fc51ba25f746c8fa609eca8aa16b81278c Mon Sep 17 00:00:00 2001 From: Maxwell Brown Date: Sat, 25 Jul 2026 13:26:38 -0400 Subject: [PATCH 4/4] Organize MCP server lifecycle tests --- .../effect/test/unstable/ai/McpServer.test.ts | 21 +----- .../Lifecycle.test.ts} | 68 ++++--------------- .../test/unstable/ai/McpServer/utils.ts | 61 +++++++++++++++++ 3 files changed, 78 insertions(+), 72 deletions(-) rename packages/effect/test/unstable/ai/{McpServerInitialization.test.ts => McpServer/Lifecycle.test.ts} (81%) create mode 100644 packages/effect/test/unstable/ai/McpServer/utils.ts diff --git a/packages/effect/test/unstable/ai/McpServer.test.ts b/packages/effect/test/unstable/ai/McpServer.test.ts index 10b5d9eead7..bba259f9277 100644 --- a/packages/effect/test/unstable/ai/McpServer.test.ts +++ b/packages/effect/test/unstable/ai/McpServer.test.ts @@ -1,10 +1,7 @@ import { assert, describe, it } from "@effect/vitest" import { assertTrue, strictEqual } from "@effect/vitest/utils" import * as Effect from "effect/Effect" -import { constVoid } from "effect/Function" import * as Layer from "effect/Layer" -import * as Logger from "effect/Logger" -import * as References from "effect/References" import * as Schema from "effect/Schema" import * as AiError from "effect/unstable/ai/AiError" import * as McpSchema from "effect/unstable/ai/McpSchema" @@ -17,6 +14,7 @@ import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest" import * as HttpRouter from "effect/unstable/http/HttpRouter" import { RpcSerialization } from "effect/unstable/rpc" import * as RpcClient from "effect/unstable/rpc/RpcClient" +import { makeServerLayer, makeWebHandler } from "./McpServer/utils.ts" const OptionalStringTool = Tool.make("OptionalStringTool", { parameters: Schema.Struct({ signature: Schema.optional(Schema.String) }), @@ -48,18 +46,7 @@ const testToolkitHandlers = TestToolkit.of({ const INTERNAL_TOOL_ERROR_MESSAGE = "Tool execution failed due to an internal server error." -const noopLogger = Logger.make(constVoid) - -const TestServerLayer = McpServer.layerHttp({ - name: "TestServer", - version: "1.0.0", - path: "/mcp" -}).pipe( - Layer.provideMerge(Layer.succeed( - References.CurrentLoggers, - new Set([noopLogger]) - )) -) +const TestServerLayer = makeServerLayer({ name: "TestServer" }) const initializePayload = { protocolVersion: "2025-06-18", @@ -84,9 +71,7 @@ const makeTestClientWith = Effect.fnUntraced(function*( } | undefined ) { const responses: Array = [] - const appLayer = options?.routerLayer ? Layer.merge(serverLayer, options.routerLayer) : serverLayer - const { handler, dispose } = HttpRouter.toWebHandler(appLayer, { disableLogger: true }) - yield* Effect.addFinalizer(() => Effect.promise(() => dispose())) + const handler = yield* makeWebHandler(serverLayer, options) let sessionId: string | null = null const customFetch: typeof fetch = async (input, init) => { diff --git a/packages/effect/test/unstable/ai/McpServerInitialization.test.ts b/packages/effect/test/unstable/ai/McpServer/Lifecycle.test.ts similarity index 81% rename from packages/effect/test/unstable/ai/McpServerInitialization.test.ts rename to packages/effect/test/unstable/ai/McpServer/Lifecycle.test.ts index bf5d5d0ca78..e321aa9a176 100644 --- a/packages/effect/test/unstable/ai/McpServerInitialization.test.ts +++ b/packages/effect/test/unstable/ai/McpServer/Lifecycle.test.ts @@ -1,56 +1,15 @@ import { assert, describe, it } from "@effect/vitest" import * as Effect from "effect/Effect" -import { constVoid } from "effect/Function" import * as Layer from "effect/Layer" -import * as Logger from "effect/Logger" -import * as References from "effect/References" import * as Schema from "effect/Schema" import * as McpSchema from "effect/unstable/ai/McpSchema" import * as McpServer from "effect/unstable/ai/McpServer" import * as Tool from "effect/unstable/ai/Tool" import * as Toolkit from "effect/unstable/ai/Toolkit" -import * as HttpRouter from "effect/unstable/http/HttpRouter" +import { makeRawHttpHarness, makeServerLayer } from "./utils.ts" -const noopLogger = Logger.make(constVoid) - -const makeServerLayer = (options?: { - readonly extensions?: Record<`${string}/${string}`, unknown> | undefined -}) => - McpServer.layerHttp({ - name: "LifecycleServer", - version: "1.0.0", - path: "/mcp", - extensions: options?.extensions - }).pipe( - Layer.provideMerge(Layer.succeed( - References.CurrentLoggers, - new Set([noopLogger]) - )) - ) - -const makeHarness = Effect.fnUntraced(function*( - serverLayer: Layer.Layer = makeServerLayer() -) { - const { dispose, handler } = HttpRouter.toWebHandler(serverLayer, { disableLogger: true }) - yield* Effect.addFinalizer(() => Effect.promise(() => dispose())) - - const post = (body: unknown, headers?: HeadersInit) => - Effect.promise(() => - handler( - new Request("http://localhost/mcp", { - method: "POST", - headers: { - accept: "application/json, text/event-stream", - "content-type": "application/json", - ...headers - }, - body: JSON.stringify(body) - }) - ) - ) - - return { post } as const -}) +const ServerLayer = makeServerLayer({ name: "LifecycleServer" }) +const makeHarness = makeRawHttpHarness(ServerLayer) const initializeRequest = (protocolVersion: string, id = 1) => ({ jsonrpc: "2.0", @@ -130,6 +89,7 @@ const FeaturesServerLayer = Layer.mergeAll( }) ).pipe( Layer.provide(makeServerLayer({ + name: "LifecycleServer", extensions: { "example/lifecycle": { enabled: true } } })) ) @@ -141,7 +101,7 @@ describe("McpServer initialization", () => { describe("1.1 Initialization", () => { it.effect("requires initialize to be the first request", () => Effect.gen(function*() { - const { post } = yield* makeHarness() + const { post } = yield* makeHarness const response = yield* post(pingRequest) assert.isAtLeast(response.status, 400) @@ -149,7 +109,7 @@ describe("McpServer initialization", () => { it.effect("rejects initialized notifications before initialize", () => Effect.gen(function*() { - const { post } = yield* makeHarness() + const { post } = yield* makeHarness const response = yield* post(initializedNotification) assert.isAtLeast(response.status, 400) @@ -157,7 +117,7 @@ describe("McpServer initialization", () => { it.effect("requires protocolVersion, capabilities, and clientInfo", () => Effect.gen(function*() { - const { post } = yield* makeHarness() + const { post } = yield* makeHarness const invalidParams = [ { capabilities: {}, @@ -191,7 +151,7 @@ describe("McpServer initialization", () => { it.effect("returns server capabilities and implementation information", () => Effect.gen(function*() { - const { post } = yield* makeHarness() + const { post } = yield* makeHarness const { message, response } = yield* initialize(post, "2025-11-25") assert.strictEqual(response.status, 200) @@ -208,7 +168,7 @@ describe("McpServer initialization", () => { it.effect("accepts initialized after a successful initialize response", () => Effect.gen(function*() { - const { post } = yield* makeHarness() + const { post } = yield* makeHarness const initialized = yield* initialize(post, "2025-11-25") const sessionId = initialized.response.headers.get("Mcp-Session-Id") assert.isNotNull(sessionId) @@ -226,7 +186,7 @@ describe("McpServer initialization", () => { describe("1.1.1 Version Negotiation", () => { it.effect("echoes every requested version supported by the server", () => Effect.gen(function*() { - const { post } = yield* makeHarness() + const { post } = yield* makeHarness const supportedVersions = ["2025-06-18", "2025-03-26", "2024-11-05", "2024-10-07"] for (let i = 0; i < supportedVersions.length; i++) { @@ -238,7 +198,7 @@ describe("McpServer initialization", () => { it.effect("negotiates an unsupported requested version to the latest supported version", () => Effect.gen(function*() { - const { post } = yield* makeHarness() + const { post } = yield* makeHarness const { message } = yield* initialize(post, "2025-11-25") assert.strictEqual(message.result.protocolVersion, "2025-06-18") @@ -248,7 +208,7 @@ describe("McpServer initialization", () => { describe("1.1.2 Capability Negotiation", () => { it.effect("advertises the capabilities provided by the server", () => Effect.gen(function*() { - const { post } = yield* makeHarness(FeaturesServerLayer) + const { post } = yield* makeRawHttpHarness(FeaturesServerLayer) const { message } = yield* initialize(post, "2025-11-25") assert.deepStrictEqual(message.result.capabilities, { @@ -264,7 +224,7 @@ describe("McpServer initialization", () => { describe("1.2 Operation", () => { it.effect("continues to use the version negotiated during initialization", () => Effect.gen(function*() { - const { post } = yield* makeHarness() + const { post } = yield* makeHarness const initialized = yield* initialize(post, "2025-03-26") const sessionId = initialized.response.headers.get("Mcp-Session-Id") assert.isNotNull(sessionId) @@ -280,7 +240,7 @@ describe("McpServer initialization", () => { describe("3. Error Handling", () => { it.effect("handles protocol version mismatch through version negotiation", () => Effect.gen(function*() { - const { post } = yield* makeHarness() + const { post } = yield* makeHarness const { message } = yield* initialize(post, "invalid-version") assert.strictEqual(message.result.protocolVersion, "2025-06-18") diff --git a/packages/effect/test/unstable/ai/McpServer/utils.ts b/packages/effect/test/unstable/ai/McpServer/utils.ts new file mode 100644 index 00000000000..4872f159222 --- /dev/null +++ b/packages/effect/test/unstable/ai/McpServer/utils.ts @@ -0,0 +1,61 @@ +import * as Effect from "effect/Effect" +import { constVoid } from "effect/Function" +import * as Layer from "effect/Layer" +import * as Logger from "effect/Logger" +import * as References from "effect/References" +import * as McpServer from "effect/unstable/ai/McpServer" +import * as HttpRouter from "effect/unstable/http/HttpRouter" + +export const MCP_ENDPOINT = "http://localhost/mcp" + +const noopLogger = Logger.make(constVoid) + +export const makeServerLayer = (options: { + readonly name: string + readonly version?: string | undefined + readonly extensions?: Record<`${string}/${string}`, unknown> | undefined +}) => + McpServer.layerHttp({ + name: options.name, + version: options.version ?? "1.0.0", + path: "/mcp", + extensions: options.extensions + }).pipe( + Layer.provideMerge(Layer.succeed( + References.CurrentLoggers, + new Set([noopLogger]) + )) + ) + +export const makeWebHandler = Effect.fnUntraced(function*( + serverLayer: Layer.Layer, + options?: { + readonly routerLayer?: Layer.Layer | undefined + } +) { + const appLayer = options?.routerLayer ? Layer.merge(serverLayer, options.routerLayer) : serverLayer + const { dispose, handler } = HttpRouter.toWebHandler(appLayer, { disableLogger: true }) + yield* Effect.addFinalizer(() => Effect.promise(() => dispose())) + return handler +}) + +export const makeRawHttpHarness = Effect.fnUntraced(function*( + serverLayer: Layer.Layer +) { + const handler = yield* makeWebHandler(serverLayer) + const post = (body: unknown, headers?: HeadersInit) => + Effect.promise(() => + handler( + new Request(MCP_ENDPOINT, { + method: "POST", + headers: { + accept: "application/json, text/event-stream", + "content-type": "application/json", + ...headers + }, + body: JSON.stringify(body) + }) + ) + ) + return { post } as const +})