Open History to see the last commit
1// Relay JSON-RPC verbatim; schema decoding would reject foreign payloads.
2// @effect-diagnostics globalJsonInEffect:off
3import * as Cause from "effect/Cause";
4import * as Effect from "effect/Effect";
5import * as Option from "effect/Option";
6import * as Fiber from "effect/Fiber";
7import * as Schema from "effect/Schema";
8import * as Scope from "effect/Scope";
9import * as Stream from "effect/Stream";
10import { discardResponseBody, responsePayloads } from "@t3tools/shared/mcpResponsePayloads";
11
12/** ACP mandates stdio MCP support; HTTP/SSE support varies. */
13
14export class AcpMcpBridgeError extends Schema.TaggedError<AcpMcpBridgeError>()(
15 "AcpMcpBridgeError",
16 { message: Schema.String, cause: Schema.optional(Schema.Defect()) },
17) {}
18
19const isAcpMcpBridgeError = Schema.is(AcpMcpBridgeError);
20
21const bridgeError = (cause: unknown): AcpMcpBridgeError =>
22 isAcpMcpBridgeError(cause)
23 ? cause
24 : new AcpMcpBridgeError({
25 message: cause instanceof Error ? cause.message : String(cause),
26 cause,
27 });
28
29const decodeJsonLine = Schema.decodeUnknownOption(Schema.fromJsonString(Schema.Unknown));
30
31interface JsonRpcEnvelope {
32 readonly id?: unknown;
33 readonly method?: unknown;
34 readonly error?: unknown;
35 readonly result?: unknown;
36}
37
38const MCP_PROTOCOL_VERSION = "2025-06-18";
39
40export interface AcpMcpStdioBridgeOptions {
41 readonly endpoint: string;
42 readonly authorization: string;
43 readonly input: NodeJS.ReadableStream;
44 readonly output: { write(chunk: string): unknown };
45 readonly fetchImplementation?: (url: string, init?: RequestInit) => Promise<Response>;
46}
47
48function asEnvelope(value: unknown): JsonRpcEnvelope | null {
49 return typeof value === "object" && value !== null ? (value as JsonRpcEnvelope) : null;
50}
51
52function protocolVersionOf(entry: unknown): string | null {
53 const result = asEnvelope(entry)?.result;
54 const version =
55 typeof result === "object" && result !== null
56 ? (result as { readonly protocolVersion?: unknown }).protocolVersion
57 : undefined;
58 return typeof version === "string" && version.length > 0 ? version : null;
59}
60
61function replyToServerRequest(envelope: JsonRpcEnvelope): unknown {
62 return envelope.method === "ping"
63 ? { jsonrpc: "2.0", id: envelope.id, result: {} }
64 : { jsonrpc: "2.0", id: envelope.id, error: { code: -32601, message: "Method not found" } };
65}
66
67export interface McpToolSessionOptions {
68 readonly endpoint: string;
69 readonly authorization: string;
70 readonly clientName: string;
71 readonly fetchImplementation?: (url: string, init?: RequestInit) => Promise<Response>;
72}
73
74export interface McpToolSession {
75 /** Resolves to the JSON-RPC `result` of `tools/call`, a `CallToolResult`. */
76 readonly call: (
77 tool: string,
78 args: Readonly<Record<string, unknown>>,
79 ) => Effect.Effect<unknown, AcpMcpBridgeError>;
80}
81
82/**
83 * One initialized MCP session over T3 Code's HTTP endpoint, terminated when
84 * the scope closes. Used by the ACP terminal fallback and `t3 thread`.
85 */
86export const mcpToolSession = (
87 options: McpToolSessionOptions,
88): Effect.Effect<McpToolSession, AcpMcpBridgeError, Scope.Scope> => {
89 const fetchImplementation = options.fetchImplementation ?? fetch;
90 let sessionId: string | null = null;
91 let protocolVersion: string | null = null;
92 let nextId = 0;
93
94 const headers = () => ({
95 accept: "application/json, text/event-stream",
96 authorization: options.authorization,
97 ...(sessionId === null ? {} : { "mcp-session-id": sessionId }),
98 ...(protocolVersion === null ? {} : { "mcp-protocol-version": protocolVersion }),
99 });
100
101 const send = Effect.fn("AcpMcpStdioBridge.send")(function* (
102 message: unknown,
103 ): Effect.fn.Return<ReadonlyArray<unknown>, AcpMcpBridgeError> {
104 const response = yield* Effect.tryPromise({
105 try: (signal) =>
106 fetchImplementation(options.endpoint, {
107 method: "POST",
108 headers: { "content-type": "application/json", ...headers() },
109 body: JSON.stringify(message),
110 signal,
111 }),
112 catch: bridgeError,
113 });
114 sessionId = response.headers.get("mcp-session-id") ?? sessionId;
115 if (!response.ok) {
116 yield* discardResponseBody(response);
117 return yield* new AcpMcpBridgeError({
118 message: `T3 Code MCP endpoint responded with HTTP ${response.status}.`,
119 });
120 }
121 const payloads: Array<unknown> = [];
122 yield* Stream.runForEach(responsePayloads(response, bridgeError), (payload) => {
123 protocolVersion = protocolVersionOf(payload) ?? protocolVersion;
124 const envelope = asEnvelope(payload);
125 if (envelope?.id !== undefined && typeof envelope.method === "string") {
126 return send(replyToServerRequest(envelope)).pipe(Effect.asVoid);
127 }
128 payloads.push(payload);
129 return Effect.void;
130 });
131 return payloads;
132 });
133
134 const request = Effect.fn("AcpMcpStdioBridge.request")(function* (
135 method: string,
136 params: unknown,
137 ): Effect.fn.Return<JsonRpcEnvelope, AcpMcpBridgeError> {
138 const id = `t3-cli-${nextId++}`;
139 const responses = yield* send({ jsonrpc: "2.0", id, method, params });
140 const envelope = asEnvelope(responses.find((entry) => asEnvelope(entry)?.id === id));
141 if (envelope === null || envelope.error !== undefined) {
142 return yield* new AcpMcpBridgeError({
143 message: `T3 Code MCP ${method} failed${envelope?.error === undefined ? "." : `: ${JSON.stringify(envelope.error)}`}`,
144 });
145 }
146 return envelope;
147 });
148
149 const terminate = Effect.suspend(() =>
150 sessionId === null
151 ? Effect.void
152 : Effect.tryPromise({
153 try: () =>
154 fetchImplementation(options.endpoint, { method: "DELETE", headers: headers() }),
155 catch: bridgeError,
156 }).pipe(Effect.flatMap(discardResponseBody), Effect.ignore),
157 );
158
159 // Registered before the handshake so a session the server opened is ended
160 // even when the handshake fails, and the handshake stays interruptible.
161 return Effect.gen(function* () {
162 yield* Effect.addFinalizer(() => terminate);
163 yield* request("initialize", {
164 protocolVersion: MCP_PROTOCOL_VERSION,
165 capabilities: {},
166 clientInfo: { name: options.clientName, version: "0.0.0" },
167 });
168 yield* send({ jsonrpc: "2.0", method: "notifications/initialized" });
169 return {
170 call: (tool, args) =>
171 request("tools/call", { name: tool, arguments: args }).pipe(
172 Effect.map((envelope) => envelope.result),
173 ),
174 } satisfies McpToolSession;
175 });
176};
177
178/**
179 * This is the terminal fallback for ACP agents that accept `mcpServers` in
180 * `session/new` but fail to expose those tools to their model. Compliant ACP
181 * agents continue to use the stdio bridge above.
182 */
183function callAcpMcpTool(options: {
184 readonly endpoint: string;
185 readonly authorization: string;
186 readonly tool: string;
187 readonly arguments: Readonly<Record<string, unknown>>;
188}): Effect.Effect<unknown, AcpMcpBridgeError> {
189 return mcpToolSession({ ...options, clientName: "t3-code-acp-cli" }).pipe(
190 Effect.flatMap((session) => session.call(options.tool, options.arguments)),
191 Effect.scoped,
192 );
193}
194
195export function runAcpMcpStdioBridge(options: AcpMcpStdioBridgeOptions): Effect.Effect<void> {
196 const fetchImplementation = options.fetchImplementation ?? fetch;
197 return Effect.gen(function* () {
198 let sessionId: string | null = null;
199 let protocolVersion: string | null = null;
200 const running = new Set<Fiber.Fiber<void>>();
201
202 const writeMessage = (message: unknown): Effect.Effect<void> =>
203 Effect.sync(() => {
204 options.output.write(`${JSON.stringify(message)}\n`);
205 });
206
207 const handleServerPayload = Effect.fn("AcpMcpStdioBridge.handleServerPayload")(function* (
208 payload: unknown,
209 ): Effect.fn.Return<void> {
210 for (const entry of Array.isArray(payload) ? payload : [payload]) {
211 protocolVersion = protocolVersionOf(entry) ?? protocolVersion;
212 yield* writeMessage(entry);
213 }
214 });
215
216 const respondWithError = (
217 requestId: unknown,
218 message: string,
219 code = -32603,
220 ): Effect.Effect<void> =>
221 writeMessage({
222 jsonrpc: "2.0",
223 id: requestId ?? null,
224 error: { code, message },
225 });
226
227 const forward: (line: string, envelope: JsonRpcEnvelope) => Effect.Effect<void> = Effect.fn(
228 "AcpMcpStdioBridge.forward",
229 )(
230 function* (line: string, envelope: JsonRpcEnvelope) {
231 const response = yield* Effect.tryPromise({
232 try: (signal) =>
233 fetchImplementation(options.endpoint, {
234 method: "POST",
235 headers: {
236 "content-type": "application/json",
237 accept: "application/json, text/event-stream",
238 authorization: options.authorization,
239 ...(sessionId === null ? {} : { "mcp-session-id": sessionId }),
240 ...(protocolVersion === null ? {} : { "mcp-protocol-version": protocolVersion }),
241 },
242 body: line,
243 signal,
244 }),
245 catch: bridgeError,
246 });
247 sessionId = response.headers.get("mcp-session-id") ?? sessionId;
248 if (!response.ok) {
249 if (envelope.id !== undefined) {
250 yield* respondWithError(
251 envelope.id,
252 `T3 Code MCP endpoint responded with HTTP ${response.status}.`,
253 );
254 }
255 return yield* discardResponseBody(response);
256 }
257 yield* Stream.runForEach(responsePayloads(response, bridgeError), handleServerPayload);
258 },
259 (effect, _line, envelope) =>
260 effect.pipe(
261 Effect.catchCause((cause) => {
262 if (envelope.id === undefined) return Effect.void;
263 const error = Cause.squash(cause);
264 return respondWithError(
265 envelope.id,
266 `T3 Code MCP bridge request failed: ${error instanceof Error ? error.message : String(error)}`,
267 );
268 }),
269 ),
270 );
271
272 // Order only session establishment; SSE responses must bypass it to avoid deadlock. Calls remain concurrent after initialization.
273 let sessionBarrier: Effect.Effect<void> = Effect.void;
274 let sessionBarrierPending = false;
275
276 const launch = Effect.fn("AcpMcpStdioBridge.launch")(function* (
277 task: Effect.Effect<void>,
278 ): Effect.fn.Return<void> {
279 const fiber = yield* Effect.forkChild(task);
280 running.add(fiber);
281 yield* Effect.forkChild(
282 Fiber.await(fiber).pipe(
283 Effect.andThen(
284 Effect.sync(() => {
285 running.delete(fiber);
286 }),
287 ),
288 ),
289 );
290 });
291
292 const dispatch = Effect.fn("AcpMcpStdioBridge.dispatch")(function* (
293 line: string,
294 ): Effect.fn.Return<void> {
295 if (line.trim().length === 0) return;
296 const parsed = decodeJsonLine(line);
297 if (Option.isNone(parsed)) {
298 return yield* respondWithError(null, "Parse error", -32700);
299 }
300 const envelope = asEnvelope(parsed.value);
301 if (envelope === null) return;
302 const isResponse = envelope.id !== undefined && ("result" in envelope || "error" in envelope);
303 const isSessionHandshake =
304 envelope.method === "initialize" || envelope.method === "notifications/initialized";
305 if (isResponse) {
306 return yield* launch(forward(line, envelope));
307 }
308 if (isSessionHandshake) {
309 const prior = sessionBarrier;
310 const fiber = yield* Effect.forkChild(prior.pipe(Effect.andThen(forward(line, envelope))));
311 running.add(fiber);
312 const barrier = Fiber.await(fiber).pipe(Effect.asVoid);
313 sessionBarrier = barrier;
314 sessionBarrierPending = true;
315 yield* Effect.forkChild(
316 Fiber.await(fiber).pipe(
317 Effect.andThen(
318 Effect.sync(() => {
319 running.delete(fiber);
320 if (sessionBarrier === barrier) sessionBarrierPending = false;
321 }),
322 ),
323 ),
324 );
325 return;
326 }
327 const task = sessionBarrierPending
328 ? sessionBarrier.pipe(Effect.andThen(forward(line, envelope)))
329 : forward(line, envelope);
330 return yield* launch(task);
331 });
332
333 // node:readline also splits on U+2028/U+2029, which JSON.stringify leaves raw inside strings.
334 yield* Stream.fromAsyncIterable(options.input, bridgeError).pipe(
335 Stream.map((chunk) => (typeof chunk === "string" ? Buffer.from(chunk) : chunk)),
336 Stream.decodeText(),
337 Stream.splitLines,
338 Stream.runForEach(dispatch),
339 Effect.ignoreCause,
340 );
341 yield* Fiber.awaitAll(running);
342 });
343}
344
345/** Avoid heavy imports on the ACP first-message path. */
346export async function runAcpMcpCliFastPath(
347 command: "acp-mcp-bridge" | "acp-mcp-call",
348 args: ReadonlyArray<string>,
349): Promise<void> {
350 const endpoint = process.env.T3_ACP_MCP_ENDPOINT;
351 const authorization = process.env.T3_ACP_MCP_AUTHORIZATION;
352 if (endpoint === undefined || authorization === undefined) {
353 process.stderr.write(`${command} requires T3_ACP_MCP_ENDPOINT and T3_ACP_MCP_AUTHORIZATION.\n`);
354 process.exitCode = 2;
355 return;
356 }
357 if (command === "acp-mcp-bridge") {
358 await Effect.runPromise(
359 runAcpMcpStdioBridge({
360 endpoint,
361 authorization,
362 input: process.stdin,
363 output: process.stdout,
364 }),
365 );
366 return;
367 }
368 const [tool, argumentsJson] = args;
369 if (tool === undefined || argumentsJson === undefined) {
370 process.stderr.write("acp-mcp-call requires <tool> and <arguments-json>.\n");
371 process.exitCode = 2;
372 return;
373 }
374 let parsed: unknown;
375 try {
376 parsed = JSON.parse(argumentsJson);
377 } catch {
378 process.stderr.write("acp-mcp-call arguments must be valid JSON.\n");
379 process.exitCode = 2;
380 return;
381 }
382 if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) {
383 process.stderr.write("acp-mcp-call arguments must be a JSON object.\n");
384 process.exitCode = 2;
385 return;
386 }
387 let result = await Effect.runPromise(
388 callAcpMcpTool({
389 endpoint,
390 authorization,
391 tool,
392 arguments: parsed as Record<string, unknown>,
393 }),
394 );
395 // Terminal callers see both fields, unlike native MCP clients that choose one.
396 if (
397 typeof result === "object" &&
398 result !== null &&
399 "structuredContent" in result &&
400 result.structuredContent !== undefined &&
401 "content" in result &&
402 Array.isArray(result.content)
403 ) {
404 const encoded = JSON.stringify(result.structuredContent);
405 result = {
406 ...result,
407 content: result.content.filter(
408 (part: unknown) =>
409 !(
410 typeof part === "object" &&
411 part !== null &&
412 "type" in part &&
413 part.type === "text" &&
414 "text" in part &&
415 part.text === encoded
416 ),
417 ),
418 };
419 }
420 process.stdout.write(`${JSON.stringify(result)}\n`);
421}