1import { assert } from "@effect/vitest"
2import * as Effect from "effect/Effect"
3import type { JsonRpcMessage } from "./McpStdioHarness.ts"
4
5export const readMcpHttpResponse = (response: Response): Effect.Effect<unknown> =>
6 Effect.promise(async () => {
7 if (!response.headers.get("content-type")?.includes("text/event-stream")) {
8 return response.json()
9 }
10 const messages = (await response.text())
11 .split("\n")
12 .filter((line) => line.startsWith("data: "))
13 .map((line) => JSON.parse(line.slice(6)))
14 return messages.at(-1)
15 })
16
17export const makeMcpSseReader = (response: Response) => {
18 assert.match(response.headers.get("content-type") ?? "", /^text\/event-stream(?:;|$)/)
19 const body = response.body
20 assert.isNotNull(body)
21 const reader = body.getReader()
22 const decoder = new TextDecoder()
23 let pending = ""
24 const takeOrEnd = Effect.fnUntraced(function*() {
25 while (true) {
26 const boundary = pending.indexOf("\n\n")
27 if (boundary !== -1) {
28 const event = pending.slice(0, boundary)
29 pending = pending.slice(boundary + 2)
30 const data = event.split("\n")
31 .filter((line) => line.startsWith("data:"))
32 .map((line) => line.slice(5).trimStart())
33 .join("\n")
34 if (data.length > 0) {
35 return JSON.parse(data) as JsonRpcMessage
36 }
37 continue
38 }
39 const chunk = yield* Effect.promise(() => reader.read())
40 if (chunk.done) {
41 return undefined
42 }
43 pending += decoder.decode(chunk.value, { stream: true }).replaceAll("\r\n", "\n")
44 }
45 })
46 const take = Effect.fnUntraced(function*() {
47 const message = yield* takeOrEnd()
48 assert.isDefined(message)
49 return message
50 })
51 return {
52 take,
53 drain: Effect.fnUntraced(function*() {
54 const messages: Array<JsonRpcMessage> = []
55 while (true) {
56 const message = yield* takeOrEnd()
57 if (message === undefined) {
58 return messages
59 }
60 messages.push(message)
61 }
62 }),
63 cancel: Effect.promise(() => reader.cancel())
64 }
65}