History

Route MCP transports through HttpClient layer

Rhys Sullivan committed 4 months ago

Commit db1ee2c1 parent f5906e36 files · +148 −6Browse files

Unified · Split

6 files
  1. packages/core/sdk/src/executor.ts +1 −0
  2. packages/core/sdk/src/plugin.ts +1 −0
  3. packages/plugins/mcp/src/sdk/connection.ts +101 −1
  4. packages/plugins/mcp/src/sdk/plugin.test.ts +32 −2
  5. packages/plugins/mcp/src/sdk/plugin.ts +12 −3
  6. packages/plugins/openapi/src/sdk/spec-blob.test.ts +1 −0
packages/core/sdk/src/executor.ts+1−0
@@ -1920,6 +1920,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 1919 unmodified lines
1920 .resolveTools({1920 .resolveTools({
1921 integration: rowToIntegration(integrationRow),1921 integration: rowToIntegration(integrationRow),
1922 config: decodeJsonColumn(integrationRow.config),1922 config: decodeJsonColumn(integrationRow.config),
1923 httpClientLayer: runtime.ctx.httpClientLayer,
1923 connection: ref,1924 connection: ref,
1924 template: existingRow ? AuthTemplateSlug.make(existingRow.template) : null,1925 template: existingRow ? AuthTemplateSlug.make(existingRow.template) : null,
1925 storage: runtime.storage,1926 storage: runtime.storage,
packages/core/sdk/src/plugin.ts+1−0
@@ -211,6 +211,7 @@ export interface ResolveToolsInput<TStore = unknown> { · 210 unmodified lines
211 * facades (e.g. a content-addressed spec blob) instead of inlining them211 * facades (e.g. a content-addressed spec blob) instead of inlining them
212 * in `config`. */212 * in `config`. */
213 readonly storage: TStore;213 readonly storage: TStore;
214 readonly httpClientLayer: Layer.Layer<HttpClient.HttpClient>;
214 /** The connection whose tools are being resolved. */215 /** The connection whose tools are being resolved. */
215 readonly connection: ConnectionRef;216 readonly connection: ConnectionRef;
216 /** Which of the integration's declared auth methods the connection binds217 /** Which of the integration's declared auth methods the connection binds
packages/plugins/mcp/src/sdk/connection.ts+101−1
@@ -2,8 +2,10 @@ import type { OAuthClientProvider } from "@modelcontextprotocol/sdk/client/auth. · 1 unmodified line
2import { Client } from "@modelcontextprotocol/sdk/client/index.js";2import { Client } from "@modelcontextprotocol/sdk/client/index.js";
3import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";3import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";
4import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";4import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
5import type { FetchLike } from "@modelcontextprotocol/sdk/shared/transport.js";
5import { CfWorkerJsonSchemaValidator } from "@modelcontextprotocol/sdk/validation/cfworker";6import { CfWorkerJsonSchemaValidator } from "@modelcontextprotocol/sdk/validation/cfworker";
6import { Effect, Predicate } from "effect";7import { Effect, Layer, Predicate, Stream } from "effect";
8import { HttpClient, HttpClientRequest } from "effect/unstable/http";
7 9
8// NOTE: `StdioClientTransport` is NOT imported eagerly. The upstream module10// NOTE: `StdioClientTransport` is NOT imported eagerly. The upstream module
9// (`@modelcontextprotocol/sdk/client/stdio.js`) touches `node:child_process`11// (`@modelcontextprotocol/sdk/client/stdio.js`) touches `node:child_process`
@@ -42,6 +44,7 @@ export type RemoteConnectorInput = Omit< · 32 unmodified lines
42 readonly headers?: Record<string, string>;44 readonly headers?: Record<string, string>;
43 readonly queryParams?: Record<string, string>;45 readonly queryParams?: Record<string, string>;
44 readonly authProvider?: OAuthClientProvider;46 readonly authProvider?: OAuthClientProvider;
47 readonly httpClientLayer?: Layer.Layer<HttpClient.HttpClient>;
45};48};
46 49
47export type StdioConnectorInput = McpStdioIntegrationConfig;50export type StdioConnectorInput = McpStdioIntegrationConfig;
@@ -60,6 +63,100 @@ const buildEndpointUrl = (endpoint: string, queryParams: Record<string, string>) · 12 unmodified lines
60 return url;63 return url;
61};64};
62 65
66type HttpMethod = Parameters<typeof HttpClientRequest.make>[0];
67const HTTP_METHODS = new Set<HttpMethod>([
68 "DELETE",
69 "GET",
70 "HEAD",
71 "OPTIONS",
72 "PATCH",
73 "POST",
74 "PUT",
75]);
76
77const httpMethodFrom = (method: string | undefined): HttpMethod => {
78 const normalized = (method ?? "GET").toUpperCase() as HttpMethod;
79 return HTTP_METHODS.has(normalized) ? normalized : "POST";
80};
81
82const headersFrom = (headers: HeadersInit | undefined): Headers =>
83 headers ? new Headers(headers) : new Headers();
84
85const recordFromHeaders = (headers: Headers): Record<string, string> =>
86 Object.fromEntries(headers.entries());
87
88const applyBody = async (
89 request: HttpClientRequest.HttpClientRequest,
90 headers: Headers,
91 body: BodyInit | null | undefined,
92): Promise<HttpClientRequest.HttpClientRequest> => {
93 if (body == null) return request;
94 const contentType = headers.get("content-type") ?? undefined;
95 if (typeof body === "string") return HttpClientRequest.bodyText(request, body, contentType);
96 if (body instanceof URLSearchParams) {
97 return HttpClientRequest.bodyText(
98 request,
99 body.toString(),
100 contentType ?? "application/x-www-form-urlencoded;charset=UTF-8",
101 );
102 }
103 if (body instanceof Uint8Array)
104 return HttpClientRequest.bodyUint8Array(request, body, contentType);
105 if (body instanceof ArrayBuffer) {
106 return HttpClientRequest.bodyUint8Array(request, new Uint8Array(body), contentType);
107 }
108 const bytes = new Uint8Array(await new Response(body).arrayBuffer());
109 return HttpClientRequest.bodyUint8Array(request, bytes, contentType);
110};
111
112const abortError = (signal: AbortSignal): unknown => {
113 if (signal.reason !== undefined) return signal.reason;
114 // oxlint-disable-next-line executor/no-error-constructor -- boundary: Fetch-compatible adapter must reject with an AbortError-shaped value
115 const error = new Error("The operation was aborted");
116 error.name = "AbortError";
117 return error;
118};
119
120const fetchFromHttpClientLayer = (
121 httpClientLayer: Layer.Layer<HttpClient.HttpClient>,
122): FetchLike => {
123 const execute: FetchLike = async (url, init) => {
124 const headers = headersFrom(init?.headers);
125 const requestWithoutBody = HttpClientRequest.make(httpMethodFrom(init?.method))(url, {
126 headers: recordFromHeaders(headers),
127 });
128 const request = await applyBody(requestWithoutBody, headers, init?.body);
129 const effect = Effect.gen(function* () {
130 const client = yield* HttpClient.HttpClient;
131 const response = yield* client.execute(request);
132 const responseHeaders = new Headers();
133 for (const [key, value] of Object.entries(response.headers)) {
134 if (value !== undefined) responseHeaders.set(key, value);
135 }
136 const body =
137 response.status === 204 || response.status === 205 || response.status === 304
138 ? null
139 : Stream.toReadableStream(response.stream);
140 return new Response(body, {
141 status: response.status,
142 headers: responseHeaders,
143 });
144 }).pipe(Effect.provide(httpClientLayer));
145 const promise = Effect.runPromise(effect);
146 if (!init?.signal) return promise;
147 // oxlint-disable-next-line executor/no-promise-reject -- boundary: Fetch-compatible adapter mirrors abort rejection semantics
148 if (init.signal.aborted) return Promise.reject(abortError(init.signal));
149 const aborted = new Promise<never>((_, reject) => {
150 // oxlint-disable-next-line executor/no-promise-reject -- boundary: Fetch-compatible adapter races the Effect request against AbortSignal
151 init.signal?.addEventListener("abort", () => reject(abortError(init.signal!)), {
152 once: true,
153 });
154 });
155 return Promise.race([promise, aborted]);
156 };
157 return execute;
158};
159
63// Use the cfworker JSON Schema validator instead of the SDK's default160// Use the cfworker JSON Schema validator instead of the SDK's default
64// (Ajv). Ajv compiles schemas via `new Function(...)`, which throws161// (Ajv). Ajv compiles schemas via `new Function(...)`, which throws
65// `Code generation from strings disallowed for this context` when the162// `Code generation from strings disallowed for this context` when the
@@ -157,6 +254,7 @@ export const createMcpConnector = (input: ConnectorInput): McpConnector => { · 91 unmodified lines
157 const headers = input.headers ?? {};254 const headers = input.headers ?? {};
158 const remoteTransport = input.remoteTransport ?? "auto";255 const remoteTransport = input.remoteTransport ?? "auto";
159 const requestInit = Object.keys(headers).length > 0 ? { headers } : undefined;256 const requestInit = Object.keys(headers).length > 0 ? { headers } : undefined;
257 const fetch = input.httpClientLayer ? fetchFromHttpClientLayer(input.httpClientLayer) : undefined;
160 258
161 const endpoint = buildEndpointUrl(input.endpoint, input.queryParams ?? {});259 const endpoint = buildEndpointUrl(input.endpoint, input.queryParams ?? {});
162 260
@@ -166,6 +264,7 @@ export const createMcpConnector = (input: ConnectorInput): McpConnector => { · 3 unmodified lines
166 new StreamableHTTPClientTransport(endpoint, {264 new StreamableHTTPClientTransport(endpoint, {
167 requestInit,265 requestInit,
168 authProvider: input.authProvider,266 authProvider: input.authProvider,
267 fetch,
169 }),268 }),
170 });269 });
171 270
@@ -175,6 +274,7 @@ export const createMcpConnector = (input: ConnectorInput): McpConnector => { · 3 unmodified lines
175 new SSEClientTransport(endpoint, {274 new SSEClientTransport(endpoint, {
176 requestInit,275 requestInit,
177 authProvider: input.authProvider,276 authProvider: input.authProvider,
277 fetch,
178 }),278 }),
179 });279 });
180 280
packages/plugins/mcp/src/sdk/plugin.test.ts+32−2
@@ -1,6 +1,11 @@
1import { describe, expect, it } from "@effect/vitest";1import { describe, expect, it } from "@effect/vitest";
2import { Effect, Option, Predicate, Schema } from "effect";2import { Effect, Layer, Option, Predicate, Schema } from "effect";
3import { HttpServerResponse } from "effect/unstable/http";3import {
4 HttpClient,
5 HttpClientRequest,
6 HttpClientResponse,
7 HttpServerResponse,
8} from "effect/unstable/http";
4 9
5import {10import {
6 AuthTemplateSlug,11 AuthTemplateSlug,
@@ -17,6 +22,7 @@ import { · 10 unmodified lines
17 serveTestHttpApp,22 serveTestHttpApp,
18} from "@executor-js/sdk/testing";23} from "@executor-js/sdk/testing";
19 24
25import { createMcpConnector } from "./connection";
20import { mcpPlugin, userFacingProbeMessage } from "./plugin";26import { mcpPlugin, userFacingProbeMessage } from "./plugin";
21import { McpInvocationError } from "./errors";27import { McpInvocationError } from "./errors";
22import { extractManifestFromListToolsResult, deriveMcpNamespace, joinToolPath } from "./manifest";28import { extractManifestFromListToolsResult, deriveMcpNamespace, joinToolPath } from "./manifest";
@@ -299,6 +305,30 @@ describe("mcpPlugin", () => { · 276 unmodified lines
299 }),305 }),
300 );306 );
301 307
308 it.effect("routes remote connector traffic through the provided HttpClient layer", () =>
309 Effect.gen(function* () {
310 const seen: string[] = [];
311 const httpClientLayer = Layer.succeed(HttpClient.HttpClient)(
312 HttpClient.make((request: HttpClientRequest.HttpClientRequest) => {
313 seen.push(request.url);
314 return Effect.succeed(
315 HttpClientResponse.fromWeb(request, new Response("blocked", { status: 403 })),
316 );
317 }),
318 );
319
320 const error = yield* createMcpConnector({
321 transport: "remote",
322 endpoint: "https://internal.example/mcp",
323 remoteTransport: "streamable-http",
324 httpClientLayer,
325 }).pipe(Effect.flip);
326
327 expect(Predicate.isTagged(error, "McpConnectionError")).toBe(true);
328 expect(seen).toEqual(["https://internal.example/mcp"]);
329 }),
330 );
331
302 it.effect("integration catalog has no configured MCP integrations initially", () =>332 it.effect("integration catalog has no configured MCP integrations initially", () =>
303 Effect.gen(function* () {333 Effect.gen(function* () {
304 const executor = yield* createExecutor(makeTestConfig({ plugins: [mcpPlugin()] as const }));334 const executor = yield* createExecutor(makeTestConfig({ plugins: [mcpPlugin()] as const }));
packages/plugins/mcp/src/sdk/plugin.ts+12−3
@@ -469,6 +469,7 @@ const buildConnectorInput = ( · 468 unmodified lines
469 values: Record<string, string | null>,469 values: Record<string, string | null>,
470 templateSlug: string | null,470 templateSlug: string | null,
471 allowStdio: boolean,471 allowStdio: boolean,
472 httpClientLayer?: Layer.Layer<HttpClient.HttpClient>,
472): Effect.Effect<ConnectorInput, McpConnectionError> => {473): Effect.Effect<ConnectorInput, McpConnectionError> => {
473 if (config.transport === "stdio") {474 if (config.transport === "stdio") {
474 if (!allowStdio) {475 if (!allowStdio) {
@@ -512,6 +513,7 @@ const buildConnectorInput = ( · 37 unmodified lines
512 queryParams: Object.keys(queryParams).length > 0 ? queryParams : undefined,513 queryParams: Object.keys(queryParams).length > 0 ? queryParams : undefined,
513 headers: Object.keys(headers).length > 0 ? headers : undefined,514 headers: Object.keys(headers).length > 0 ? headers : undefined,
514 authProvider,515 authProvider,
516 httpClientLayer,
515 });517 });
516};518};
517 519
@@ -636,6 +638,7 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { · 118 unmodified lines
636 endpoint: trimmed,638 endpoint: trimmed,
637 headers: probeHeaders,639 headers: probeHeaders,
638 queryParams: probeQueryParams,640 queryParams: probeQueryParams,
641 httpClientLayer,
639 });642 });
640 643
641 const result = yield* discoverTools(connector).pipe(644 const result = yield* discoverTools(connector).pipe(
@@ -929,7 +932,7 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { · 287 unmodified lines
929 // Discovery failures (auth not ready, server down) yield an empty tool set932 // Discovery failures (auth not ready, server down) yield an empty tool set
930 // rather than failing — the connection still lands and can be refreshed.933 // rather than failing — the connection still lands and can be refreshed.
931 // -----------------------------------------------------------------------934 // -----------------------------------------------------------------------
932 resolveTools: ({ config, connection, template, getValues }) =>935 resolveTools: ({ config, connection, template, getValues, httpClientLayer }) =>
933 Effect.gen(function* () {936 Effect.gen(function* () {
934 const parsed = parseMcpIntegrationConfig(config);937 const parsed = parseMcpIntegrationConfig(config);
935 if (!parsed) return { tools: [] as readonly ToolDef[] };938 if (!parsed) return { tools: [] as readonly ToolDef[] };
@@ -945,6 +948,7 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { · 9 unmodified lines
945 values,948 values,
946 template === null ? null : String(template),949 template === null ? null : String(template),
947 allowStdio,950 allowStdio,
951 httpClientLayer,
948 ).pipe(952 ).pipe(
949 Effect.map((ci) => createMcpConnector(ci)),953 Effect.map((ci) => createMcpConnector(ci)),
950 Effect.result,954 Effect.result,
@@ -968,7 +972,7 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { · 17 unmodified lines
968 }),972 }),
969 ) as Effect.Effect<{ readonly tools: readonly ToolDef[] }, StorageFailure>,973 ) as Effect.Effect<{ readonly tools: readonly ToolDef[] }, StorageFailure>,
970 974
971 invokeTool: ({ toolRow, credential, args, elicit }) =>975 invokeTool: ({ ctx, toolRow, credential, args, elicit }) =>
972 Effect.gen(function* () {976 Effect.gen(function* () {
973 const parsed = parseMcpIntegrationConfig(credential.config);977 const parsed = parseMcpIntegrationConfig(credential.config);
974 if (!parsed) {978 if (!parsed) {
@@ -1013,6 +1017,7 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { · 38 unmodified lines
1013 credential.values,1017 credential.values,
1014 String(credential.template),1018 String(credential.template),
1015 allowStdio,1019 allowStdio,
1020 options?.httpClientLayer ?? ctx.httpClientLayer,
1016 ).pipe(Effect.map((ci) => createMcpConnector(ci)));1021 ).pipe(Effect.map((ci) => createMcpConnector(ci)));
1017 1022
1018 const raw = yield* invokeMcpTool({1023 const raw = yield* invokeMcpTool({
@@ -1087,7 +1092,11 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { · 68 unmodified lines
1087 const name = parsed.value.hostname || "mcp";1092 const name = parsed.value.hostname || "mcp";
1088 const slug = deriveMcpNamespace({ endpoint: trimmed });1093 const slug = deriveMcpNamespace({ endpoint: trimmed });
1089 1094
1090 const connector = createMcpConnector({ transport: "remote", endpoint: trimmed });1095 const connector = createMcpConnector({
1096 transport: "remote",
1097 endpoint: trimmed,
1098 httpClientLayer,
1099 });
1091 1100
1092 const connected = yield* discoverTools(connector).pipe(1101 const connected = yield* discoverTools(connector).pipe(
1093 Effect.map(() => true),1102 Effect.map(() => true),
packages/plugins/openapi/src/sdk/spec-blob.test.ts+1−0
@@ -161,6 +161,7 @@ describe("OpenAPI plugin — spec blob storage", () => { · 160 unmodified lines
161 },161 },
162 template: null,162 template: null,
163 storage,163 storage,
164 httpClientLayer: FetchHttpClient.layer,
164 getValue: () => Effect.succeed(null),165 getValue: () => Effect.succeed(null),
165 getValues: () => Effect.succeed({}),166 getValues: () => Effect.succeed({}),
166 });167 });