History

Reuse downstream MCP sessions across tool calls (#1435)

Rhys Sullivan committed 3 months ago

Commit af95edb1 parent 44f29bc10 files · +615 −97Browse files

Unified · Split

10 files
  1. .changeset/mcp-session-reuse.md +5 −0
  2. e2e/scenarios/mcp-session-state.test.ts +157 −0
  3. packages/plugins/mcp/src/sdk/connection-pool.test.ts +156 −0
  4. packages/plugins/mcp/src/sdk/connection-pool.ts +136 −0
  5. packages/plugins/mcp/src/sdk/elicitation.test.ts +53 −78
  6. packages/plugins/mcp/src/sdk/errors.ts +4 −0
  7. packages/plugins/mcp/src/sdk/invoke.ts +31 −11
  8. packages/plugins/mcp/src/sdk/multi-placement-auth.test.ts +6 −2
  9. packages/plugins/mcp/src/sdk/plugin.ts +38 −2
  10. packages/plugins/mcp/src/testing/server.ts +29 −4
.changeset/mcp-session-reuse.md+5−0
@@ -0,0 +1,5 @@
1---
2"@executor-js/plugin-mcp": patch
3---
4
5Reuse downstream MCP sessions across tool calls. Remote MCP invocations now lease connections from a per-plugin-instance pool (one idle session per resolved credential identity, exclusive per invoke, 5-minute idle TTL) instead of dialing a fresh connection per call, so servers that key state by `Mcp-Session-Id` (workspace selection and similar) see consecutive calls in the same session. A reused session rejected with HTTP 404 is redialed once transparently; stdio transports and endpoint probing remain per-call.
e2e/scenarios/mcp-session-state.test.ts+157−0
@@ -0,0 +1,157 @@
1// Cross-target: downstream MCP session continuity — the "an agent can use a
2// stateful MCP server" promise. Servers like Render's MCP key state by
3// Mcp-Session-Id (select_workspace, then every later call reads the
4// selection). If executor dials a fresh downstream connection per tool call,
5// every call lands in a brand-new session: select_workspace succeeds, and the
6// very next call reports no workspace selected. This scenario drives that
7// journey against a session-stateful MCP fixture and asserts state set by one
8// tool call is visible to the next call in the same execution.
9import { randomBytes } from "node:crypto";
10
11import { expect } from "@effect/vitest";
12import { Effect, Predicate } from "effect";
13import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
14import { composePluginApi } from "@executor-js/api/server";
15import { mcpHttpPlugin } from "@executor-js/plugin-mcp/api";
16import { serveMcpServer } from "@executor-js/plugin-mcp/testing";
17import { AuthTemplateSlug, ConnectionName, IntegrationSlug } from "@executor-js/sdk/shared";
18
19import { scenario } from "../src/scenario";
20import { Api, Target } from "../src/services";
21
22const api = composePluginApi([mcpHttpPlugin()] as const);
23
24const freshSlug = (prefix: string): string => `${prefix}_${randomBytes(4).toString("hex")}`;
25
26// A session-stateful MCP server, shaped like Render's workspace selection:
27// `select_workspace` stores the choice in the session (each MCP session gets
28// its own server instance from the factory, exactly like state keyed by
29// Mcp-Session-Id), and `get_workspace` answers from that same session state.
30const makeSessionStatefulMcpServer = () => () => {
31 const server = new McpServer(
32 { name: "session-stateful-test-server", version: "1.0.0" },
33 { capabilities: {} },
34 );
35 // Per-session: lives in this server instance, created per MCP session.
36 let selected: string | null = null;
37
38 server.registerTool(
39 "select_workspace",
40 {
41 description: "Selects the workspace for this MCP session",
42 inputSchema: {},
43 },
44 async () => {
45 selected = "ws-e2e";
46 return { content: [{ type: "text" as const, text: "selected:ws-e2e" }] };
47 },
48 );
49
50 server.registerTool(
51 "get_workspace",
52 {
53 description: "Returns the workspace selected earlier in this MCP session",
54 inputSchema: {},
55 },
56 async () => ({
57 content: [
58 {
59 type: "text" as const,
60 text: selected === null ? "no workspace selected" : `workspace:${selected}`,
61 },
62 ],
63 }),
64 );
65
66 return server;
67};
68
69// One sandbox execution, two dependent tool calls — the smallest agent journey
70// that relies on downstream session state.
71const selectThenReadCode = (slug: string) => `
72const selected = await tools.${slug}.org.main.select_workspace({});
73const read = await tools.${slug}.org.main.get_workspace({});
74return JSON.stringify({ selected, read });
75`;
76
77scenario(
78 "MCP · session state set by one tool call is visible to the next call",
79 {},
80 Effect.scoped(
81 Effect.gen(function* () {
82 const target = yield* Target;
83 const { client: makeApiClient } = yield* Api;
84 const identity = yield* target.newIdentity();
85 const client = yield* makeApiClient(api, identity);
86 const slug = freshSlug("mcp_state");
87
88 const server = yield* serveMcpServer(makeSessionStatefulMcpServer());
89
90 yield* client.mcp.addServer({
91 payload: {
92 transport: "remote",
93 name: "Session-stateful MCP",
94 endpoint: server.url,
95 slug,
96 remoteTransport: "streamable-http",
97 },
98 });
99
100 yield* Effect.gen(function* () {
101 yield* client.connections.create({
102 payload: {
103 owner: "org",
104 name: ConnectionName.make("main"),
105 integration: IntegrationSlug.make(slug),
106 template: AuthTemplateSlug.make("none"),
107 value: "",
108 },
109 });
110
111 // Only observe the execution's own traffic, not discovery's.
112 yield* server.clearRequests;
113
114 const executed = yield* client.executions.execute({
115 payload: { code: selectThenReadCode(slug), autoApprove: true },
116 });
117 expect(executed.status, "the two-call execution completed").toBe("completed");
118
119 // THE promise: the selection made by the first call is what the second
120 // call reads. A per-call connection lands the second call in a fresh
121 // downstream session, which answers "no workspace selected".
122 expect(executed.text, "the first call selected the workspace").toContain("selected:ws-e2e");
123 expect(executed.text, "the second call sees the same session's selection").toContain(
124 "workspace:ws-e2e",
125 );
126
127 // Wire-level corroboration from the fixture's request ledger: every
128 // session-bound request of this execution carried ONE Mcp-Session-Id.
129 const sessionIds = new Set(
130 (yield* server.requests)
131 .map((request) => request.sessionId)
132 .filter(Predicate.isNotUndefined),
133 );
134 expect([...sessionIds].length, "both tool calls rode a single downstream MCP session").toBe(
135 1,
136 );
137 }).pipe(
138 Effect.ensuring(
139 Effect.gen(function* () {
140 yield* client.connections
141 .remove({
142 params: {
143 owner: "org",
144 integration: IntegrationSlug.make(slug),
145 name: ConnectionName.make("main"),
146 },
147 })
148 .pipe(Effect.ignore);
149 yield* client.mcp
150 .removeServer({ params: { slug: IntegrationSlug.make(slug) } })
151 .pipe(Effect.ignore);
152 }),
153 ),
154 );
155 }),
156 ),
157);
packages/plugins/mcp/src/sdk/connection-pool.test.ts+156−0
@@ -0,0 +1,156 @@
1import { describe, expect, it } from "@effect/vitest";
2import { Effect } from "effect";
3import { ElicitationResponse, type Elicit } from "@executor-js/sdk";
4
5import { createMcpConnector } from "./connection";
6import { createMcpConnectionPool } from "./connection-pool";
7import { invokeMcpTool } from "./invoke";
8import { makeEchoMcpServer, serveMcpServer } from "../testing";
9
10const acceptAll: Elicit = () =>
11 Effect.succeed(ElicitationResponse.make({ action: "accept", content: { approved: true } }));
12
13const invoke = (input: {
14 readonly endpoint: string;
15 readonly pool: ReturnType<typeof createMcpConnectionPool>;
16 readonly toolName: string;
17 readonly args: Record<string, unknown>;
18 readonly elicit?: Elicit;
19}) =>
20 invokeMcpTool({
21 toolId: input.toolName,
22 toolName: input.toolName,
23 args: input.args,
24 transport: "streamable-http",
25 connector: createMcpConnector({
26 transport: "remote",
27 endpoint: input.endpoint,
28 remoteTransport: "streamable-http",
29 }),
30 connectionPool: input.pool,
31 connectionPoolKey: input.endpoint,
32 elicit: input.elicit ?? acceptAll,
33 });
34
35describe("MCP connection pool", () => {
36 it.effect("reuses one session for sequential invokes", () =>
37 Effect.scoped(
38 Effect.gen(function* () {
39 const server = yield* serveMcpServer(() => makeEchoMcpServer());
40 const pool = createMcpConnectionPool();
41
42 const first = yield* invoke({
43 endpoint: server.endpoint,
44 pool,
45 toolName: "echo",
46 args: { value: "first" },
47 });
48 const second = yield* invoke({
49 endpoint: server.endpoint,
50 pool,
51 toolName: "echo",
52 args: { value: "second" },
53 });
54
55 expect(first).toMatchObject({ content: [{ type: "text", text: "first" }] });
56 expect(second).toMatchObject({ content: [{ type: "text", text: "second" }] });
57 expect(server.sessionCount()).toBe(1);
58 yield* pool.close();
59 }),
60 ),
61 );
62
63 it.effect("leases separate sessions to concurrent invokes", () =>
64 Effect.scoped(
65 Effect.gen(function* () {
66 const server = yield* serveMcpServer(() => makeEchoMcpServer());
67 const pool = createMcpConnectionPool();
68
69 const results = yield* Effect.all(
70 [
71 invoke({
72 endpoint: server.endpoint,
73 pool,
74 toolName: "echo",
75 args: { value: "left" },
76 }),
77 invoke({
78 endpoint: server.endpoint,
79 pool,
80 toolName: "echo",
81 args: { value: "right" },
82 }),
83 ],
84 { concurrency: "unbounded" },
85 );
86
87 expect(results).toEqual([
88 expect.objectContaining({ content: [{ type: "text", text: "left" }] }),
89 expect.objectContaining({ content: [{ type: "text", text: "right" }] }),
90 ]);
91 expect(server.sessionCount()).toBe(2);
92 yield* pool.close();
93 }),
94 ),
95 );
96
97 it.effect("redials once when a reused session has expired", () =>
98 Effect.scoped(
99 Effect.gen(function* () {
100 const server = yield* serveMcpServer(() => makeEchoMcpServer());
101 const pool = createMcpConnectionPool();
102
103 yield* invoke({
104 endpoint: server.endpoint,
105 pool,
106 toolName: "echo",
107 args: { value: "before" },
108 });
109 yield* server.forgetSessions;
110 const after = yield* invoke({
111 endpoint: server.endpoint,
112 pool,
113 toolName: "echo",
114 args: { value: "after" },
115 });
116
117 expect(after).toMatchObject({ content: [{ type: "text", text: "after" }] });
118 expect(server.sessionCount()).toBe(2);
119 yield* pool.close();
120 }),
121 ),
122 );
123
124 it.effect("drops a connection after a transport-level invocation failure", () =>
125 Effect.scoped(
126 Effect.gen(function* () {
127 const server = yield* serveMcpServer(() => makeEchoMcpServer());
128 const pool = createMcpConnectionPool();
129
130 yield* invoke({
131 endpoint: server.endpoint,
132 pool,
133 toolName: "echo",
134 args: { value: "before" },
135 });
136 yield* server.rejectNextSessionRequest(500);
137 yield* invoke({
138 endpoint: server.endpoint,
139 pool,
140 toolName: "echo",
141 args: { value: "fails" },
142 }).pipe(Effect.flip);
143 const after = yield* invoke({
144 endpoint: server.endpoint,
145 pool,
146 toolName: "echo",
147 args: { value: "after" },
148 });
149
150 expect(after).toMatchObject({ content: [{ type: "text", text: "after" }] });
151 expect(server.sessionCount()).toBe(2);
152 yield* pool.close();
153 }),
154 ),
155 );
156});
packages/plugins/mcp/src/sdk/connection-pool.ts+136−0
@@ -0,0 +1,136 @@
1import { Cause, Effect, Exit, Predicate } from "effect";
2
3import type { McpConnection, McpConnector } from "./connection";
4import type { McpInvocationError } from "./errors";
5
6const IDLE_TTL_MS = 5 * 60 * 1_000;
7
8type IdleConnection = {
9 readonly connection: McpConnection;
10 readonly idleSince: number;
11};
12
13type ConnectionLease = {
14 readonly connection: McpConnection;
15 readonly reused: boolean;
16};
17
18const closeQuietly = (connection: McpConnection): Effect.Effect<void> =>
19 Effect.tryPromise(() => connection.close()).pipe(Effect.ignore);
20
21const isMcpInvocationError = (error: unknown): error is McpInvocationError =>
22 Predicate.isTagged(error, "McpInvocationError");
23
24const isDeadConnectionFailure = (error: unknown): boolean => {
25 if (Predicate.isTagged(error, "McpConnectionError")) return true;
26 if (!isMcpInvocationError(error)) return false;
27 return (
28 error.transportFailure === true ||
29 error.status === 400 ||
30 error.status === 404 ||
31 error.status === 408
32 );
33};
34
35const shouldDropConnection = <A, E>(exit: Exit.Exit<A, E>): boolean => {
36 if (Exit.isSuccess(exit)) return false;
37 const failures = exit.cause.reasons.filter(Cause.isFailReason);
38 if (failures.length === 0) return true;
39 return failures.some((failure) => isDeadConnectionFailure(failure.error));
40};
41
42/** A per-plugin-instance pool that gives each invocation an exclusive MCP
43 * connection lease while retaining at most one idle session per identity. */
44export interface McpConnectionPool {
45 readonly withConnection: <A, E, R>(
46 key: string,
47 connector: McpConnector,
48 use: (connection: McpConnection) => Effect.Effect<A, E, R>,
49 ) => Effect.Effect<A, E | Effect.Error<McpConnector>, R>;
50 /** Closes and removes every currently idle connection. Leased connections
51 * remain owned by their in-flight invocation and follow normal release. */
52 readonly close: () => Effect.Effect<void>;
53}
54
55/** Creates an MCP connection pool with lazy five-minute idle eviction and one
56 * automatic fresh-dial retry for a reused session rejected with HTTP 404. */
57export const createMcpConnectionPool = (): McpConnectionPool => {
58 const idle = new Map<string, IdleConnection>();
59
60 const acquire = (key: string, connector: McpConnector, forceFresh: boolean) =>
61 Effect.gen(function* () {
62 if (forceFresh) {
63 const connection = yield* connector;
64 return { connection, reused: false } satisfies ConnectionLease;
65 }
66
67 const taken = yield* Effect.sync(() => {
68 const entry = idle.get(key);
69 if (!entry) return { connection: undefined, expired: undefined };
70 idle.delete(key);
71 if (Date.now() - entry.idleSince >= IDLE_TTL_MS) {
72 return { connection: undefined, expired: entry.connection };
73 }
74 return { connection: entry.connection, expired: undefined };
75 });
76
77 if (taken.expired) yield* closeQuietly(taken.expired);
78 if (taken.connection) {
79 return { connection: taken.connection, reused: true } satisfies ConnectionLease;
80 }
81
82 const connection = yield* connector;
83 return { connection, reused: false } satisfies ConnectionLease;
84 });
85
86 const release = <A, E>(key: string, lease: ConnectionLease, exit: Exit.Exit<A, E>) =>
87 Effect.gen(function* () {
88 if (shouldDropConnection(exit)) {
89 yield* closeQuietly(lease.connection);
90 return;
91 }
92
93 const displaced = yield* Effect.sync(() => {
94 if (idle.has(key)) return lease.connection;
95 idle.set(key, { connection: lease.connection, idleSince: Date.now() });
96 return undefined;
97 });
98 if (displaced) yield* closeQuietly(displaced);
99 });
100
101 const withConnection: McpConnectionPool["withConnection"] = (key, connector, use) => {
102 let reused = false;
103 const run = (forceFresh: boolean) =>
104 Effect.acquireUseRelease(
105 acquire(key, connector, forceFresh),
106 (lease) => {
107 if (!forceFresh) reused = lease.reused;
108 return use(lease.connection);
109 },
110 (lease, exit) => release(key, lease, exit),
111 );
112
113 return run(false).pipe(
114 Effect.catch((error) =>
115 reused && isMcpInvocationError(error) && error.status === 404
116 ? run(true)
117 : Effect.fail(error),
118 ),
119 );
120 };
121
122 const close = () =>
123 Effect.gen(function* () {
124 const connections = yield* Effect.sync(() => {
125 const connections = [...idle.values()].map((entry) => entry.connection);
126 idle.clear();
127 return connections;
128 });
129 yield* Effect.forEach(connections, closeQuietly, {
130 concurrency: "unbounded",
131 discard: true,
132 });
133 });
134
135 return { withConnection, close };
136};
packages/plugins/mcp/src/sdk/elicitation.test.ts+53−78
@@ -1,5 +1,5 @@
1import { describe, expect, it } from "@effect/vitest";1import { describe, expect, it } from "@effect/vitest";
2import { Effect, Schema } from "effect";2import { Effect, Predicate, Schema, Semaphore } from "effect";
3import { CfWorkerJsonSchemaValidator } from "@modelcontextprotocol/sdk/validation/cfworker";3import { CfWorkerJsonSchemaValidator } from "@modelcontextprotocol/sdk/validation/cfworker";
4import type { JsonSchemaType } from "@modelcontextprotocol/sdk/validation/types";4import type { JsonSchemaType } from "@modelcontextprotocol/sdk/validation/types";
5 5
@@ -26,7 +26,10 @@ import { makeElicitationMcpServer, serveMcpServer } from "../testing"; · 20 unmodified lines
26 26
27const isFormElicitation = Schema.is(FormElicitation);27const isFormElicitation = Schema.is(FormElicitation);
28 28
29const serveElicitationTestServer = serveMcpServer(makeElicitationMcpServer);29const elicitationServerSemaphore = Semaphore.makeUnsafe(1);
30const serveElicitationTestServer = Effect.acquireRelease(elicitationServerSemaphore.take(1), () =>
31 elicitationServerSemaphore.release(1).pipe(Effect.asVoid),
32).pipe(Effect.flatMap(() => serveMcpServer(makeElicitationMcpServer)));
30 33
31const schemaValidator = new CfWorkerJsonSchemaValidator({ shortcircuit: false });34const schemaValidator = new CfWorkerJsonSchemaValidator({ shortcircuit: false });
32 35
@@ -56,8 +59,9 @@ const TEMPLATE = AuthTemplateSlug.make("none"); · 23 unmodified lines
56// ---------------------------------------------------------------------------59// ---------------------------------------------------------------------------
57 60
58const makeTestExecutor = (serverUrl: string) =>61const makeTestExecutor = (serverUrl: string) =>
59 createExecutor(62 Effect.acquireRelease(
60 makeTestConfig({ plugins: [memoryCredentialsPlugin(), mcpPlugin()] as const }),63 createExecutor(makeTestConfig({ plugins: [memoryCredentialsPlugin(), mcpPlugin()] as const })),
64 (executor) => executor.close().pipe(Effect.ignore),
61 ).pipe(65 ).pipe(
62 Effect.tap((executor) =>66 Effect.tap((executor) =>
63 Effect.gen(function* () {67 Effect.gen(function* () {
@@ -85,7 +89,7 @@ const findTool = (tools: readonly Tool[], name: string): Tool => · 21 unmodified lines
85// ---------------------------------------------------------------------------89// ---------------------------------------------------------------------------
86 90
87describe("MCP elicitation (end-to-end)", () => {91describe("MCP elicitation (end-to-end)", () => {
88 it.effect("form elicitation accepted → tool returns approved result", () =>92 it.effect("reuses the session while installing each invoke's elicitation handler", () =>
89 Effect.gen(function* () {93 Effect.gen(function* () {
90 const server = yield* serveElicitationTestServer;94 const server = yield* serveElicitationTestServer;
91 const executor = yield* makeTestExecutor(server.url);95 const executor = yield* makeTestExecutor(server.url);
@@ -93,12 +97,19 @@ describe("MCP elicitation (end-to-end)", () => { · 1 unmodified line
93 const tools = yield* executor.tools.list();97 const tools = yield* executor.tools.list();
94 const gatedEcho = findTool(tools, "gated_echo");98 const gatedEcho = findTool(tools, "gated_echo");
95 99
96 const elicitationMessages: string[] = [];100 const acceptedMessages: string[] = [];
101 const declinedMessages: string[] = [];
102 let capturedAddress: string | undefined;
103 let capturedArgs: unknown;
104 let capturedRequest: unknown;
97 105
98 const options: InvokeOptions = {106 const options: InvokeOptions = {
99 onElicitation: (ctx) => {107 onElicitation: (ctx) => {
108 capturedAddress = String(ctx.address);
109 capturedArgs = ctx.args;
110 capturedRequest = ctx.request;
100 if (isFormElicitation(ctx.request)) {111 if (isFormElicitation(ctx.request)) {
101 elicitationMessages.push(ctx.request.message);112 acceptedMessages.push(ctx.request.message);
102 }113 }
103 return Effect.succeed(114 return Effect.succeed(
104 ElicitationResponse.make({115 ElicitationResponse.make({
@@ -109,36 +120,50 @@ describe("MCP elicitation (end-to-end)", () => { · 4 unmodified lines
109 },120 },
110 };121 };
111 122
112 const result = yield* executor.execute(gatedEcho.address, { value: "hello" }, options);123 yield* server.clearRequests;
113 124 const accepted = yield* executor.execute(gatedEcho.address, { value: "hello" }, options);
114 expect(result).toMatchObject({125 const declined = yield* executor.execute(
115 ok: true,
116 data: { content: [{ type: "text", text: "approved:hello" }] },
117 });
118 expect(elicitationMessages.length).toBeGreaterThanOrEqual(1);
119 expect(elicitationMessages.some((m) => m.includes('Approve echo for "hello"?'))).toBe(true);
120 }),
121 );
122
123 it.effect("form elicitation declined → tool returns denied result", () =>
124 Effect.gen(function* () {
125 const server = yield* serveElicitationTestServer;
126 const executor = yield* makeTestExecutor(server.url);
127 const tools = yield* executor.tools.list();
128 const gatedEcho = findTool(tools, "gated_echo");
129
130 const result = yield* executor.execute(
131 gatedEcho.address,126 gatedEcho.address,
132 { value: "nope" },127 { value: "nope" },
133 {128 {
134 onElicitation: () => Effect.succeed(ElicitationResponse.make({ action: "decline" })),129 onElicitation: (ctx) => {
130 if (isFormElicitation(ctx.request)) {
131 declinedMessages.push(ctx.request.message);
132 }
133 return Effect.succeed(ElicitationResponse.make({ action: "decline" }));
134 },
135 },135 },
136 );136 );
137 yield* executor.close();
137 138
138 expect(result).toMatchObject({139 expect(accepted).toMatchObject({
140 ok: true,
141 data: { content: [{ type: "text", text: "approved:hello" }] },
142 });
143 expect(declined).toMatchObject({
139 ok: true,144 ok: true,
140 data: { content: [{ type: "text", text: "denied:nope" }] },145 data: { content: [{ type: "text", text: "denied:nope" }] },
141 });146 });
147 expect(acceptedMessages).toEqual(['Approve echo for "hello"?']);
148 expect(declinedMessages).toEqual(['Approve echo for "nope"?']);
149 expect(capturedAddress).toBe(String(gatedEcho.address));
150 expect(capturedArgs).toEqual({ value: "hello" });
151 expect(isFormElicitation(capturedRequest)).toBe(true);
152 const form = capturedRequest as FormElicitation;
153 expect(form.message).toContain('Approve echo for "hello"?');
154 expect(form.requestedSchema).toEqual({
155 type: "object",
156 properties: {
157 approved: { type: "boolean", title: "Approve" },
158 },
159 required: ["approved"],
160 });
161 const sessionIds = new Set(
162 (yield* server.requests)
163 .map((request) => request.sessionId)
164 .filter(Predicate.isNotUndefined),
165 );
166 expect(sessionIds.size).toBe(1);
142 }),167 }),
143 );168 );
144 169
@@ -316,54 +341,4 @@ describe("MCP elicitation (end-to-end)", () => { · 171 unmodified lines
316 expect(integration?.description).toBe("Gmail");341 expect(integration?.description).toBe("Gmail");
317 }),342 }),
318 );343 );
319
320 it.effect("handler receives correct address, args, and FormElicitation schema", () =>
321 Effect.gen(function* () {
322 const server = yield* serveElicitationTestServer;
323 const executor = yield* makeTestExecutor(server.url);
324 const tools = yield* executor.tools.list();
325 const gatedEcho = findTool(tools, "gated_echo");
326
327 let capturedAddress: string | undefined;
328 let capturedArgs: unknown;
329 let capturedRequest: unknown;
330
331 yield* executor.execute(
332 gatedEcho.address,
333 { value: "ctx-test" },
334 {
335 onElicitation: (ctx) => {
336 capturedAddress = String(ctx.address);
337 capturedArgs = ctx.args;
338 capturedRequest = ctx.request;
339 return Effect.succeed(
340 ElicitationResponse.make({
341 action: "accept",
342 content: { approved: true },
343 }),
344 );
345 },
346 },
347 );
348
349 expect(capturedAddress).toBe(String(gatedEcho.address));
350 expect(capturedArgs).toEqual({ value: "ctx-test" });
351 expect(isFormElicitation(capturedRequest)).toBe(true);
352
353 const form = capturedRequest as FormElicitation;
354 expect(form.message).toContain('Approve echo for "ctx-test"?');
355 expect(form.requestedSchema).toEqual({
356 type: "object",
357 properties: {
358 approved: { type: "boolean", title: "Approve" },
359 },
360 required: ["approved"],
361 });
362 }),
363 );
364
365 // removed: "connection is reused across multiple tool calls" — v2 does not
366 // cache MCP client connections across invocations (Hyperdrive request-scoped
367 // rule; each invoke dials and closes its own connection). Session reuse is no
368 // longer a property of the plugin, so the assertion no longer applies.
369});344});
packages/plugins/mcp/src/sdk/errors.ts+4−0
@@ -38,6 +38,10 @@ export class McpInvocationError extends Data.TaggedError("McpInvocationError")<{ · 37 unmodified lines
38 readonly toolName: string;38 readonly toolName: string;
39 readonly message: string;39 readonly message: string;
40 readonly status?: number;40 readonly status?: number;
41 /** The SDK failure came from the HTTP/SSE transport rather than a JSON-RPC
42 * protocol response. Used only to decide whether a leased session is safe
43 * to return to the idle pool. */
44 readonly transportFailure?: boolean;
41 /** The server rejected the call as an unknown tool (protocol error), which45 /** The server rejected the call as an unknown tool (protocol error), which
42 * means the persisted catalog has drifted from the server's live tool set. */46 * means the persisted catalog has drifted from the server's live tool set. */
43 readonly unknownTool?: boolean;47 readonly unknownTool?: boolean;
packages/plugins/mcp/src/sdk/invoke.ts+31−11
@@ -2,9 +2,12 @@ · 1 unmodified line
2// MCP tool invocation — shared helper called from plugin.invokeTool.2// MCP tool invocation — shared helper called from plugin.invokeTool.
3//3//
4// Responsible for:4// Responsible for:
5// 1. Dialing a fresh MCP client connection for the call (no DB-connection5// 1. Leasing remote connections from a per-plugin-instance pool when one is
6// caching — request-scoped per the Hyperdrive rule; each invoke acquires6// provided. MCP streamable-http is a series of HTTP requests keyed by the
7// and releases its own connection).7// application-level `Mcp-Session-Id`, not a raw pooled socket, and servers
8// legitimately retain state in that session. The pool keeps one idle
9// connection per resolved identity and leases it exclusively per invoke;
10// stdio and callers without a pool remain strictly per-call.
8// 2. Installing a per-invocation `ElicitRequestSchema` handler that bridges11// 2. Installing a per-invocation `ElicitRequestSchema` handler that bridges
9// MCP's elicit capability into the host's elicit function threaded via12// MCP's elicit capability into the host's elicit function threaded via
10// `InvokeToolInput.elicit`.13// `InvokeToolInput.elicit`.
@@ -30,6 +33,7 @@ import { · 19 unmodified lines
30 33
31import { McpConnectionError, McpInvocationError, McpOAuthReauthorizationRequired } from "./errors";34import { McpConnectionError, McpInvocationError, McpOAuthReauthorizationRequired } from "./errors";
32import type { McpConnection, McpConnector } from "./connection";35import type { McpConnection, McpConnector } from "./connection";
36import type { McpConnectionPool } from "./connection-pool";
33import { httpStatusFromCause, insufficientScopeFromCause } from "./http-status";37import { httpStatusFromCause, insufficientScopeFromCause } from "./http-status";
34 38
35// ---------------------------------------------------------------------------39// ---------------------------------------------------------------------------
@@ -181,13 +185,17 @@ const useConnection = ( · 145 unmodified lines
181 message: `MCP tool call failed for ${toolName}`,185 message: `MCP tool call failed for ${toolName}`,
182 status: 403,186 status: 403,
183 insufficientScope: true,187 insufficientScope: true,
188 transportFailure: true,
184 });189 });
185 }190 }
186 const status = httpStatusFromCause(cause);191 const status = httpStatusFromCause(cause);
192 // oxlint-disable-next-line executor/no-instanceof-tagged-error -- boundary: MCP SDK protocol failures are its McpError subclass; transport failures use other error shapes
193 const protocolFailure = cause instanceof McpError;
187 return new McpInvocationError({194 return new McpInvocationError({
188 toolName,195 toolName,
189 message: `MCP tool call failed for ${toolName}`,196 message: `MCP tool call failed for ${toolName}`,
190 ...(status === undefined ? {} : { status }),197 ...(status === undefined ? {} : { status }),
198 ...(!protocolFailure ? { transportFailure: true } : {}),
191 ...(isUnknownToolCause(cause, toolName) ? { unknownTool: true } : {}),199 ...(isUnknownToolCause(cause, toolName) ? { unknownTool: true } : {}),
192 ...(status === 403 && insufficientScopeFromCause(cause)200 ...(status === 403 && insufficientScopeFromCause(cause)
193 ? { insufficientScope: true }201 ? { insufficientScope: true }
@@ -211,8 +219,12 @@ export interface InvokeMcpToolInput { · 17 unmodified lines
211 readonly toolName: string;219 readonly toolName: string;
212 readonly args: unknown;220 readonly args: unknown;
213 readonly transport: string;221 readonly transport: string;
214 /** Dials a fresh connection. The connection is closed after the call. */222 /** Dials a fresh connection when no reusable remote session is available. */
215 readonly connector: McpConnector;223 readonly connector: McpConnector;
224 /** Optional per-plugin-instance pool and resolved remote identity key. Both
225 * must be present to enable reuse; otherwise the connection is per-call. */
226 readonly connectionPool?: McpConnectionPool;
227 readonly connectionPoolKey?: string;
216 readonly elicit: Elicit;228 readonly elicit: Elicit;
217 /** Fired when the server sends `notifications/tools/list_changed` during229 /** Fired when the server sends `notifications/tools/list_changed` during
218 * the call window. Synchronous and non-throwing by contract; the caller230 * the call window. Synchronous and non-throwing by contract; the caller
@@ -228,6 +240,20 @@ export const invokeMcpTool = ( · 9 unmodified lines
228> =>240> =>
229 Effect.gen(function* () {241 Effect.gen(function* () {
230 const args = argsRecord(input.args);242 const args = argsRecord(input.args);
243 const use = (connection: McpConnection) =>
244 useConnection(connection, input.toolName, args, input.elicit, input.onToolListChanged);
245
246 if (input.connectionPool && input.connectionPoolKey) {
247 return yield* input.connectionPool.withConnection(
248 input.connectionPoolKey,
249 input.connector.pipe(
250 Effect.withSpan("plugin.mcp.connection.acquire", {
251 attributes: { "plugin.mcp.transport": input.transport },
252 }),
253 ),
254 use,
255 );
256 }
231 257
232 const connection = yield* Effect.acquireRelease(258 const connection = yield* Effect.acquireRelease(
233 input.connector.pipe(259 input.connector.pipe(
@@ -248,13 +274,7 @@ export const invokeMcpTool = ( · 14 unmodified lines
248 ),274 ),
249 );275 );
250 276
251 return yield* useConnection(277 return yield* use(connection);
252 connection,
253 input.toolName,
254 args,
255 input.elicit,
256 input.onToolListChanged,
257 );
258 }).pipe(278 }).pipe(
259 Effect.scoped,279 Effect.scoped,
260 Effect.withSpan("plugin.mcp.invoke", {280 Effect.withSpan("plugin.mcp.invoke", {
packages/plugins/mcp/src/sdk/multi-placement-auth.test.ts+6−2
@@ -169,7 +169,9 @@ describe("MCP multi-placement auth", () => { · 168 unmodified lines
169 { marker: "h" },169 { marker: "h" },
170 { onElicitation: "accept-all" },170 { onElicitation: "accept-all" },
171 );171 );
172 const headerRequests = (yield* server.requests).slice(beforeHeader);172 const headerRequests = (yield* server.requests)
173 .slice(beforeHeader)
174 .filter((request) => request.method === "POST");
173 expect(headerRequests.every((r) => r.authorization === "Bearer header-secret")).toBe(true);175 expect(headerRequests.every((r) => r.authorization === "Bearer header-secret")).toBe(true);
174 expect(headerRequests.every((r) => !r.url.includes("auth_token="))).toBe(true);176 expect(headerRequests.every((r) => !r.url.includes("auth_token="))).toBe(true);
175 177
@@ -179,7 +181,9 @@ describe("MCP multi-placement auth", () => { · 3 unmodified lines
179 { marker: "q" },181 { marker: "q" },
180 { onElicitation: "accept-all" },182 { onElicitation: "accept-all" },
181 );183 );
182 const queryRequests = (yield* server.requests).slice(beforeQuery);184 const queryRequests = (yield* server.requests)
185 .slice(beforeQuery)
186 .filter((request) => request.method === "POST");
183 expect(queryRequests.every((r) => r.url.includes("auth_token=query-secret"))).toBe(true);187 expect(queryRequests.every((r) => r.url.includes("auth_token=query-secret"))).toBe(true);
184 expect(queryRequests.every((r) => r.authorization === undefined)).toBe(true);188 expect(queryRequests.every((r) => r.authorization === undefined)).toBe(true);
185 }),189 }),
packages/plugins/mcp/src/sdk/plugin.ts+38−2
@@ -41,6 +41,7 @@ import { · 40 unmodified lines
41} from "@executor-js/sdk/http-auth";41} from "@executor-js/sdk/http-auth";
42 42
43import { createMcpConnector, type ConnectorInput, type McpConnector } from "./connection";43import { createMcpConnector, type ConnectorInput, type McpConnector } from "./connection";
44import { createMcpConnectionPool } from "./connection-pool";
44import { discoverTools } from "./discover";45import { discoverTools } from "./discover";
45import {46import {
46 McpConnectionError,47 McpConnectionError,
@@ -615,6 +616,33 @@ const buildConnectorInput = ( · 568 unmodified lines
615 });616 });
616};617};
617 618
619// Record fields are key-sorted before stringifying: the key only guards
620// same-identity reuse (a mismatch merely costs a fresh dial), but insertion
621// order must not make two equal identities look distinct.
622const sortedRecord = (
623 record: Record<string, string | null | undefined> | undefined,
624): Record<string, string | null> =>
625 Object.fromEntries(
626 Object.entries(record ?? {})
627 .filter((entry): entry is [string, string | null] => entry[1] !== undefined)
628 .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)),
629 );
630
631const connectionPoolKey = (
632 input: Extract<ConnectorInput, { readonly transport: "remote" }>,
633 template: string,
634 values: Record<string, string | null>,
635): string =>
636 JSON.stringify({
637 endpoint: input.endpoint,
638 transport: input.transport,
639 remoteTransport: input.remoteTransport,
640 headers: sortedRecord(input.headers),
641 queryParams: sortedRecord(input.queryParams),
642 template,
643 values: sortedRecord(values),
644 });
645
618// ---------------------------------------------------------------------------646// ---------------------------------------------------------------------------
619// Declared auth methods — project the stored MCP config into the catalog's647// Declared auth methods — project the stored MCP config into the catalog's
620// plugin-agnostic `AuthMethodDescriptor[]`, one per declared method. Pure and648// plugin-agnostic `AuthMethodDescriptor[]`, one per declared method. Pure and
@@ -699,6 +727,7 @@ export interface McpPluginOptions { · 78 unmodified lines
699 727
700export const mcpPlugin = definePlugin((options?: McpPluginOptions) => {728export const mcpPlugin = definePlugin((options?: McpPluginOptions) => {
701 const allowStdio = options?.dangerouslyAllowStdioMCP ?? false;729 const allowStdio = options?.dangerouslyAllowStdioMCP ?? false;
730 const connectionPool = createMcpConnectionPool();
702 731
703 const presetEntries = (732 const presetEntries = (
704 allowStdio733 allowStdio
@@ -731,6 +760,7 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { · 26 unmodified lines
731 // factory reads `allowStdio` and gates the stdio tab + presets.760 // factory reads `allowStdio` and gates the stdio tab + presets.
732 clientConfig: { allowStdio },761 clientConfig: { allowStdio },
733 storage: () => ({}),762 storage: () => ({}),
763 close: () => connectionPool.close(),
734 764
735 extension: (ctx: PluginCtx) => {765 extension: (ctx: PluginCtx) => {
736 const httpClientLayer = options?.httpClientLayer ?? ctx.httpClientLayer;766 const httpClientLayer = options?.httpClientLayer ?? ctx.httpClientLayer;
@@ -1266,13 +1296,18 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { · 529 unmodified lines
1266 }1296 }
1267 }1297 }
1268 1298
1269 const connector: McpConnector = yield* buildConnectorInput(1299 const connectorInput = yield* buildConnectorInput(
1270 parsed,1300 parsed,
1271 credential.values,1301 credential.values,
1272 String(credential.template),1302 String(credential.template),
1273 allowStdio,1303 allowStdio,
1274 options?.httpClientLayer ?? ctx.httpClientLayer,1304 options?.httpClientLayer ?? ctx.httpClientLayer,
1275 ).pipe(Effect.map((ci) => createMcpConnector(ci)));1305 );
1306 const connector: McpConnector = createMcpConnector(connectorInput);
1307 const poolKey =
1308 connectorInput.transport === "remote"
1309 ? connectionPoolKey(connectorInput, String(credential.template), credential.values)
1310 : undefined;
1276 1311
1277 const connectionRef = {1312 const connectionRef = {
1278 owner: credential.owner,1313 owner: credential.owner,
@@ -1293,6 +1328,7 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { · 14 unmodified lines
1293 args,1328 args,
1294 transport,1329 transport,
1295 connector,1330 connector,
1331 ...(poolKey === undefined ? {} : { connectionPool, connectionPoolKey: poolKey }),
1296 elicit,1332 elicit,
1297 onToolListChanged: () => {1333 onToolListChanged: () => {
1298 toolListChanged = true;1334 toolListChanged = true;
packages/plugins/mcp/src/testing/server.ts+29−4
@@ -12,6 +12,10 @@ export type McpTestServer = { · 11 unmodified lines
12 readonly sessionCount: () => number;12 readonly sessionCount: () => number;
13 readonly requests: Effect.Effect<readonly McpTestRequest[]>;13 readonly requests: Effect.Effect<readonly McpTestRequest[]>;
14 readonly clearRequests: Effect.Effect<void>;14 readonly clearRequests: Effect.Effect<void>;
15 /** Drops all server-side session registrations without notifying clients. */
16 readonly forgetSessions: Effect.Effect<void>;
17 /** Rejects the next request carrying an MCP session id with this status. */
18 readonly rejectNextSessionRequest: (status: number) => Effect.Effect<void>;
15};19};
16 20
17export type McpTestRequest = {21export type McpTestRequest = {
@@ -64,9 +68,11 @@ export const serveMcpServer = (factory: () => McpServer, options: McpTestServerO · 46 unmodified lines
64 Effect.acquireRelease(68 Effect.acquireRelease(
65 Effect.gen(function* () {69 Effect.gen(function* () {
66 const transports = new Map<string, StreamableHTTPServerTransport>();70 const transports = new Map<string, StreamableHTTPServerTransport>();
71 const allTransports = new Set<StreamableHTTPServerTransport>();
67 const requests = yield* Ref.make<readonly McpTestRequest[]>([]);72 const requests = yield* Ref.make<readonly McpTestRequest[]>([]);
68 const path = options.path ?? "/";73 const path = options.path ?? "/";
69 let sessions = 0;74 let sessions = 0;
75 let nextSessionRequestStatus: number | undefined;
70 76
71 const handleMcpRequest = (77 const handleMcpRequest = (
72 request: http.IncomingMessage,78 request: http.IncomingMessage,
@@ -130,6 +136,13 @@ export const serveMcpServer = (factory: () => McpServer, options: McpTestServerO · 57 unmodified lines
130 }136 }
131 }137 }
132 138
139 if (sessionId && request.method === "POST" && nextSessionRequestStatus !== undefined) {
140 const status = nextSessionRequestStatus;
141 nextSessionRequestStatus = undefined;
142 writeText(response, status, `Forced HTTP ${status}`);
143 return;
144 }
145
133 const existingTransport = sessionId ? transports.get(sessionId) : undefined;146 const existingTransport = sessionId ? transports.get(sessionId) : undefined;
134 if (sessionId && !existingTransport) {147 if (sessionId && !existingTransport) {
135 writeText(response, 404, "Session not found");148 writeText(response, 404, "Session not found");
@@ -150,6 +163,7 @@ export const serveMcpServer = (factory: () => McpServer, options: McpTestServerO · 14 unmodified lines
150 transports.set(sid, transport);163 transports.set(sid, transport);
151 },164 },
152 });165 });
166 allTransports.add(transport);
153 sessions += 1;167 sessions += 1;
154 168
155 const mcpServer = factory();169 const mcpServer = factory();
@@ -202,21 +216,32 @@ export const serveMcpServer = (factory: () => McpServer, options: McpTestServerO · 46 unmodified lines
202 sessionCount: () => sessions,216 sessionCount: () => sessions,
203 requests: Ref.get(requests),217 requests: Ref.get(requests),
204 clearRequests: Ref.set(requests, []),218 clearRequests: Ref.set(requests, []),
219 forgetSessions: Effect.sync(() => transports.clear()),
220 rejectNextSessionRequest: (status: number) =>
221 Effect.sync(() => {
222 nextSessionRequestStatus = status;
223 }),
205 close: Effect.gen(function* () {224 close: Effect.gen(function* () {
206 for (const transport of transports.values()) {225 for (const transport of allTransports) {
207 yield* Effect.tryPromise({226 yield* Effect.tryPromise({
208 try: () => transport.close(),227 try: () => transport.close(),
209 catch: (cause) => new McpTestServerError({ cause }),228 catch: (cause) => new McpTestServerError({ cause }),
210 }).pipe(Effect.ignore);229 }).pipe(Effect.ignore);
211 }230 }
212 yield* Effect.sync(() => {231 yield* Effect.callback<void>((resume) => {
213 nodeServer.close();232 nodeServer.close(() => resume(Effect.void));
214 nodeServer.closeAllConnections?.();233 nodeServer.closeAllConnections?.();
215 });234 });
235 // closeAllConnections destroys sockets out from under in-flight SDK
236 // reads; give those rejections a beat to settle before the scope
237 // ends so they surface inside the test, not as unhandled rejections.
238 // A raw timer, not Effect.sleep: this runs inside it.effect tests
239 // whose TestClock never advances wall-clock sleeps.
240 yield* Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, 25)));
216 }),241 }),
217 };242 };
218 }),243 }),
219 (server) => server.close,244 (server) => server.close.pipe(Effect.ignore),
220 ).pipe(Effect.map(({ close: _close, ...server }) => server));245 ).pipe(Effect.map(({ close: _close, ...server }) => server));
221 246
222export const serveMcpServerWithOAuth = (247export const serveMcpServerWithOAuth = (