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