1// The bridge relays opaque JSON-RPC lines verbatim; schema-decoding foreign
2// payloads here would reject traffic it must pass through untouched.
3// @effect-diagnostics nodeBuiltinImport:off preferSchemaOverJson:off
4import * as NodeReadline from "node:readline";
5
6import * as Cause from "effect/Cause";
7import * as Effect from "effect/Effect";
8import * as Fiber from "effect/Fiber";
9import * as Stream from "effect/Stream";
10import { discardResponseBody, responsePayloads } from "@t3tools/shared/mcpResponsePayloads";
11
12/**
13 * Stdio-to-HTTP bridge for T3's MCP endpoint.
14 *
15 * ACP agents must support stdio MCP servers, while optional http/sse support
16 * is unevenly implemented. `t3 acp-mcp-bridge` runs as the stdio MCP server an
17 * ACP agent spawns and forwards each JSON-RPC line to T3's authenticated
18 * streamable-HTTP endpoint: single JSON responses and SSE streams are written
19 * back as newline-delimited JSON-RPC, notification acknowledgements (202/204)
20 * produce no output, and the `mcp-session-id` / negotiated protocol version
21 * are replayed on subsequent requests.
22 *
23 * This entry sits on the ACP first-message critical path (bin.ts fast-paths
24 * it before the CLI graph), so it imports only effect core modules; keep the
25 * import list lean when changing it.
26 */
27
28export class AcpMcpBridgeError extends Error {
29 readonly _tag = "AcpMcpBridgeError";
30}
31
32const bridgeError = (cause: unknown): AcpMcpBridgeError =>
33 cause instanceof AcpMcpBridgeError
34 ? cause
35 : new AcpMcpBridgeError(cause instanceof Error ? cause.message : String(cause), { cause });
36
37interface JsonRpcEnvelope {
38 readonly id?: unknown;
39 readonly method?: unknown;
40 readonly error?: unknown;
41 readonly result?: unknown;
42}
43
44const MCP_PROTOCOL_VERSION = "2025-06-18";
45
46export interface AcpMcpStdioBridgeOptions {
47 readonly endpoint: string;
48 readonly authorization: string;
49 readonly input: NodeJS.ReadableStream;
50 readonly output: { write(chunk: string): unknown };
51 readonly fetchImplementation?: (url: string, init?: RequestInit) => Promise<Response>;
52}
53
54function asEnvelope(value: unknown): JsonRpcEnvelope | null {
55 return typeof value === "object" && value !== null ? (value as JsonRpcEnvelope) : null;
56}
57
58function protocolVersionOf(entry: unknown): string | null {
59 const result = asEnvelope(entry)?.result;
60 const version =
61 typeof result === "object" && result !== null
62 ? (result as { readonly protocolVersion?: unknown }).protocolVersion
63 : undefined;
64 return typeof version === "string" && version.length > 0 ? version : null;
65}
66
67export interface AcpMcpToolCallOptions {
68 readonly endpoint: string;
69 readonly authorization: string;
70 readonly tool: string;
71 readonly arguments: Readonly<Record<string, unknown>>;
72 readonly fetchImplementation?: (url: string, init?: RequestInit) => Promise<Response>;
73}
74
75/**
76 * Call one MCP tool through a fresh authenticated HTTP session.
77 *
78 * This is the terminal fallback for ACP agents that accept `mcpServers` in
79 * `session/new` but fail to expose those tools to their model. Compliant ACP
80 * agents continue to use the stdio bridge above.
81 */
82export function callAcpMcpTool(
83 options: AcpMcpToolCallOptions,
84): Effect.Effect<unknown, AcpMcpBridgeError> {
85 const fetchImplementation = options.fetchImplementation ?? fetch;
86 return Effect.gen(function* () {
87 // The bridge is single-fibered at creation time; concurrent sends only
88 // read these after the sequential handshake, so plain locals suffice.
89 let sessionId: string | null = null;
90 let protocolVersion: string | null = null;
91
92 const send = (message: unknown): Effect.Effect<ReadonlyArray<unknown>, AcpMcpBridgeError> =>
93 Effect.gen(function* () {
94 const response = yield* Effect.tryPromise({
95 try: () =>
96 fetchImplementation(options.endpoint, {
97 method: "POST",
98 headers: {
99 "content-type": "application/json",
100 accept: "application/json, text/event-stream",
101 authorization: options.authorization,
102 ...(sessionId === null ? {} : { "mcp-session-id": sessionId }),
103 ...(protocolVersion === null ? {} : { "mcp-protocol-version": protocolVersion }),
104 },
105 body: JSON.stringify(message),
106 }),
107 catch: bridgeError,
108 });
109 sessionId = response.headers.get("mcp-session-id") ?? sessionId;
110 if (!response.ok) {
111 yield* discardResponseBody(response);
112 return yield* Effect.fail(
113 new AcpMcpBridgeError(`T3 Code MCP endpoint responded with HTTP ${response.status}.`),
114 );
115 }
116 const payloads = yield* Stream.runCollect(responsePayloads(response, bridgeError));
117 for (const payload of payloads) {
118 protocolVersion = protocolVersionOf(payload) ?? protocolVersion;
119 }
120 return payloads;
121 });
122
123 const initializeId = "t3-acp-cli-initialize";
124 const initialized = yield* send({
125 jsonrpc: "2.0",
126 id: initializeId,
127 method: "initialize",
128 params: {
129 protocolVersion: MCP_PROTOCOL_VERSION,
130 capabilities: {},
131 clientInfo: { name: "t3-code-acp-cli", version: "0.0.0" },
132 },
133 });
134 const initializeResponse = initialized.find((entry) => asEnvelope(entry)?.id === initializeId);
135 if (initializeResponse === undefined || asEnvelope(initializeResponse)?.error !== undefined) {
136 return yield* Effect.fail(
137 new AcpMcpBridgeError("T3 Code MCP endpoint rejected initialization."),
138 );
139 }
140 yield* send({ jsonrpc: "2.0", method: "notifications/initialized" });
141
142 const callId = "t3-acp-cli-tool-call";
143 const responses = yield* send({
144 jsonrpc: "2.0",
145 id: callId,
146 method: "tools/call",
147 params: { name: options.tool, arguments: options.arguments },
148 });
149 const response = responses.find((entry) => asEnvelope(entry)?.id === callId);
150 const envelope = asEnvelope(response);
151 if (envelope === null || envelope.error !== undefined) {
152 return yield* Effect.fail(
153 new AcpMcpBridgeError(
154 `T3 Code MCP tool call failed${envelope?.error === undefined ? "." : `: ${JSON.stringify(envelope.error)}`}`,
155 ),
156 );
157 }
158 return envelope.result;
159 });
160}
161
162export function runAcpMcpStdioBridge(options: AcpMcpStdioBridgeOptions): Effect.Effect<void> {
163 const fetchImplementation = options.fetchImplementation ?? fetch;
164 return Effect.gen(function* () {
165 // Dispatch happens synchronously on the single stream fiber, so plain
166 // locals carry the shared session state; forwards mutate them in
167 // completion order exactly like the wire does.
168 let sessionId: string | null = null;
169 let protocolVersion: string | null = null;
170 const running = new Set<Fiber.Fiber<void>>();
171
172 const writeMessage = (message: unknown): Effect.Effect<void> =>
173 Effect.sync(() => {
174 options.output.write(`${JSON.stringify(message)}\n`);
175 });
176
177 const handleServerPayload = (payload: unknown): Effect.Effect<void> =>
178 Effect.gen(function* () {
179 for (const entry of Array.isArray(payload) ? payload : [payload]) {
180 protocolVersion = protocolVersionOf(entry) ?? protocolVersion;
181 yield* writeMessage(entry);
182 }
183 });
184
185 const respondWithError = (
186 requestId: unknown,
187 message: string,
188 code = -32603,
189 ): Effect.Effect<void> =>
190 writeMessage({
191 jsonrpc: "2.0",
192 id: requestId ?? null,
193 error: { code, message },
194 });
195
196 const forward = (line: string, envelope: JsonRpcEnvelope): Effect.Effect<void> =>
197 Effect.gen(function* () {
198 const response = yield* Effect.tryPromise({
199 try: () =>
200 fetchImplementation(options.endpoint, {
201 method: "POST",
202 headers: {
203 "content-type": "application/json",
204 accept: "application/json, text/event-stream",
205 authorization: options.authorization,
206 ...(sessionId === null ? {} : { "mcp-session-id": sessionId }),
207 ...(protocolVersion === null ? {} : { "mcp-protocol-version": protocolVersion }),
208 },
209 body: line,
210 }),
211 catch: bridgeError,
212 });
213 sessionId = response.headers.get("mcp-session-id") ?? sessionId;
214 if (!response.ok) {
215 if (envelope.id !== undefined) {
216 yield* respondWithError(
217 envelope.id,
218 `T3 Code MCP endpoint responded with HTTP ${response.status}.`,
219 );
220 }
221 return yield* discardResponseBody(response);
222 }
223 yield* Stream.runForEach(responsePayloads(response, bridgeError), handleServerPayload);
224 }).pipe(
225 Effect.catchCause((cause) => {
226 if (envelope.id === undefined) return Effect.void;
227 const error = Cause.squash(cause);
228 return respondWithError(
229 envelope.id,
230 `T3 Code MCP bridge request failed: ${error instanceof Error ? error.message : String(error)}`,
231 );
232 }),
233 );
234
235 // Only session establishment is ordered; client responses to
236 // server-initiated SSE requests must bypass the barrier or they would
237 // deadlock the still-open HTTP exchange, and normal MCP calls stay
238 // concurrent after initialization.
239 let sessionBarrier: Effect.Effect<void> = Effect.void;
240 let sessionBarrierPending = false;
241
242 const launch = (task: Effect.Effect<void>): Effect.Effect<void> =>
243 Effect.gen(function* () {
244 const fiber = yield* Effect.forkChild(task);
245 running.add(fiber);
246 yield* Effect.forkChild(
247 Fiber.await(fiber).pipe(
248 Effect.andThen(
249 Effect.sync(() => {
250 running.delete(fiber);
251 }),
252 ),
253 ),
254 );
255 });
256
257 const dispatch = (line: string): Effect.Effect<void> =>
258 Effect.gen(function* () {
259 if (line.trim().length === 0) return;
260 let parsed: unknown;
261 try {
262 parsed = JSON.parse(line);
263 } catch {
264 return yield* respondWithError(null, "Parse error", -32700);
265 }
266 const envelope = asEnvelope(parsed);
267 if (envelope === null) return;
268 const isResponse =
269 envelope.id !== undefined && ("result" in envelope || "error" in envelope);
270 const isSessionHandshake =
271 envelope.method === "initialize" || envelope.method === "notifications/initialized";
272 if (isResponse) {
273 return yield* launch(forward(line, envelope));
274 }
275 if (isSessionHandshake) {
276 const prior = sessionBarrier;
277 const fiber = yield* Effect.forkChild(
278 prior.pipe(Effect.andThen(forward(line, envelope))),
279 );
280 running.add(fiber);
281 const barrier = Fiber.await(fiber).pipe(Effect.asVoid);
282 sessionBarrier = barrier;
283 sessionBarrierPending = true;
284 yield* Effect.forkChild(
285 Fiber.await(fiber).pipe(
286 Effect.andThen(
287 Effect.sync(() => {
288 running.delete(fiber);
289 if (sessionBarrier === barrier) sessionBarrierPending = false;
290 }),
291 ),
292 ),
293 );
294 return;
295 }
296 const task = sessionBarrierPending
297 ? sessionBarrier.pipe(Effect.andThen(forward(line, envelope)))
298 : forward(line, envelope);
299 return yield* launch(task);
300 });
301
302 const reader = NodeReadline.createInterface({ input: options.input });
303 yield* Stream.runForEach(Stream.fromAsyncIterable(reader, bridgeError), dispatch).pipe(
304 Effect.catchCause(() => Effect.void),
305 );
306 yield* Fiber.awaitAll(running);
307 });
308}
309
310/**
311 * Argv runner for `t3 acp-mcp-bridge` and `t3 acp-mcp-call`, shared by the
312 * fast-path dispatch in bin.ts and the full CLI's command handlers. Kept free
313 * of heavy imports: these commands run on the ACP first-message critical path.
314 */
315export async function runAcpMcpCliFastPath(
316 command: "acp-mcp-bridge" | "acp-mcp-call",
317 args: ReadonlyArray<string>,
318): Promise<void> {
319 const endpoint = process.env.T3_ACP_MCP_ENDPOINT;
320 const authorization = process.env.T3_ACP_MCP_AUTHORIZATION;
321 if (endpoint === undefined || authorization === undefined) {
322 process.stderr.write(`${command} requires T3_ACP_MCP_ENDPOINT and T3_ACP_MCP_AUTHORIZATION.\n`);
323 process.exitCode = 2;
324 return;
325 }
326 if (command === "acp-mcp-bridge") {
327 await Effect.runPromise(
328 runAcpMcpStdioBridge({
329 endpoint,
330 authorization,
331 input: process.stdin,
332 output: process.stdout,
333 }),
334 );
335 return;
336 }
337 const [tool, argumentsJson] = args;
338 if (tool === undefined || argumentsJson === undefined) {
339 process.stderr.write("acp-mcp-call requires <tool> and <arguments-json>.\n");
340 process.exitCode = 2;
341 return;
342 }
343 let parsed: unknown;
344 try {
345 parsed = JSON.parse(argumentsJson);
346 } catch {
347 process.stderr.write("acp-mcp-call arguments must be valid JSON.\n");
348 process.exitCode = 2;
349 return;
350 }
351 if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) {
352 process.stderr.write("acp-mcp-call arguments must be a JSON object.\n");
353 process.exitCode = 2;
354 return;
355 }
356 const result = await Effect.runPromise(
357 callAcpMcpTool({
358 endpoint,
359 authorization,
360 tool,
361 arguments: parsed as Record<string, unknown>,
362 }),
363 );
364 process.stdout.write(`${JSON.stringify(result)}\n`);
365}