1// MCP responses carry opaque JSON-RPC that callers relay verbatim.
2// @effect-diagnostics preferSchemaOverJson:off
3/**
4 * Reads the JSON-RPC payloads of a response from an MCP streamable-HTTP
5 * endpoint. Used on the ACP bridge fast path, so it imports only effect core.
6 *
7 * @module mcpResponsePayloads
8 */
9import * as Effect from "effect/Effect";
10import * as Stream from "effect/Stream";
11
12async function* sseDataLines(response: Response): AsyncGenerator<string> {
13 if (response.body === null) return;
14 const decoder = new TextDecoder();
15 let buffered = "";
16 for await (const chunk of response.body) {
17 buffered += decoder.decode(chunk as Uint8Array, { stream: true });
18 let separatorIndex = buffered.search(/\n\n|\r\n\r\n/u);
19 while (separatorIndex !== -1) {
20 const rawEvent = buffered.slice(0, separatorIndex);
21 buffered = buffered.slice(separatorIndex).replace(/^(?:\r?\n){2}/u, "");
22 const data = rawEvent
23 .split(/\r?\n/u)
24 .filter((line) => line.startsWith("data:"))
25 .map((line) => line.slice("data:".length).trimStart())
26 .join("\n");
27 if (data.length > 0) yield data;
28 separatorIndex = buffered.search(/\n\n|\r\n\r\n/u);
29 }
30 }
31}
32
33export const discardResponseBody = (response: Response): Effect.Effect<void> =>
34 Effect.promise(() => response.body?.cancel().catch(() => undefined) ?? Promise.resolve());
35
36/**
37 * Every JSON-RPC payload carried by a response, in arrival order: nothing for
38 * notification acknowledgements, each SSE `data:` event as it streams in, or
39 * the single JSON body.
40 */
41export function responsePayloads<E>(
42 response: Response,
43 onError: (cause: unknown) => E,
44): Stream.Stream<unknown, E> {
45 if (response.status === 202 || response.status === 204) {
46 return Stream.unwrap(discardResponseBody(response).pipe(Effect.as(Stream.empty)));
47 }
48 const contentType = response.headers.get("content-type") ?? "";
49 if (contentType.includes("text/event-stream")) {
50 return Stream.fromAsyncIterable(sseDataLines(response), onError).pipe(
51 Stream.mapEffect((data) => Effect.try({ try: () => JSON.parse(data), catch: onError })),
52 );
53 }
54 return Stream.unwrap(
55 Effect.tryPromise({ try: () => response.text(), catch: onError }).pipe(
56 Effect.flatMap((text) =>
57 text.trim().length === 0
58 ? Effect.succeed(Stream.empty)
59 : Effect.try({ try: () => JSON.parse(text) as unknown, catch: onError }).pipe(
60 Effect.map((payload) => Stream.fromIterable([payload])),
61 ),
62 ),
63 ),
64 );
65}