1import { assert, describe, it } from "@effect/vitest"
2import type * as McpProtocol from "effect/ai/McpProtocol"
3import * as McpSchema from "effect/ai/McpSchema"
4import * as McpServer from "effect/ai/McpServer"
5import * as Context from "effect/Context"
6import * as Deferred from "effect/Deferred"
7import * as Effect from "effect/Effect"
8import * as Exit from "effect/Exit"
9import * as Fiber from "effect/Fiber"
10import * as Layer from "effect/Layer"
11import { initializeHttpSession, makeHttpHarness } from "../TestUtils/McpHttpHarness.ts"
12import { makeMcpSseReader } from "../TestUtils/McpHttpResponse.ts"
13import { makeServerLayer } from "../TestUtils/McpServerLayer.ts"
14import { makeMcpStdioHarness } from "../TestUtils/McpStdioHarness.ts"
15import { McpConformance, type McpConformanceLayer } from "./McpConformance.ts"
16
17export const suite = (protocol: McpProtocol.ProtocolAdapter, layer: McpConformanceLayer) =>
18 it.layer(layer)(`Mcp Conformance (${protocol.protocolVersion})`, (it) => {
19 describe("Utilities", () => {
20 describe("Ping", () => {
21 // https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/ping
22 it.effect("MUST respond to a client ping with an empty result", () =>
23 Effect.gen(function*() {
24 const test = yield* McpConformance
25 const initialized = yield* test.initialize()
26 yield* test.notifyInitialized(initialized)
27
28 const response = yield* test.ping(initialized, { id: 42 })
29 const message = yield* test.decodeResult(response)
30
31 assert.strictEqual(response.status, 200)
32 assert.strictEqual(message.id, 42)
33 assert.deepStrictEqual(message.result, {})
34 }))
35 })
36
37 describe("Cancellation", () => {
38 // https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/cancellation
39 it.effect("MUST not send a response to a cancellation notification", () =>
40 Effect.gen(function*() {
41 const test = yield* McpConformance
42 const initialized = yield* test.initialize()
43 yield* test.notifyInitialized(initialized)
44
45 const response = yield* test.send(initialized, {
46 jsonrpc: "2.0",
47 method: "notifications/cancelled",
48 params: {
49 requestId: "unknown-request",
50 reason: "No longer needed"
51 }
52 })
53
54 assert.strictEqual(response.status, 202)
55 assert.strictEqual(yield* Effect.promise(() => response.text()), "")
56 }))
57 it.effect("SHOULD stop work and suppress the response after cancellation", () =>
58 Effect.gen(function*() {
59 const entered = yield* Deferred.make<void>()
60 const release = yield* Deferred.make<void>()
61 const interrupted = yield* Deferred.make<void>()
62 yield* Effect.addFinalizer(() => Deferred.succeed(release, void 0))
63 const fixture = yield* makeMcpStdioHarness(protocol)
64 const cancelledRequestId = "cancelled-tool-call"
65 const pingRequestId = "post-cancellation-ping"
66
67 yield* fixture.server.addTool({
68 tool: new McpSchema.Tool({
69 name: "GatedTool",
70 inputSchema: { type: "object", properties: {} }
71 }),
72 annotations: Context.empty(),
73 handle: () =>
74 Deferred.succeed(entered, void 0).pipe(
75 Effect.andThen(Deferred.await(release)),
76 Effect.onInterrupt(() => Deferred.succeed(interrupted, void 0)),
77 Effect.as(
78 new McpSchema.CallToolResult({
79 content: [{ type: "text", text: "released" }]
80 })
81 )
82 )
83 })
84
85 yield* fixture.initialize()
86 yield* fixture.sendRaw({
87 jsonrpc: "2.0",
88 id: cancelledRequestId,
89 method: "tools/call",
90 params: { name: "GatedTool", arguments: {} }
91 })
92 yield* Deferred.await(entered)
93 yield* fixture.sendNotification("notifications/cancelled", {
94 requestId: cancelledRequestId,
95 reason: "No longer needed"
96 })
97 yield* Deferred.await(interrupted)
98 yield* fixture.sendRaw({
99 jsonrpc: "2.0",
100 id: pingRequestId,
101 method: "ping"
102 })
103
104 while (true) {
105 const message = yield* fixture.takeMessage
106 assert.notStrictEqual(message.id, cancelledRequestId)
107 if (message.id === pingRequestId) {
108 assert.deepStrictEqual(message.result, {})
109 break
110 }
111 }
112 }))
113 it.effect("SHOULD ignore cancellation for an unknown request identifier", () =>
114 Effect.gen(function*() {
115 const test = yield* McpConformance
116 const initialized = yield* test.initialize()
117 yield* test.notifyInitialized(initialized)
118
119 const response = yield* test.send(initialized, {
120 jsonrpc: "2.0",
121 method: "notifications/cancelled",
122 params: { requestId: 999 }
123 })
124
125 assert.strictEqual(response.status, 202)
126 assert.strictEqual(yield* Effect.promise(() => response.text()), "")
127 }))
128 it.effect("SHOULD allow a later request to reuse an unknown cancelled identifier", () =>
129 Effect.gen(function*() {
130 const fixture = yield* makeMcpStdioHarness(protocol)
131 yield* fixture.initialize()
132 yield* fixture.sendNotification("notifications/cancelled", {
133 requestId: "reused-id"
134 })
135
136 const reused = yield* fixture.sendRequest("ping", undefined, "reused-id").pipe(Effect.forkChild)
137 yield* Effect.yieldNow
138 const control = yield* fixture.sendRequest("ping", undefined, "control-id")
139 assert.deepStrictEqual(control.result, {})
140 yield* Effect.yieldNow
141
142 const reusedExit = reused.pollUnsafe()
143 assert(reusedExit !== undefined && Exit.isSuccess(reusedExit))
144 assert.deepStrictEqual(reusedExit.value.result, {})
145 }))
146 it.effect("SHOULD ignore cancellation for an already completed request identifier", () =>
147 Effect.gen(function*() {
148 const test = yield* McpConformance
149 const initialized = yield* test.initialize()
150 yield* test.notifyInitialized(initialized)
151 yield* test.ping(initialized, { id: 11 })
152
153 const response = yield* test.send(initialized, {
154 jsonrpc: "2.0",
155 method: "notifications/cancelled",
156 params: { requestId: 11 }
157 })
158
159 assert.strictEqual(response.status, 202)
160 assert.strictEqual(yield* Effect.promise(() => response.text()), "")
161 }))
162 })
163
164 describe("Progress", () => {
165 // https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/progress
166 // NOTE: Smoke test only. The client capability accepts this one-way notification,
167 // but McpServer does not expose an observer for its decoded payload.
168 it.effect("MUST accept string progress tokens", () =>
169 Effect.gen(function*() {
170 const test = yield* McpConformance
171 const initialized = yield* test.initialize()
172 yield* test.notifyInitialized(initialized)
173
174 const response = yield* test.send(initialized, {
175 jsonrpc: "2.0",
176 method: "notifications/progress",
177 params: {
178 progressToken: "task-1",
179 progress: 1
180 }
181 })
182
183 assert.strictEqual(response.status, 202)
184 assert.strictEqual(yield* Effect.promise(() => response.text()), "")
185 }))
186 // NOTE: Smoke test only. The client capability accepts this one-way notification,
187 // but McpServer does not expose an observer for its decoded payload.
188 it.effect("MUST accept numeric progress tokens", () =>
189 Effect.gen(function*() {
190 const test = yield* McpConformance
191 const initialized = yield* test.initialize()
192 yield* test.notifyInitialized(initialized)
193
194 const response = yield* test.send(initialized, {
195 jsonrpc: "2.0",
196 method: "notifications/progress",
197 params: {
198 progressToken: 12,
199 progress: 1
200 }
201 })
202
203 assert.strictEqual(response.status, 202)
204 assert.strictEqual(yield* Effect.promise(() => response.text()), "")
205 }))
206 // NOTE: Smoke test only. The client capability accepts this one-way notification,
207 // but McpServer does not expose an observer for its decoded payload.
208 it.effect("SCHEMA accepts the optional total", () =>
209 Effect.gen(function*() {
210 const test = yield* McpConformance
211 const initialized = yield* test.initialize()
212 yield* test.notifyInitialized(initialized)
213
214 const response = yield* test.send(initialized, {
215 jsonrpc: "2.0",
216 method: "notifications/progress",
217 params: {
218 progressToken: "task-with-total",
219 progress: 1,
220 total: 2
221 }
222 })
223
224 assert.strictEqual(response.status, 202)
225 assert.strictEqual(yield* Effect.promise(() => response.text()), "")
226 }))
227 })
228 })
229 })
230
231export const statelessModernSuite = (
232 protocol: McpProtocol.ProtocolAdapter,
233 layer: McpConformanceLayer
234) =>
235 it.layer(layer)(`Mcp Conformance (${protocol.protocolVersion})`, (it) => {
236 describe("Utilities > Stateless modern", () => {
237 describe("Cancellation", () => {
238 // https://modelcontextprotocol.io/specification/2026-07-28/basic/utilities/cancellation
239 it.effect("should send no response when a cancellation notification is received", () =>
240 Effect.gen(function*() {
241 const test = yield* McpConformance
242 const discovered = yield* test.initialize()
243 const response = yield* test.send(discovered, {
244 jsonrpc: "2.0",
245 method: "notifications/cancelled",
246 params: { requestId: "unknown-request", reason: "No longer needed" }
247 })
248
249 assert.strictEqual(response.status, 202)
250 assert.strictEqual(yield* Effect.promise(() => response.text()), "")
251 }))
252
253 // https://modelcontextprotocol.io/specification/2026-07-28/basic/utilities/cancellation
254 it.effect("should interrupt work and suppress its response when an active request is cancelled", () =>
255 Effect.gen(function*() {
256 const entered = yield* Deferred.make<void>()
257 const release = yield* Deferred.make<void>()
258 const interrupted = yield* Deferred.make<void>()
259 yield* Effect.addFinalizer(() => Deferred.succeed(release, void 0))
260 const fixture = yield* makeMcpStdioHarness(protocol)
261 const cancelledRequestId = "cancelled-tool-call"
262
263 yield* fixture.server.addTool({
264 tool: new McpSchema.Tool({
265 name: "ModernGatedTool",
266 inputSchema: { type: "object", properties: {} }
267 }),
268 annotations: Context.empty(),
269 handle: () =>
270 Deferred.succeed(entered, void 0).pipe(
271 Effect.andThen(Deferred.await(release)),
272 Effect.onInterrupt(() => Deferred.succeed(interrupted, void 0)),
273 Effect.as(new McpSchema.CallToolResult({ content: [{ type: "text", text: "released" }] }))
274 )
275 })
276
277 yield* fixture.initialize()
278 yield* fixture.startRequest("tools/call", {
279 name: "ModernGatedTool",
280 arguments: {}
281 }, cancelledRequestId)
282 yield* Deferred.await(entered)
283 yield* fixture.sendNotification("notifications/cancelled", {
284 requestId: cancelledRequestId,
285 reason: "No longer needed"
286 })
287 yield* Deferred.await(interrupted)
288
289 const control = yield* fixture.sendRequest("server/discover", {}, "post-cancellation-discover")
290 assert.strictEqual(control.id, "post-cancellation-discover")
291 }))
292
293 // https://modelcontextprotocol.io/specification/2026-07-28/basic/utilities/cancellation
294 it.effect("should allow a later request to reuse an identifier cancelled before it was active", () =>
295 Effect.gen(function*() {
296 const fixture = yield* makeMcpStdioHarness(protocol)
297 yield* fixture.initialize()
298 yield* fixture.sendNotification("notifications/cancelled", { requestId: "reused-id" })
299
300 const reused = yield* fixture.sendRequest("server/discover", {}, "reused-id")
301 assert.strictEqual(reused.id, "reused-id")
302 assert.property(reused, "result")
303 }))
304 })
305 })
306 })
307
308export const statefulLegacySuite = (protocol: McpProtocol.ProtocolAdapter, layer: McpConformanceLayer) =>
309 it.layer(layer, { excludeTestServices: true })(`Mcp Conformance (${protocol.protocolVersion})`, (it) => {
310 if (protocol.runtime.transport.jsonRpc.acceptsBatches) {
311 describe("Utilities > Cancellation > STDIO", () => {
312 // Cancellation recipients SHOULD stop work and not send the cancelled request's response.
313 // Batch-capable transports must still deliver the remaining requests' responses.
314 // https://modelcontextprotocol.io/specification/2025-03-26/basic/utilities/cancellation#behavior-requirements
315 // https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#stdio
316 it.effect("should preserve the remaining batch result when a sibling STDIO request is cancelled", () =>
317 Effect.gen(function*() {
318 const entered = yield* Deferred.make<void>()
319 const interrupted = yield* Deferred.make<void>()
320 const fixture = yield* makeMcpStdioHarness(protocol)
321 yield* fixture.server.addTool({
322 tool: new McpSchema.Tool({ name: "Wait", inputSchema: { type: "object" } }),
323 annotations: Context.empty(),
324 handle: () =>
325 Deferred.succeed(entered, void 0).pipe(
326 Effect.andThen(Effect.never),
327 Effect.onInterrupt(() => Deferred.succeed(interrupted, void 0))
328 )
329 })
330 yield* fixture.initialize()
331 yield* fixture.takeFrame
332 yield* fixture.sendRaw([
333 { jsonrpc: "2.0", id: 1, method: "tools/call", params: { name: "Wait" } },
334 { jsonrpc: "2.0", id: "1", method: "ping" }
335 ])
336 yield* Deferred.await(entered)
337 yield* fixture.sendNotification("notifications/cancelled", { requestId: 1 })
338 yield* Deferred.await(interrupted)
339 const response = yield* fixture.takeFrame.pipe(Effect.timeout("1 second"))
340 assert.deepStrictEqual([response].flat(), [{ jsonrpc: "2.0", id: "1", result: {} }])
341 }))
342 })
343 }
344 describe("Utilities > Cancellation > HTTP", () => {
345 const serverLayer = makeServerLayer({ name: "HttpCancellation", protocols: [protocol] })
346 // https://modelcontextprotocol.io/specification/2025-11-25/basic/utilities/cancellation
347 it.effect("should cancel only the owning request when HTTP sessions reuse a request ID", () =>
348 Effect.gen(function*() {
349 const interrupted = yield* Deferred.make<void>()
350 const otherInterrupted = yield* Deferred.make<void>()
351 const releaseOther = yield* Deferred.make<void>()
352 const registration = Layer.effectDiscard(Effect.gen(function*() {
353 const server = yield* McpServer.McpServer
354 for (const name of ["Wait", "Keep"] as const) {
355 yield* server.addTool({
356 tool: new McpSchema.Tool({ name, inputSchema: { type: "object" } }),
357 annotations: Context.empty(),
358 handle: () =>
359 server.notifications["notifications/message"]({ level: "error", data: name }).pipe(
360 Effect.andThen(name === "Wait" ? Effect.never : Deferred.await(releaseOther)),
361 Effect.onInterrupt(() =>
362 Deferred.succeed(name === "Wait" ? interrupted : otherInterrupted, void 0)
363 ),
364 Effect.as(new McpSchema.CallToolResult({ content: [{ type: "text", text: "kept" }] }))
365 )
366 })
367 }
368 }))
369 const harness = yield* makeHttpHarness(registration.pipe(Layer.provideMerge(serverLayer)))
370 const otherHeaders = yield* initializeHttpSession(harness, protocol)
371 const otherResponse = yield* harness.post({
372 jsonrpc: "2.0",
373 id: "wait",
374 method: "tools/call",
375 params: { name: "Keep", arguments: {} }
376 }, otherHeaders)
377 const otherStream = makeMcpSseReader(otherResponse)
378 yield* Effect.addFinalizer(() => otherStream.cancel)
379 assert.strictEqual((yield* otherStream.take()).method, "notifications/message")
380 const headers = yield* initializeHttpSession(harness, protocol)
381 const response = yield* harness.post({
382 jsonrpc: "2.0",
383 id: "wait",
384 method: "tools/call",
385 params: { name: "Wait", arguments: {} }
386 }, headers)
387 const stream = makeMcpSseReader(response)
388 yield* Effect.addFinalizer(() => stream.cancel)
389 assert.strictEqual((yield* stream.take()).method, "notifications/message")
390 const cancelled = yield* harness.post({
391 jsonrpc: "2.0",
392 method: "notifications/cancelled",
393 params: { requestId: "wait" }
394 }, headers)
395 assert.strictEqual(cancelled.status, 202)
396 yield* Deferred.await(interrupted).pipe(Effect.timeout("1 second"))
397 assert.deepStrictEqual(yield* stream.drain().pipe(Effect.timeout("1 second")), [])
398 yield* Deferred.succeed(releaseOther, void 0)
399 assert.deepStrictEqual(yield* otherStream.drain().pipe(Effect.timeout("1 second")), [{
400 jsonrpc: "2.0",
401 id: "wait",
402 result: { content: [{ type: "text", text: "kept" }] }
403 }])
404 assert.isFalse(yield* Deferred.isDone(otherInterrupted))
405 }))
406
407 it.effect("should close without a reply when a tool is cancelled before emitting HTTP output", () =>
408 Effect.gen(function*() {
409 const entered = yield* Deferred.make<void>()
410 const interrupted = yield* Deferred.make<void>()
411 const registration = Layer.effectDiscard(McpServer.McpServer.use((server) =>
412 server.addTool({
413 tool: new McpSchema.Tool({ name: "Wait", inputSchema: { type: "object" } }),
414 annotations: Context.empty(),
415 handle: () =>
416 Deferred.succeed(entered, void 0).pipe(
417 Effect.andThen(Effect.never),
418 Effect.onInterrupt(() => Deferred.succeed(interrupted, void 0))
419 )
420 })
421 ))
422 const harness = yield* makeHttpHarness(registration.pipe(Layer.provideMerge(serverLayer)))
423 const headers = yield* initializeHttpSession(harness, protocol)
424 const pending = yield* harness.post({
425 jsonrpc: "2.0",
426 id: "wait",
427 method: "tools/call",
428 params: { name: "Wait", arguments: {} }
429 }, headers).pipe(Effect.forkChild)
430 yield* Deferred.await(entered)
431 const cancelled = yield* harness.post({
432 jsonrpc: "2.0",
433 method: "notifications/cancelled",
434 params: { requestId: "wait" }
435 }, headers)
436 assert.strictEqual(cancelled.status, 202)
437 yield* Deferred.await(interrupted).pipe(Effect.timeout("1 second"))
438 const response = yield* Fiber.join(pending).pipe(Effect.timeout("1 second"))
439 assert.strictEqual(yield* Effect.promise(() => response.text()), "")
440 }))
441
442 if (protocol.runtime.transport.jsonRpc.acceptsBatches) {
443 // https://modelcontextprotocol.io/specification/2025-03-26/basic/utilities/cancellation
444 // https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#sending-messages-to-the-server
445 for (const cancelled of ["one", "all"] as const) {
446 it.effect(`should omit cancelled replies and preserve remaining results when cancellation affects ${cancelled} of the batch requests`, () =>
447 Effect.gen(function*() {
448 const release = yield* Deferred.make<void>()
449 const interrupted = yield* Deferred.make<void>()
450 const otherInterrupted = yield* Deferred.make<void>()
451 const registration = Layer.effectDiscard(Effect.gen(function*() {
452 const server = yield* McpServer.McpServer
453 for (const name of ["Cancel", "Keep"] as const) {
454 yield* server.addTool({
455 tool: new McpSchema.Tool({ name, inputSchema: { type: "object" } }),
456 annotations: Context.empty(),
457 handle: () =>
458 server.notifications["notifications/message"]({ level: "error", data: name }).pipe(
459 Effect.andThen(name === "Cancel" ? Effect.never : Deferred.await(release)),
460 Effect.onInterrupt(() =>
461 Deferred.succeed(name === "Cancel" ? interrupted : otherInterrupted, void 0)
462 ),
463 Effect.as(new McpSchema.CallToolResult({ content: [{ type: "text", text: name }] }))
464 )
465 })
466 }
467 }))
468 const harness = yield* makeHttpHarness(registration.pipe(Layer.provideMerge(
469 makeServerLayer({ name: "BatchCancellation", protocols: [protocol] })
470 )))
471 const headers = yield* initializeHttpSession(harness, protocol)
472 const response = yield* harness.post(
473 ["Cancel", "Keep"].map((name) => ({
474 jsonrpc: "2.0",
475 id: name === "Cancel" ? 1 : "1",
476 method: "tools/call",
477 params: { name, arguments: {} }
478 })),
479 headers
480 )
481 const stream = makeMcpSseReader(response)
482 yield* Effect.addFinalizer(() => stream.cancel)
483 assert.strictEqual((yield* stream.take()).method, "notifications/message")
484 assert.strictEqual((yield* stream.take()).method, "notifications/message")
485 yield* harness.post(
486 { jsonrpc: "2.0", method: "notifications/cancelled", params: { requestId: 1 } },
487 headers
488 )
489 yield* Deferred.await(interrupted).pipe(Effect.timeout("1 second"))
490 assert.isFalse(yield* Deferred.isDone(otherInterrupted))
491 if (cancelled === "all") {
492 yield* harness.post(
493 { jsonrpc: "2.0", method: "notifications/cancelled", params: { requestId: "1" } },
494 headers
495 )
496 yield* Deferred.await(otherInterrupted).pipe(Effect.timeout("1 second"))
497 } else {
498 yield* Deferred.succeed(release, void 0)
499 }
500 const messages = yield* stream.drain().pipe(Effect.timeout("1 second"))
501 assert.deepStrictEqual(
502 messages.flat(),
503 cancelled === "all" ? [] : [{
504 jsonrpc: "2.0",
505 id: "1",
506 result: { content: [{ type: "text", text: "Keep" }] }
507 }]
508 )
509 assert.strictEqual(yield* Deferred.isDone(otherInterrupted), cancelled === "all")
510 }))
511 }
512 }
513 })
514 })