1import { assert, describe, it } from "@effect/vitest"
2import * as McpCore from "effect/ai/internal/mcpCore"
3import type * as McpProtocolInternal from "effect/ai/internal/mcpProtocol"
4import * as McpProtocol2026 from "effect/ai/internal/mcpProtocol/v2026_07_28"
5import * as McpSchema2026 from "effect/ai/internal/mcpSchema/v2026_07_28"
6import * as McpProtocol from "effect/ai/McpProtocol"
7import * as McpSchema from "effect/ai/McpSchema"
8import * as McpServer from "effect/ai/McpServer"
9import type * as Arr from "effect/Array"
10import * as Clock from "effect/Clock"
11import * as Context from "effect/Context"
12import * as Deferred from "effect/Deferred"
13import * as Effect from "effect/Effect"
14import * as Fiber from "effect/Fiber"
15import * as Layer from "effect/Layer"
16import * as Predicate from "effect/Predicate"
17import * as PubSub from "effect/PubSub"
18import * as Queue from "effect/Queue"
19import * as Rpc from "effect/rpc/Rpc"
20import type * as RpcMessage from "effect/rpc/RpcMessage"
21import { RequestId } from "effect/rpc/RpcMessage"
22import type * as RpcSerialization from "effect/rpc/RpcSerialization"
23import * as RpcServer from "effect/rpc/RpcServer"
24import * as Schema from "effect/Schema"
25import * as TestClock from "effect/testing/TestClock"
26import { makeHttpHarness } from "../TestUtils/McpHttpHarness.ts"
27import { makeMcpSseReader } from "../TestUtils/McpHttpResponse.ts"
28import { makeServerLayer } from "../TestUtils/McpServerLayer.ts"
29import { type JsonRpcMessage, makeMcpStdioHarness, type McpStdioHarness } from "../TestUtils/McpStdioHarness.ts"
30import { McpConformance, type McpConformanceLayer } from "./McpConformance.ts"
31
32const subscriptionIdKey = "io.modelcontextprotocol/subscriptionId"
33
34const paramsOf = (message: JsonRpcMessage): Record<string, unknown> => {
35 assert.isObject(message.params)
36 return message.params as Record<string, unknown>
37}
38
39const subscriptionIdOf = (message: JsonRpcMessage): string | number => {
40 const params = paramsOf(message)
41 assert.isObject(params._meta)
42 const subscriptionId = (params._meta as Record<string, unknown>)[subscriptionIdKey]
43 if (typeof subscriptionId === "string" || typeof subscriptionId === "number") {
44 return subscriptionId
45 }
46 return assert.fail("Expected a string or numeric subscription identifier")
47}
48
49const makeTool = (name: string) => ({
50 tool: new McpSchema.Tool({ name, inputSchema: { type: "object", properties: {} } }),
51 annotations: Context.empty(),
52 handle: () => Effect.succeed(new McpSchema.CallToolResult({ content: [] }))
53})
54
55const makePrompt = (name: string) => ({
56 prompt: new McpSchema.Prompt({ name }),
57 annotations: Context.empty(),
58 completions: {},
59 handle: () =>
60 Effect.succeed(
61 new McpSchema.GetPromptResult({
62 messages: [{ role: "user", content: { type: "text", text: name } }]
63 })
64 )
65})
66
67const makeResource = (uri: string) => ({
68 resource: new McpSchema.Resource({ uri, name: uri }),
69 annotations: Context.empty(),
70 handle: Effect.succeed(McpSchema.ReadResourceResult.make({ contents: [] }))
71})
72
73const subscriptionRegistrations = Layer.effectDiscard(
74 Effect.gen(function*() {
75 const server = yield* McpServer.McpServer
76 yield* server.addTool(makeTool("subscription-registration-tool"))
77 yield* server.addPrompt(makePrompt("subscription-registration-prompt"))
78 yield* server.addResource(makeResource("file:///subscription-registration-resource"))
79 })
80)
81
82const toolOnlyRegistration = Layer.effectDiscard(
83 McpServer.McpServer.use((server) => server.addTool(makeTool("tool-only-registration")))
84)
85
86const makeSubscriptionHarness = (
87 protocol: McpProtocol.ProtocolAdapter,
88 protocols: Arr.NonEmptyReadonlyArray<McpProtocol.ProtocolAdapter> = [protocol]
89) => makeMcpStdioHarness(protocol, protocols, subscriptionRegistrations)
90
91const listen = (
92 fixture: McpStdioHarness,
93 id: string | number,
94 notifications: Record<string, unknown>
95) =>
96 fixture.flushListChanged.pipe(
97 Effect.andThen(fixture.startRequest("subscriptions/listen", { notifications }, id))
98 )
99
100const assertAcknowledged = (
101 message: JsonRpcMessage,
102 id: string | number,
103 notifications: Record<string, unknown>
104) => {
105 assert.strictEqual(message.method, "notifications/subscriptions/acknowledged")
106 assert.notProperty(message, "id")
107 assert.strictEqual(subscriptionIdOf(message), id)
108 assert.deepStrictEqual(paramsOf(message).notifications, notifications)
109}
110
111const httpMetadata = (protocol: McpProtocol.ProtocolAdapter) => ({
112 "io.modelcontextprotocol/protocolVersion": protocol.protocolVersion,
113 "io.modelcontextprotocol/clientCapabilities": {},
114 "io.modelcontextprotocol/clientInfo": { name: "subscription-http-client", version: "1.0.0" }
115})
116
117const httpListenRequest = (protocol: McpProtocol.ProtocolAdapter, id: string | number) => ({
118 jsonrpc: "2.0",
119 id,
120 method: "subscriptions/listen",
121 params: {
122 notifications: { toolsListChanged: true },
123 _meta: httpMetadata(protocol)
124 }
125})
126
127const httpHeaders = (protocol: McpProtocol.ProtocolAdapter): HeadersInit => ({
128 "MCP-Protocol-Version": protocol.protocolVersion,
129 "Mcp-Method": "subscriptions/listen"
130})
131
132const makeHttpSubscriptionHarness = Effect.fnUntraced(function*(
133 protocol: McpProtocol.ProtocolAdapter,
134 clockLayer: Layer.Layer<never> = Layer.empty
135) {
136 const serverReady = yield* Deferred.make<McpServer.McpServer["Service"]>()
137 const registrations = Layer.effectDiscard(
138 Effect.gen(function*() {
139 const server = yield* McpServer.McpServer
140 yield* server.addTool(makeTool("http-subscription-baseline"))
141 yield* Deferred.succeed(serverReady, server)
142 })
143 )
144 const harness = yield* makeHttpHarness(
145 registrations.pipe(
146 Layer.provideMerge(makeServerLayer({
147 name: "SubscriptionConformance",
148 protocols: [protocol]
149 })),
150 Layer.provideMerge(clockLayer)
151 )
152 )
153 return { harness, serverReady }
154})
155
156const makeTrackingClock = Effect.gen(function*() {
157 const clock = yield* Clock.Clock
158 let activeSleeps = 0
159 const trackingClock: Clock.Clock = {
160 ...clock,
161 sleep: (duration) =>
162 Effect.acquireUseRelease(
163 Effect.sync(() => {
164 activeSleeps++
165 }),
166 () => clock.sleep(duration),
167 () =>
168 Effect.sync(() => {
169 activeSleeps--
170 })
171 )
172 }
173 return { layer: Layer.succeed(Clock.Clock, trackingClock), activeSleeps: () => activeSleeps }
174})
175
176export const suite = (protocol: McpProtocol.ProtocolAdapter, layer: McpConformanceLayer) =>
177 it.layer(layer)(`Mcp Conformance (${protocol.protocolVersion})`, (it) => {
178 describe("Subscriptions", () => {
179 // SEP-2575 replaces unsolicited list-change notifications and resources/subscribe with subscriptions/listen.
180 // https://modelcontextprotocol.io/seps/2575-stateless-mcp
181 // https://modelcontextprotocol.io/specification/2026-07-28/basic/patterns/subscriptions
182 it.effect("should advertise supported subscription capabilities when features are registered", () =>
183 Effect.gen(function*() {
184 const test = yield* McpConformance
185 const discovered = yield* test.initialize({ server: "features" })
186
187 assert.deepStrictEqual(discovered.message.result.capabilities.tools, { listChanged: true })
188 assert.deepStrictEqual(discovered.message.result.capabilities.resources, {
189 listChanged: true,
190 subscribe: true
191 })
192 assert.deepStrictEqual(discovered.message.result.capabilities.prompts, { listChanged: true })
193 }))
194
195 it.effect("should update discovery and subscription capabilities after late registration", () =>
196 Effect.gen(function*() {
197 const fixture = yield* makeMcpStdioHarness(protocol)
198 const before = yield* fixture.initialize()
199 const beforeResult = before.result
200 if (!Predicate.isObject(beforeResult)) {
201 return assert.fail("Expected discovery result before late registration")
202 }
203 const beforeCapabilities = beforeResult.capabilities
204 if (!Predicate.isObject(beforeCapabilities)) {
205 return assert.fail("Expected discovery capabilities before late registration")
206 }
207 assert.notProperty(beforeCapabilities, "tools")
208
209 yield* fixture.server.addTool(makeTool("late-registration"))
210
211 const after = yield* fixture.initialize()
212 const afterResult = after.result
213 if (!Predicate.isObject(afterResult)) {
214 return assert.fail("Expected discovery result after late registration")
215 }
216 const afterCapabilities = afterResult.capabilities
217 if (!Predicate.isObject(afterCapabilities)) {
218 return assert.fail("Expected discovery capabilities after late registration")
219 }
220 assert.deepStrictEqual(afterCapabilities.tools, { listChanged: true })
221
222 const request = yield* listen(fixture, "late-registration-listener", { toolsListChanged: true })
223 assertAcknowledged(
224 yield* fixture.takeMessage,
225 "late-registration-listener",
226 { toolsListChanged: true }
227 )
228 yield* request.cancel("test complete")
229 }))
230
231 it.effect("should not advertise subscriptions and should reject listen when the transport cannot send notifications", () =>
232 Effect.gen(function*() {
233 const outbound = yield* Queue.unbounded<RpcMessage.FromServerEncoded>()
234 const disconnects = yield* Queue.unbounded<number>()
235 const writeRequest = yield* Deferred.make<
236 (clientId: number, message: RpcMessage.FromClientEncoded) => Effect.Effect<void>
237 >()
238 const transport = yield* RpcServer.Protocol.make((write) =>
239 Deferred.succeed(writeRequest, write).pipe(
240 Effect.as({
241 disconnects,
242 send: (_clientId, message) => Queue.offer(outbound, message).pipe(Effect.asVoid),
243 end: (_clientId) => Effect.void,
244 clientIds: Effect.succeed(new Set([0])),
245 initialMessage: Effect.succeedNone,
246 supportsAck: false,
247 supportsTransferables: false,
248 supportsSpanPropagation: false,
249 supportsNotifications: false,
250 codecFor: Schema.toCodecJson as RpcSerialization.CodecFor
251 })
252 )
253 )
254 yield* Effect.gen(function*() {
255 yield* Layer.build(
256 subscriptionRegistrations.pipe(
257 Layer.provideMerge(
258 McpServer.layer({
259 name: "SubscriptionConformance",
260 version: "1.0.0",
261 protocols: [protocol]
262 }).pipe(Layer.provide(Layer.succeed(RpcServer.Protocol, transport)))
263 )
264 )
265 )
266 return yield* Effect.never
267 }).pipe(Effect.scoped, Effect.forkScoped)
268 const send = yield* Deferred.await(writeRequest)
269 const metadata = httpMetadata(protocol)
270
271 yield* send(0, {
272 _tag: "Request",
273 id: 1,
274 tag: "server/discover",
275 payload: { _meta: metadata },
276 headers: []
277 })
278 const discovery = yield* Queue.take(outbound)
279 if (discovery._tag !== "Exit" || discovery.exit._tag !== "Success") {
280 return assert.fail("Expected successful discovery response")
281 }
282 assert(Predicate.isObject(discovery.exit.value))
283 assert(Predicate.isObject(discovery.exit.value.capabilities))
284 assert.deepStrictEqual(discovery.exit.value.capabilities.tools, { listChanged: false })
285 assert.deepStrictEqual(discovery.exit.value.capabilities.resources, {
286 listChanged: false,
287 subscribe: false
288 })
289 assert.deepStrictEqual(discovery.exit.value.capabilities.prompts, { listChanged: false })
290
291 yield* send(0, {
292 _tag: "Request",
293 id: 2,
294 tag: "subscriptions/listen",
295 payload: { notifications: { toolsListChanged: true }, _meta: metadata },
296 headers: []
297 })
298 const listenResponse = yield* Queue.take(outbound)
299 if (listenResponse._tag !== "Exit" || listenResponse.exit._tag !== "Failure") {
300 return assert.fail("Expected failed subscription response")
301 }
302 const failure = listenResponse.exit.cause.find((cause) => cause._tag === "Fail")
303 assert(failure !== undefined && failure._tag === "Fail")
304 assert(Predicate.isObject(failure.error))
305 assert.strictEqual(failure.error.code, McpSchema.METHOD_NOT_FOUND_ERROR_CODE)
306 }))
307
308 it.effect("should buffer matching events while the subscription acknowledgment is blocked", () =>
309 Effect.gen(function*() {
310 const core = yield* McpCore.make
311 const events = yield* PubSub.unbounded<McpProtocolInternal.CanonicalServerNotification>()
312 const released = yield* Deferred.make<void>()
313 const acknowledgmentStarted = yield* Deferred.make<void>()
314 const releaseAcknowledgment = yield* Deferred.make<void>()
315 const sent = yield* Queue.unbounded<McpProtocol.ProjectedNotification>()
316 const handlers = McpProtocol2026.makeHandlers(core, undefined, {
317 subscribeServerNotifications: Effect.acquireRelease(
318 PubSub.subscribe(events),
319 () => Deferred.succeed(released, undefined)
320 ),
321 sendNotification: (_protocolVersion, _clientId, notification) =>
322 Effect.gen(function*() {
323 if (notification.tag === McpSchema2026.SubscriptionsAcknowledgedNotification._tag) {
324 yield* Deferred.succeed(acknowledgmentStarted, undefined)
325 yield* Deferred.await(releaseAcknowledgment)
326 }
327 yield* Queue.offer(sent, notification)
328 }),
329 supportedVersions: [McpSchema2026.protocolVersion],
330 serverInfo: { name: "SubscriptionConformance", version: "1.0.0" },
331 registrationPresence: Effect.succeed({ tools: true, resources: false, prompts: false })
332 })
333 const listener = yield* handlers["subscriptions/listen"](
334 { notifications: { toolsListChanged: true }, _meta: httpMetadata(protocol) },
335 { client: new Rpc.ServerClient(1), requestId: RequestId("blocked-acknowledgment") }
336 ).pipe(Effect.scoped, Effect.forkScoped)
337
338 yield* Deferred.await(acknowledgmentStarted)
339 yield* PubSub.publish(events, {
340 notification: McpCore.ServerNotification.ToolsChanged({})
341 })
342 yield* Deferred.succeed(releaseAcknowledgment, undefined)
343
344 const acknowledgment = yield* Queue.take(sent)
345 const notification = yield* Queue.take(sent)
346 assert.strictEqual(acknowledgment.tag, McpSchema2026.SubscriptionsAcknowledgedNotification._tag)
347 assert.strictEqual(notification.tag, McpSchema2026.ToolListChangedNotification._tag)
348
349 yield* Fiber.interrupt(listener)
350 yield* Deferred.await(released)
351 }))
352
353 it.effect("should release the scoped subscription when a blocked listener is interrupted", () =>
354 Effect.gen(function*() {
355 const core = yield* McpCore.make
356 const events = yield* PubSub.unbounded<McpProtocolInternal.CanonicalServerNotification>()
357 const released = yield* Deferred.make<void>()
358 const waitingForEvent = yield* Deferred.make<void>()
359 const handlers = McpProtocol2026.makeHandlers(core, undefined, {
360 subscribeServerNotifications: Effect.acquireRelease(
361 PubSub.subscribe(events).pipe(
362 Effect.map((subscription) =>
363 new Proxy(subscription, {
364 get(target, property, receiver) {
365 if (property !== "subscription") {
366 return Reflect.get(target, property, receiver)
367 }
368 return new Proxy(target.subscription, {
369 get(backing, backingProperty, backingReceiver) {
370 if (backingProperty !== "poll") {
371 return Reflect.get(backing, backingProperty, backingReceiver)
372 }
373 return () => {
374 Deferred.doneUnsafe(waitingForEvent, Effect.void)
375 return backing.poll()
376 }
377 }
378 })
379 }
380 })
381 )
382 ),
383 () => Deferred.succeed(released, undefined)
384 ),
385 sendNotification: () => Effect.void,
386 supportedVersions: [McpSchema2026.protocolVersion],
387 serverInfo: { name: "SubscriptionConformance", version: "1.0.0" },
388 registrationPresence: Effect.succeed({ tools: true, resources: false, prompts: false })
389 })
390 const listener = yield* handlers["subscriptions/listen"](
391 { notifications: { toolsListChanged: true }, _meta: httpMetadata(protocol) },
392 { client: new Rpc.ServerClient(1), requestId: RequestId("blocked-listener") }
393 ).pipe(Effect.scoped, Effect.forkScoped)
394
395 yield* Deferred.await(waitingForEvent)
396 yield* Fiber.interrupt(listener)
397 yield* Deferred.await(released)
398 }))
399
400 it.effect("should acknowledge with the exact identifier when it is numeric or string-valued", () =>
401 Effect.gen(function*() {
402 const fixture = yield* makeSubscriptionHarness(protocol)
403 yield* fixture.server.addTool(makeTool("subscription-baseline"))
404 yield* fixture.initialize()
405
406 for (const id of [42, "subscription-42"] as const) {
407 const notifications = { toolsListChanged: true }
408 const request = yield* listen(fixture, id, notifications)
409 assertAcknowledged(yield* fixture.takeMessage, id, notifications)
410 yield* request.cancel("test complete")
411 }
412 }))
413
414 it.effect("should deliver a change notification when its kind is requested", () =>
415 Effect.gen(function*() {
416 const fixture = yield* makeSubscriptionHarness(protocol)
417 yield* fixture.server.addTool(makeTool("subscription-tool-baseline"))
418 yield* fixture.server.addPrompt(makePrompt("subscription-prompt-baseline"))
419 yield* fixture.initialize()
420 const id = "filtered-subscription"
421 const notifications = {
422 toolsListChanged: true,
423 resourcesListChanged: true
424 }
425 const request = yield* listen(fixture, id, notifications)
426 assertAcknowledged(yield* fixture.takeMessage, id, notifications)
427
428 yield* fixture.server.addPrompt(makePrompt("must-not-be-delivered"))
429 yield* fixture.server.addTool(makeTool("must-be-delivered"))
430 yield* fixture.flushListChanged
431
432 const toolsChanged = yield* fixture.takeMessage
433 assert.strictEqual(toolsChanged.method, "notifications/tools/list_changed")
434 assert.strictEqual(subscriptionIdOf(toolsChanged), id)
435
436 yield* fixture.server.addResource(makeResource("file:///list-change-sentinel"))
437 yield* fixture.flushListChanged
438 const resourcesChanged = yield* fixture.takeMessage
439 assert.strictEqual(resourcesChanged.method, "notifications/resources/list_changed")
440 assert.strictEqual(subscriptionIdOf(resourcesChanged), id)
441
442 yield* request.cancel()
443 }))
444
445 it.effect("should deliver a resource update when its URI is subscribed", () =>
446 Effect.gen(function*() {
447 const fixture = yield* makeSubscriptionHarness(protocol)
448 yield* fixture.initialize()
449 const id = "resource-uri-subscription"
450 const notifications = { resourceSubscriptions: ["file:///subscribed"] }
451 const request = yield* listen(fixture, id, notifications)
452 assertAcknowledged(yield* fixture.takeMessage, id, notifications)
453
454 yield* fixture.server.notifications["notifications/resources/updated"]({ uri: "file:///other" })
455 yield* fixture.server.notifications["notifications/resources/updated"]({ uri: "file:///subscribed" })
456 const resourceUpdated = yield* fixture.takeMessage
457 assert.strictEqual(resourceUpdated.method, "notifications/resources/updated")
458 assert.strictEqual(subscriptionIdOf(resourceUpdated), id)
459 assert.strictEqual(paramsOf(resourceUpdated).uri, "file:///subscribed")
460 yield* request.cancel()
461 }))
462
463 it.effect("should acknowledge only the supported subset when requested filters exceed server capabilities", () =>
464 Effect.gen(function*() {
465 const fixture = yield* makeMcpStdioHarness(protocol, [protocol], toolOnlyRegistration)
466 yield* fixture.initialize()
467 const request = yield* listen(fixture, "honored-subset", {
468 toolsListChanged: true,
469 promptsListChanged: true
470 })
471
472 assertAcknowledged(yield* fixture.takeMessage, request.id, { toolsListChanged: true })
473 yield* request.cancel()
474
475 const unsupported = yield* listen(fixture, "unsupported-filters", {
476 toolsListChanged: false,
477 promptsListChanged: true
478 })
479 assertAcknowledged(yield* fixture.takeMessage, unsupported.id, {})
480 yield* unsupported.cancel()
481 }))
482
483 it.effect("should deliver a matching event to each subscription", () =>
484 Effect.gen(function*() {
485 const fixture = yield* makeSubscriptionHarness(protocol)
486 yield* fixture.initialize()
487 const first = yield* listen(fixture, "fanout-first", { toolsListChanged: true })
488 const second = yield* listen(fixture, "fanout-second", { toolsListChanged: true })
489 assertAcknowledged(yield* fixture.takeMessage, first.id, { toolsListChanged: true })
490 assertAcknowledged(yield* fixture.takeMessage, second.id, { toolsListChanged: true })
491
492 yield* fixture.server.addTool(makeTool("fanout-event"))
493 yield* fixture.flushListChanged
494 const delivered = [yield* fixture.takeMessage, yield* fixture.takeMessage]
495 assert.deepStrictEqual(
496 delivered.map(subscriptionIdOf).sort(),
497 [first.id, second.id].sort()
498 )
499 yield* first.cancel()
500 yield* second.cancel()
501 }))
502
503 it.effect("should keep another subscription active when its peer is cancelled", () =>
504 Effect.gen(function*() {
505 const fixture = yield* makeSubscriptionHarness(protocol)
506 yield* fixture.server.addTool(makeTool("concurrent-tool-baseline"))
507 yield* fixture.initialize()
508 const tools = yield* listen(fixture, "tools-subscription", { toolsListChanged: true })
509 const resources = yield* listen(fixture, "resources-subscription", {
510 resourceSubscriptions: ["file:///sentinel"]
511 })
512 assertAcknowledged(yield* fixture.takeMessage, tools.id, { toolsListChanged: true })
513 assertAcknowledged(yield* fixture.takeMessage, resources.id, {
514 resourceSubscriptions: ["file:///sentinel"]
515 })
516
517 yield* fixture.server.addTool(makeTool("concurrent-tool-event"))
518 yield* fixture.flushListChanged
519 const toolsChanged = yield* fixture.takeMessage
520 assert.strictEqual(toolsChanged.method, "notifications/tools/list_changed")
521 assert.strictEqual(subscriptionIdOf(toolsChanged), tools.id)
522
523 yield* fixture.server.notifications["notifications/resources/updated"]({ uri: "file:///sentinel" })
524 const resourceUpdated = yield* fixture.takeMessage
525 assert.strictEqual(resourceUpdated.method, "notifications/resources/updated")
526 assert.strictEqual(subscriptionIdOf(resourceUpdated), resources.id)
527
528 yield* tools.cancel("only tools are no longer needed")
529 yield* fixture.sendRequest("ping", {})
530 yield* fixture.server.addTool(makeTool("cancelled-tools-must-not-receive"))
531 yield* fixture.flushListChanged
532 yield* fixture.server.notifications["notifications/resources/updated"]({ uri: "file:///sentinel" })
533
534 const surviving = yield* fixture.takeMessage
535 assert.strictEqual(surviving.method, "notifications/resources/updated")
536 assert.strictEqual(subscriptionIdOf(surviving), resources.id)
537 yield* resources.cancel()
538 }))
539
540 it.effect("should preserve notification metadata and own the subscription identifier", () =>
541 Effect.gen(function*() {
542 const fixture = yield* makeSubscriptionHarness(protocol)
543 yield* fixture.initialize()
544 const request = yield* listen(fixture, "authoritative-subscription", { toolsListChanged: true })
545 assertAcknowledged(yield* fixture.takeMessage, request.id, { toolsListChanged: true })
546
547 yield* fixture.server.notifications["notifications/tools/list_changed"]({
548 _meta: {
549 source: "metadata-sentinel",
550 [subscriptionIdKey]: "untrusted-subscription"
551 }
552 })
553 const notification = yield* fixture.takeMessage
554 const params = paramsOf(notification)
555 const metadata = params._meta
556 assert.strictEqual(notification.method, "notifications/tools/list_changed")
557 assert(Predicate.isObject(metadata))
558 assert.strictEqual(metadata.source, "metadata-sentinel")
559 assert.strictEqual(metadata[subscriptionIdKey], request.id)
560 yield* request.cancel()
561 }))
562
563 it.effect("should deliver modern changes only when an active subscription matches", () =>
564 Effect.gen(function*() {
565 const fixture = yield* makeSubscriptionHarness(protocol)
566 yield* fixture.initialize()
567 yield* fixture.server.addTool(makeTool("unsolicited-modern-tool"))
568 yield* fixture.flushListChanged
569
570 const request = yield* listen(fixture, "prompt-sentinel", { promptsListChanged: true })
571 assertAcknowledged(yield* fixture.takeMessage, request.id, { promptsListChanged: true })
572 yield* fixture.server.addPrompt(makePrompt("prompt-sentinel"))
573 yield* fixture.flushListChanged
574 const notification = yield* fixture.takeMessage
575 assert.strictEqual(notification.method, "notifications/prompts/list_changed")
576 assert.strictEqual(subscriptionIdOf(notification), request.id)
577 yield* request.cancel()
578 }))
579
580 it.effect("should preserve legacy notification delivery when modern and legacy adapters are configured", () =>
581 Effect.gen(function*() {
582 const legacy = McpProtocol.v2025_06_18
583 const fixture = yield* makeSubscriptionHarness(legacy, [protocol, legacy])
584 yield* fixture.server.addTool(makeTool("mixed-era-baseline"))
585 yield* fixture.initialize()
586 yield* fixture.sendRequest("tools/list", {})
587 yield* fixture.server.addTool(makeTool("mixed-era-legacy-event"))
588 yield* fixture.flushListChanged
589
590 const notification = yield* fixture.awaitOutboundMethod("notifications/tools/list_changed")
591 assert.notProperty(notification, "id")
592 assert.notProperty(paramsOf(notification), "_meta")
593 }))
594
595 it.effect("should keep idle HTTP subscriptions alive with periodic SSE comments", () =>
596 Effect.gen(function*() {
597 const clock = yield* makeTrackingClock
598 const { harness, serverReady } = yield* makeHttpSubscriptionHarness(
599 protocol,
600 clock.layer
601 )
602 yield* harness.post({
603 jsonrpc: "2.0",
604 id: "ready",
605 method: "server/discover",
606 params: { _meta: httpMetadata(protocol) }
607 }, { "MCP-Protocol-Version": protocol.protocolVersion, "Mcp-Method": "server/discover" })
608 yield* TestClock.adjust("1 second")
609 const response = yield* harness.post(httpListenRequest(protocol, "idle-http"), httpHeaders(protocol))
610 assert.match(response.headers.get("content-type") ?? "", /^text\/event-stream/)
611 assert.isNotNull(response.body)
612 const reader = response.body.getReader()
613 yield* Effect.addFinalizer(() => Effect.promise(() => reader.cancel()))
614 const read = Effect.promise(() => reader.read())
615 const decoder = new TextDecoder()
616 const acknowledgment = decoder.decode((yield* read).value)
617 assert.isTrue(acknowledgment.startsWith("data: "))
618 assertAcknowledged(JSON.parse(acknowledgment.slice(6)), "idle-http", { toolsListChanged: true })
619
620 const next = yield* read.pipe(Effect.forkChild)
621 yield* TestClock.adjust("14 seconds")
622 assert.isUndefined(next.pollUnsafe())
623 yield* TestClock.adjust("1 second")
624 assert.isDefined(next.pollUnsafe())
625 assert.strictEqual(decoder.decode((yield* Fiber.join(next)).value), ": keepalive\n\n")
626 const heartbeat = yield* read.pipe(Effect.forkChild)
627 yield* TestClock.adjust("15 seconds")
628 assert.strictEqual(decoder.decode((yield* Fiber.join(heartbeat)).value), ": keepalive\n\n")
629
630 const tools = yield* harness.post({
631 jsonrpc: "2.0",
632 id: "tools-during-subscription",
633 method: "tools/list",
634 params: { _meta: httpMetadata(protocol) }
635 }, { "MCP-Protocol-Version": protocol.protocolVersion, "Mcp-Method": "tools/list" })
636 assert.match(tools.headers.get("content-type") ?? "", /^application\/json/)
637 const listed = yield* Effect.promise(() => tools.json())
638 assert.strictEqual(listed.id, "tools-during-subscription")
639 assert.deepStrictEqual(listed.result.tools, [
640 { name: "http-subscription-baseline", inputSchema: { type: "object", properties: {} } }
641 ])
642 assert.strictEqual(clock.activeSleeps(), 1)
643
644 const server = yield* Deferred.await(serverReady)
645 yield* server.notifications["notifications/tools/list_changed"]({})
646 const notification = decoder.decode((yield* read).value)
647 assert.isTrue(notification.startsWith("data: "))
648 const message = JSON.parse(notification.slice(6))
649 assert.strictEqual(message.method, "notifications/tools/list_changed")
650 assert.strictEqual(subscriptionIdOf(message), "idle-http")
651 yield* Effect.promise(() => reader.cancel())
652 assert.strictEqual(clock.activeSleeps(), 0)
653 yield* TestClock.adjust("30 seconds")
654 assert.deepStrictEqual(yield* read, { done: true, value: undefined })
655 assert.strictEqual(clock.activeSleeps(), 0)
656 }))
657
658 it.effect("should stream acknowledgment before matching events when using HTTP", () =>
659 Effect.gen(function*() {
660 const { harness, serverReady } = yield* makeHttpSubscriptionHarness(protocol)
661 const id = "reusable-http-subscription"
662
663 const firstResponse = yield* harness.post(httpListenRequest(protocol, id), httpHeaders(protocol))
664 const server = yield* Deferred.await(serverReady)
665 const first = makeMcpSseReader(firstResponse)
666 assertAcknowledged(yield* first.take(), id, { toolsListChanged: true })
667 yield* server.addTool(makeTool("http-subscription-event"))
668 const notification = yield* first.take()
669 assert.strictEqual(notification.method, "notifications/tools/list_changed")
670 assert.strictEqual(subscriptionIdOf(notification), id)
671 yield* first.cancel
672 }))
673
674 it.effect("should close only the HTTP subscription stream when its pending backlog overflows", () =>
675 Effect.gen(function*() {
676 const clock = yield* makeTrackingClock
677 const { harness, serverReady } = yield* makeHttpSubscriptionHarness(protocol, clock.layer)
678 const response = yield* harness.post(
679 httpListenRequest(protocol, "overflowing-http-subscription"),
680 httpHeaders(protocol)
681 )
682 const server = yield* Deferred.await(serverReady)
683 const subscription = makeMcpSseReader(response)
684 assertAcknowledged(
685 yield* subscription.take(),
686 "overflowing-http-subscription",
687 { toolsListChanged: true }
688 )
689
690 for (let index = 0; index < 1_000; index++) {
691 yield* server.notifications["notifications/tools/list_changed"]({})
692 yield* Effect.yieldNow
693 }
694
695 const remaining = yield* subscription.drain()
696 assert.notInclude(remaining.map((message) => message.method), "notifications/cancelled")
697 assert.strictEqual(clock.activeSleeps(), 0)
698
699 const discovery = yield* harness.post({
700 jsonrpc: "2.0",
701 id: "server-remains-available",
702 method: "server/discover",
703 params: { _meta: httpMetadata(protocol) }
704 }, {
705 "MCP-Protocol-Version": protocol.protocolVersion,
706 "Mcp-Method": "server/discover"
707 })
708 assert.strictEqual(discovery.status, 200)
709 }))
710 })
711 })