Unified · Split
| @@ -1,6 +1,6 @@ | ||
| 1 | 1 | import { WorkerTransport, type WorkerTransportOptions } from "agents/mcp"; |
| 2 | 2 | import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; |
| 3 | −import { Data, Effect } from "effect"; | |
| 3 | +import { Data, Effect, Exit } from "effect"; | |
| 4 | 4 | |
| 5 | 5 | export class McpWorkerTransportError extends Data.TaggedError("McpWorkerTransportError")<{ |
| 6 | 6 | readonly cause: unknown; |
| @@ -64,19 +64,20 @@ const extractJsonRpcRequestIdKeys = async (request: Request): Promise<ReadonlyAr · 57 unmodified lines | ||
| 64 | 64 | const contentType = request.headers.get("content-type") ?? ""; |
| 65 | 65 | if (!contentType.includes("application/json")) return []; |
| 66 | 66 | |
| 67 | − try { | |
| 68 | − const parsed = (await request.clone().json()) as unknown; | |
| 69 | − const messages = Array.isArray(parsed) ? parsed : [parsed]; | |
| 70 | − return messages.flatMap((message) => { | |
| 71 | − if (!message || typeof message !== "object") return []; | |
| 72 | − const rpc = message as JsonRpcLike; | |
| 73 | − if (typeof rpc.method !== "string") return []; | |
| 74 | − const key = jsonRpcRequestIdKey(rpc.id); | |
| 75 | − return key ? [key] : []; | |
| 76 | − }); | |
| 77 | − } catch { | |
| 67 | + const parsed = await Effect.runPromiseExit( | |
| 68 | + Effect.tryPromise(() => request.clone().json()), | |
| 69 | + ); | |
| 70 | + if (Exit.isFailure(parsed)) { | |
| 78 | 71 | return []; |
| 79 | 72 | } |
| 73 | + const messages = Array.isArray(parsed.value) ? parsed.value : [parsed.value]; | |
| 74 | + return messages.flatMap((message) => { | |
| 75 | + if (!message || typeof message !== "object") return []; | |
| 76 | + const rpc = message as JsonRpcLike; | |
| 77 | + if (typeof rpc.method !== "string") return []; | |
| 78 | + const key = jsonRpcRequestIdKey(rpc.id); | |
| 79 | + return key ? [key] : []; | |
| 80 | + }); | |
| 80 | 81 | }; |
| 81 | 82 | |
| 82 | 83 | // Hard ceiling on how long a same-id JSON-RPC request will wait for an |
| @@ -109,9 +110,12 @@ export class JsonRpcRequestIdQueue { · 26 unmodified lines | ||
| 109 | 110 | this.inFlight.set(id, current); |
| 110 | 111 | } |
| 111 | 112 | |
| 113 | + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: promise queue must release in-flight ids after callback completion | |
| 112 | 114 | try { |
| 113 | 115 | if (previous.length > 0) { |
| 114 | − const settled = Promise.all(previous.map((p) => p.catch(() => undefined))); | |
| 116 | + const settled = Promise.all( | |
| 117 | + previous.map((p) => Effect.runPromise(Effect.ignore(Effect.tryPromise(() => p)))), | |
| 118 | + ); | |
| 115 | 119 | const timeout = new Promise<"timeout">((resolve) => |
| 116 | 120 | setTimeout(() => resolve("timeout"), this.previousTimeoutMs), |
| 117 | 121 | ); |
| @@ -183,9 +187,13 @@ export const makeMcpWorkerTransport = ( · 65 unmodified lines | ||
| 183 | 187 | return result.response; |
| 184 | 188 | }), |
| 185 | 189 | close: () => |
| 186 | − Effect.promise(() => transport.close().catch(() => undefined)).pipe( | |
| 190 | + Effect.ignore( | |
| 191 | + Effect.tryPromise({ | |
| 192 | + try: () => transport.close(), | |
| 193 | + catch: (cause) => new McpWorkerTransportError({ cause }), | |
| 194 | + }), | |
| 195 | + ).pipe( | |
| 187 | 196 | Effect.withSpan("mcp.worker_transport.close"), |
| 188 | − Effect.orDie, | |
| 189 | 197 | ), |
| 190 | 198 | } satisfies McpWorkerTransport; |
| 191 | 199 | }).pipe(Effect.withSpan("mcp.worker_transport.make")); |