1// @effect-diagnostics globalJsonInEffect:off
2import * as Effect from "effect/Effect";
3import * as Schema from "effect/Schema";
4import * as Semaphore from "effect/Semaphore";
5import * as Stream from "effect/Stream";
6import type * as AcpSchema from "effect-acp/compat";
7
8import { responsePayloads } from "@t3tools/shared/mcpResponsePayloads";
9
10const MAX_CONNECTIONS = 16;
11const MAX_MESSAGE_BYTES = 8 * 1024 * 1024;
12
13interface JsonRpcEnvelope {
14 readonly id?: unknown;
15 readonly method?: unknown;
16 readonly error?: unknown;
17 readonly result?: unknown;
18}
19
20interface Connection {
21 readonly mutex: Semaphore.Semaphore;
22 sessionId: string | null;
23 protocolVersion: string | null;
24 nextRequestId: number;
25}
26
27export class AcpMcpOverAcpError extends Schema.TaggedError<AcpMcpOverAcpError>()(
28 "AcpMcpOverAcpError",
29 { message: Schema.String, cause: Schema.optional(Schema.Defect()) },
30) {}
31
32const isAcpMcpOverAcpError = Schema.is(AcpMcpOverAcpError);
33
34const bridgeError = (cause: unknown): AcpMcpOverAcpError =>
35 isAcpMcpOverAcpError(cause)
36 ? cause
37 : new AcpMcpOverAcpError({
38 message: cause instanceof Error ? cause.message : String(cause),
39 cause,
40 });
41
42const asEnvelope = (value: unknown): JsonRpcEnvelope | null =>
43 typeof value === "object" && value !== null && !Array.isArray(value)
44 ? (value as JsonRpcEnvelope)
45 : null;
46
47function protocolVersionOf(payload: unknown): string | null {
48 const result = asEnvelope(payload)?.result;
49 if (typeof result !== "object" || result === null || Array.isArray(result)) return null;
50 const protocolVersion = (result as { readonly protocolVersion?: unknown }).protocolVersion;
51 return typeof protocolVersion === "string" && protocolVersion.length > 0 ? protocolVersion : null;
52}
53
54function errorMessage(error: unknown): string {
55 if (typeof error !== "object" || error === null || Array.isArray(error)) {
56 return "The MCP server returned an error.";
57 }
58 const message = (error as { readonly message?: unknown }).message;
59 return typeof message === "string" && message.length > 0
60 ? message
61 : "The MCP server returned an error.";
62}
63
64export interface AcpMcpOverAcpBridgeOptions {
65 readonly endpoint: string;
66 readonly authorization: string;
67 readonly allocateConnectionId: Effect.Effect<string>;
68 readonly fetchImplementation?: (url: string, init?: RequestInit) => Promise<Response>;
69}
70
71export interface AcpMcpOverAcpBridge {
72 readonly connect: (
73 request: AcpSchema.ConnectMcpRequest,
74 ) => Effect.Effect<AcpSchema.ConnectMcpResponse, AcpMcpOverAcpError>;
75 readonly message: (
76 request: AcpSchema.MessageMcpRequest,
77 ) => Effect.Effect<AcpSchema.MessageMcpResponse, AcpMcpOverAcpError>;
78 readonly notification: (
79 request: AcpSchema.MessageMcpNotification,
80 ) => Effect.Effect<void, AcpMcpOverAcpError>;
81 readonly disconnect: (
82 request: AcpSchema.DisconnectMcpRequest,
83 ) => Effect.Effect<AcpSchema.DisconnectMcpResponse, AcpMcpOverAcpError>;
84 readonly dispose: Effect.Effect<void>;
85}
86
87export const makeAcpMcpOverAcpBridge = Effect.fn("AcpMcpOverAcpBridge.make")(function* (
88 options: AcpMcpOverAcpBridgeOptions,
89): Effect.fn.Return<AcpMcpOverAcpBridge> {
90 const fetchImplementation = options.fetchImplementation ?? fetch;
91 const connections = yield* Effect.sync(() => new Map<string, Connection>());
92
93 const connectionFor = (connectionId: string): Effect.Effect<Connection, AcpMcpOverAcpError> =>
94 Effect.suspend(() => {
95 const connection = connections.get(connectionId);
96 return connection === undefined
97 ? Effect.fail(
98 new AcpMcpOverAcpError({
99 message: `Unknown MCP-over-ACP connection "${connectionId}".`,
100 }),
101 )
102 : Effect.succeed(connection);
103 });
104
105 // Only initialization needs ordering; negotiated calls may wait independently.
106 const untilNegotiated = <A, E>(connection: Connection, effect: Effect.Effect<A, E>) =>
107 Effect.suspend(() =>
108 connection.protocolVersion === null ? connection.mutex.withPermits(1)(effect) : effect,
109 );
110
111 const send = (
112 connection: Connection,
113 message: unknown,
114 ): Effect.Effect<ReadonlyArray<unknown>, AcpMcpOverAcpError> =>
115 untilNegotiated(
116 connection,
117 Effect.gen(function* () {
118 const body = JSON.stringify(message);
119 if (Buffer.byteLength(body) > MAX_MESSAGE_BYTES) {
120 return yield* new AcpMcpOverAcpError({ message: "MCP-over-ACP message exceeds 8 MiB." });
121 }
122 const response = yield* Effect.tryPromise({
123 try: (signal) =>
124 fetchImplementation(options.endpoint, {
125 method: "POST",
126 signal,
127 headers: {
128 "content-type": "application/json",
129 accept: "application/json, text/event-stream",
130 authorization: options.authorization,
131 ...(connection.sessionId === null
132 ? {}
133 : { "mcp-session-id": connection.sessionId }),
134 ...(connection.protocolVersion === null
135 ? {}
136 : { "mcp-protocol-version": connection.protocolVersion }),
137 },
138 body,
139 }),
140 catch: bridgeError,
141 });
142 connection.sessionId = response.headers.get("mcp-session-id") ?? connection.sessionId;
143 if (!response.ok) {
144 yield* Effect.promise(
145 () => response.body?.cancel().catch(() => undefined) ?? Promise.resolve(),
146 );
147 return yield* new AcpMcpOverAcpError({
148 message: `T3 Code MCP endpoint responded with HTTP ${response.status}.`,
149 });
150 }
151 const payloads = [...(yield* Stream.runCollect(responsePayloads(response, bridgeError)))];
152 for (const payload of payloads) {
153 connection.protocolVersion = protocolVersionOf(payload) ?? connection.protocolVersion;
154 }
155 return payloads;
156 }).pipe(Effect.mapError(bridgeError)),
157 );
158
159 const disconnect = Effect.fn("mcpOverAcpBridge.disconnect")(function* (
160 request: AcpSchema.DisconnectMcpRequest,
161 ): Effect.fn.Return<AcpSchema.DisconnectMcpResponse, AcpMcpOverAcpError> {
162 const connection = yield* connectionFor(request.connectionId);
163 connections.delete(request.connectionId);
164 const sessionId = connection.sessionId;
165 if (sessionId !== null) {
166 const response = yield* Effect.tryPromise({
167 try: (signal) =>
168 fetchImplementation(options.endpoint, {
169 method: "DELETE",
170 signal,
171 headers: {
172 authorization: options.authorization,
173 "mcp-session-id": sessionId,
174 ...(connection.protocolVersion === null
175 ? {}
176 : { "mcp-protocol-version": connection.protocolVersion }),
177 },
178 }),
179 catch: bridgeError,
180 });
181 if (!response.ok && response.status !== 404) {
182 return yield* new AcpMcpOverAcpError({
183 message: `T3 Code MCP endpoint rejected disconnect with HTTP ${response.status}.`,
184 });
185 }
186 yield* Effect.promise(
187 () => response.body?.cancel().catch(() => undefined) ?? Promise.resolve(),
188 );
189 }
190 return {};
191 });
192
193 return {
194 connect: Effect.fn("mcpOverAcpBridge.connect")(function* (request) {
195 if (request.serverId !== "t3-code") {
196 return yield* new AcpMcpOverAcpError({
197 message: `Unknown ACP MCP server "${request.serverId}".`,
198 });
199 }
200 if (connections.size >= MAX_CONNECTIONS) {
201 return yield* new AcpMcpOverAcpError({ message: "Too many MCP-over-ACP connections." });
202 }
203 const connectionId = yield* options.allocateConnectionId;
204 connections.set(connectionId, {
205 mutex: yield* Semaphore.make(1),
206 sessionId: null,
207 protocolVersion: null,
208 nextRequestId: 0,
209 });
210 return { connectionId };
211 }),
212 message: Effect.fn("mcpOverAcpBridge.message")(function* (request) {
213 const connection = yield* connectionFor(request.connectionId);
214 const id = ++connection.nextRequestId;
215 const payloads = yield* send(connection, {
216 jsonrpc: "2.0",
217 id,
218 method: request.method,
219 ...(request.params == null ? {} : { params: request.params }),
220 });
221 const response = payloads.find((payload) => asEnvelope(payload)?.id === id);
222 const envelope = asEnvelope(response);
223 if (envelope === null) {
224 return yield* new AcpMcpOverAcpError({
225 message: "MCP server did not return a matching response.",
226 });
227 }
228 if (envelope.error !== undefined) {
229 return yield* new AcpMcpOverAcpError({ message: errorMessage(envelope.error) });
230 }
231 return (envelope.result ?? null) as AcpSchema.MessageMcpResponse;
232 }),
233 notification: Effect.fn("mcpOverAcpBridge.notification")(function* (request) {
234 const connection = yield* connectionFor(request.connectionId);
235 yield* send(connection, {
236 jsonrpc: "2.0",
237 method: request.method,
238 ...(request.params == null ? {} : { params: request.params }),
239 });
240 }),
241 disconnect,
242 dispose: Effect.suspend(() =>
243 Effect.forEach(
244 [...connections.keys()],
245 (connectionId) => disconnect({ connectionId }).pipe(Effect.ignore),
246 { discard: true },
247 ),
248 ),
249 };
250});