1import {
2 type EnvironmentId,
3 OrchestratorMcpFailure,
4 type PeerLinkError,
5 type ProviderInteractionMode,
6 type OrchestrationV2LinkedSender,
7 type RuntimeMode,
8} from "@t3tools/contracts";
9import * as Context from "effect/Context";
10import * as Effect from "effect/Effect";
11import * as Layer from "effect/Layer";
12import * as Ref from "effect/Ref";
13import * as Result from "effect/Result";
14import * as Schema from "effect/Schema";
15import * as HttpBody from "effect/http/HttpBody";
16import * as HttpClient from "effect/http/HttpClient";
17import * as HttpClientRequest from "effect/http/HttpClientRequest";
18
19import * as McpInvocationContext from "../mcp/McpInvocationContext.ts";
20import * as PeerLinks from "./PeerLinks.ts";
21
22/** The only MCP protocol version a T3 Code `/mcp` registers. */
23const PROTOCOL_VERSION = "2025-06-18";
24
25/** A title can be any length; the header carrying it stays well under HTTP header limits. */
26const SENDER_TITLE_MAX_LENGTH = 200;
27
28/** The limits a call carries to the peer: the calling agent's modes here. */
29export interface PeerCallLimits {
30 readonly runtimeMode: RuntimeMode;
31 readonly interactionMode: ProviderInteractionMode;
32}
33
34/**
35 * Calls a linked environment's T3 MCP tools as this environment's outside
36 * agent. Each call carries the calling agent's limits in `T3-Mode-Limit`, so
37 * the peer enforces the narrower of those and the link's own access.
38 */
39export class PeerMcpClient extends Context.Service<
40 PeerMcpClient,
41 {
42 /** Calls `tool` on the peer and decodes its structured result with `success`. */
43 readonly call: <S extends Schema.Decoder<unknown>>(input: {
44 readonly environmentId: EnvironmentId;
45 readonly tool: string;
46 readonly arguments: Record<string, unknown>;
47 readonly limits: PeerCallLimits;
48 /** The thread here the call is made for, sent as `T3-Sender-Thread`. */
49 readonly sender?: OrchestrationV2LinkedSender;
50 readonly success: S;
51 }) => Effect.Effect<S["Type"], OrchestratorMcpFailure>;
52 }
53>()("t3/peer/PeerMcpClient") {}
54
55const ToolCallResponse = Schema.Struct({
56 result: Schema.optional(
57 Schema.Struct({
58 content: Schema.Array(
59 Schema.Struct({ type: Schema.String, text: Schema.optional(Schema.String) }),
60 ),
61 structuredContent: Schema.optional(Schema.Unknown),
62 isError: Schema.optional(Schema.Boolean),
63 }),
64 ),
65 error: Schema.optional(Schema.Struct({ code: Schema.Number, message: Schema.String })),
66});
67const decodeToolCallResponse = Schema.decodeUnknownEffect(Schema.fromJsonString(ToolCallResponse));
68const decodeFailure = Schema.decodeUnknownOption(
69 Schema.fromJsonString(
70 Schema.Struct({ code: OrchestratorMcpFailure.fields.code, message: Schema.String }),
71 ),
72);
73
74/** A peer link problem, as the agent that asked hears it. */
75const fromLinkError = (error: PeerLinkError) =>
76 new OrchestratorMcpFailure({
77 code:
78 error.reason === "unknown_link"
79 ? "invalid_request"
80 : error.reason === "expired" || error.reason === "incompatible"
81 ? "capability_denied"
82 : "orchestration_error",
83 message: error.message,
84 });
85
86/** Not final, so a task already running there resumes once the beta is on there again. */
87const linkedEnvironmentsOffThere = new OrchestratorMcpFailure({
88 code: "orchestration_error",
89 message:
90 "The linked environment has Linked environments turned off. Turn it on in its Settings → Labs.",
91});
92
93const stoppedAnswering = new OrchestratorMcpFailure({
94 code: "orchestration_error",
95 message: "The linked environment stopped answering.",
96});
97
98/** The body of an MCP response: JSON, or a single server-sent event carrying JSON. */
99const responseJson = (contentType: string | undefined, text: string) =>
100 contentType?.includes("text/event-stream") === true
101 ? text
102 .split("\n")
103 .filter((line) => line.startsWith("data:"))
104 .map((line) => line.slice("data:".length).trim())
105 .join("")
106 : text;
107
108const make = Effect.gen(function* () {
109 const links = yield* PeerLinks.PeerLinks;
110 const httpClient = yield* HttpClient.HttpClient;
111 // One MCP session per peer and address, reused across calls. The peer keeps
112 // sessions in memory, so a restart there answers 404 and we start a new one.
113 const sessions = yield* Ref.make<ReadonlyMap<string, string>>(new Map());
114
115 const post = (url: string, token: string, body: unknown, headers: Record<string, string>) =>
116 httpClient.execute(
117 HttpClientRequest.post(`${url}/mcp`).pipe(
118 HttpClientRequest.setHeaders({
119 accept: "application/json, text/event-stream",
120 authorization: `Bearer ${token}`,
121 ...headers,
122 }),
123 HttpClientRequest.setBody(HttpBody.text(JSON.stringify(body), "application/json")),
124 ),
125 );
126
127 const openSession = Effect.fn("PeerMcpClient.openSession")(
128 function* (url: string, token: string, limitHeader: Record<string, string>) {
129 const initialized = yield* post(
130 url,
131 token,
132 {
133 jsonrpc: "2.0",
134 id: 0,
135 method: "initialize",
136 params: {
137 protocolVersion: PROTOCOL_VERSION,
138 capabilities: {},
139 clientInfo: { name: "t3code-peer", version: "1" },
140 },
141 },
142 limitHeader,
143 );
144 const sessionId = initialized.headers["mcp-session-id"];
145 if (initialized.status === 403) return yield* linkedEnvironmentsOffThere;
146 if (initialized.status !== 200 || sessionId === undefined) {
147 return yield* new OrchestratorMcpFailure({
148 code: initialized.status === 401 ? "capability_denied" : "orchestration_error",
149 message:
150 initialized.status === 401
151 ? "The linked environment no longer accepts this link. It may have been revoked there; link it again."
152 : `The linked environment refused to start an MCP session (HTTP ${initialized.status}).`,
153 });
154 }
155 yield* post(
156 url,
157 token,
158 { jsonrpc: "2.0", method: "notifications/initialized" },
159 { ...limitHeader, "mcp-session-id": sessionId, "mcp-protocol-version": PROTOCOL_VERSION },
160 );
161 return sessionId;
162 },
163 // Only the handshake is bounded: a forwarded tool call may wait for minutes.
164 (effect, url) => Effect.timeout(effect, PeerLinks.probeTimeout(url)),
165 );
166
167 const call: PeerMcpClient["Service"]["call"] = Effect.fn("PeerMcpClient.call")(
168 function* (input) {
169 const resolved = yield* links
170 .resolve(input.environmentId)
171 .pipe(Effect.mapError(fromLinkError));
172 const limitHeader = {
173 [McpInvocationContext.MODE_LIMIT_HEADER]: `${input.limits.runtimeMode}/${input.limits.interactionMode}`,
174 };
175 const senderHeader =
176 input.sender === undefined
177 ? {}
178 : {
179 [McpInvocationContext.SENDER_THREAD_HEADER]: [
180 input.sender.environmentId,
181 input.sender.threadId,
182 // Whole code points, so a cut never leaves half an emoji for encodeURIComponent.
183 Array.from(input.sender.title).slice(0, SENDER_TITLE_MAX_LENGTH).join(""),
184 ]
185 // encodeURIComponent throws on an unpaired surrogate, which a title can hold.
186 .map((part) => encodeURIComponent(part.toWellFormed()))
187 .join("/"),
188 };
189 const sessionKey = `${input.environmentId} ${resolved.url}`;
190 const callOnce = (sessionId: string) =>
191 post(
192 resolved.url,
193 resolved.token,
194 {
195 jsonrpc: "2.0",
196 id: 1,
197 method: "tools/call",
198 params: { name: input.tool, arguments: input.arguments },
199 },
200 {
201 ...limitHeader,
202 ...senderHeader,
203 "mcp-session-id": sessionId,
204 "mcp-protocol-version": PROTOCOL_VERSION,
205 },
206 );
207 const existing = (yield* Ref.get(sessions)).get(sessionKey);
208 const sessionId = existing ?? (yield* openSession(resolved.url, resolved.token, limitHeader));
209 // The initialized session outlives a tool request, including a long wait or cancellation.
210 if (existing === undefined) {
211 yield* Ref.update(sessions, (current) => new Map(current).set(sessionKey, sessionId));
212 }
213 let response = yield* callOnce(sessionId);
214 if (response.status === 404) {
215 const renewed = yield* openSession(resolved.url, resolved.token, limitHeader);
216 yield* Ref.update(sessions, (current) => new Map(current).set(sessionKey, renewed));
217 response = yield* callOnce(renewed);
218 }
219 if (response.status === 401) {
220 return yield* new OrchestratorMcpFailure({
221 code: "capability_denied",
222 message:
223 "The linked environment no longer accepts this link. It may have been revoked there; link it again.",
224 });
225 }
226 if (response.status === 403) return yield* linkedEnvironmentsOffThere;
227 if (response.status !== 200) {
228 return yield* new OrchestratorMcpFailure({
229 code: "orchestration_error",
230 message: `The linked environment answered HTTP ${response.status}.`,
231 });
232 }
233 const body = yield* decodeToolCallResponse(
234 responseJson(response.headers["content-type"], yield* response.text),
235 ).pipe(
236 Effect.mapError(
237 () =>
238 new OrchestratorMcpFailure({
239 code: "orchestration_error",
240 message: "The linked environment answered in an unexpected shape.",
241 }),
242 ),
243 );
244 if (body.error !== undefined) {
245 return Result.fail(
246 new OrchestratorMcpFailure({ code: "invalid_request", message: body.error.message }),
247 );
248 }
249 if (body.result === undefined) {
250 return yield* new OrchestratorMcpFailure({
251 code: "invalid_request",
252 message: "The linked environment returned no result.",
253 });
254 }
255 const text = body.result.content.map((part) => part.text ?? "").join("");
256 if (body.result.isError === true) {
257 // A refusal keeps the peer's own code, so the agent sees why it was refused.
258 const refused = decodeFailure(text);
259 return Result.fail(
260 refused._tag === "Some"
261 ? new OrchestratorMcpFailure({
262 code: refused.value.code,
263 message: refused.value.message,
264 })
265 : new OrchestratorMcpFailure({
266 code: "orchestration_error",
267 message: text || "The linked environment refused the call.",
268 }),
269 );
270 }
271 const success = Schema.fromJsonString(input.success);
272 return yield* Schema.decodeEffect(success)(text).pipe(
273 Effect.map(Result.succeed),
274 Effect.mapError(
275 () =>
276 new OrchestratorMcpFailure({
277 code: "orchestration_error",
278 message:
279 "The linked environment's answer did not match this version. Update both environments.",
280 }),
281 ),
282 );
283 },
284 (effect, input) =>
285 effect.pipe(
286 Effect.tapError((error) =>
287 links.recordOutcome(input.environmentId, { error: error.message }),
288 ),
289 // A refusal is still an answer: the link works, so only this call fails.
290 Effect.tap(() => links.recordOutcome(input.environmentId, { error: null })),
291 // Recorded above with the transport's own message; the agent hears it plainly.
292 Effect.catchTags({
293 HttpClientError: () => Effect.fail(stoppedAnswering),
294 TimeoutError: () => Effect.fail(stoppedAnswering),
295 }),
296 Effect.flatMap(Effect.fromResult),
297 ),
298 );
299
300 return PeerMcpClient.of({ call });
301});
302
303export const layer = Layer.effect(PeerMcpClient, make);