History

Merge pull request #714 from RhysSullivan/codex/mcp-connection-identity

Include source identity in MCP connection state

Rhys Sullivan committed 6 months ago

Commit 87452272 parents d4839a1 545fb9a3 files · +102 −13Browse files

Unified · Split

3 files
  1. packages/plugins/mcp/src/sdk/connection-pool.test.ts +49 −2
  2. packages/plugins/mcp/src/sdk/invoke.ts +51 −11
  3. packages/plugins/mcp/src/sdk/plugin.ts +2 −0
packages/plugins/mcp/src/sdk/connection-pool.test.ts+49−2
@@ -123,7 +123,7 @@ const serveMcpServer = Effect.acquireRelease( · 122 unmodified lines
123// Helper — one executor, one mcp source pointed at the test server123// Helper — one executor, one mcp source pointed at the test server
124// ---------------------------------------------------------------------------124// ---------------------------------------------------------------------------
125 125
126const makeTestExecutor = (serverUrl: string) =>126const makeTestExecutor = (serverUrl: string, namespace = "pool-test") =>
127 createExecutor(127 createExecutor(
128 makeTestConfig({128 makeTestConfig({
129 plugins: [mcpPlugin()] as const,129 plugins: [mcpPlugin()] as const,
@@ -133,7 +133,8 @@ const makeTestExecutor = (serverUrl: string) => · 3 unmodified lines
133 executor.mcp.addSource({133 executor.mcp.addSource({
134 transport: "remote",134 transport: "remote",
135 scope: "test-scope",135 scope: "test-scope",
136 name: "pool-test",136 name: namespace,
137 namespace,
137 endpoint: serverUrl,138 endpoint: serverUrl,
138 }),139 }),
139 ),140 ),
@@ -207,4 +208,50 @@ describe("MCP connection pooling (regression)", () => { · 67 unmodified lines
207 expect(server.sessionCount()).toBe(baseline);208 expect(server.sessionCount()).toBe(baseline);
208 }),209 }),
209 );210 );
211
212 it.effect("different sources with the same endpoint use separate cached connections", () =>
213 Effect.gen(function* () {
214 const server = yield* serveMcpServer;
215 const executor = yield* createExecutor(
216 makeTestConfig({
217 plugins: [mcpPlugin()] as const,
218 }),
219 );
220 yield* executor.mcp.addSource({
221 transport: "remote",
222 scope: "test-scope",
223 name: "pool-a",
224 namespace: "pool_a",
225 endpoint: server.url,
226 });
227 yield* executor.mcp.addSource({
228 transport: "remote",
229 scope: "test-scope",
230 name: "pool-b",
231 namespace: "pool_b",
232 endpoint: server.url,
233 });
234
235 const tools = yield* executor.tools.list();
236 const echoTools = tools.filter((t) => t.name === "echo");
237 expect(echoTools).toHaveLength(2);
238
239 const sessionsAfterAddSource = server.sessionCount();
240
241 yield* executor.tools.invoke(
242 echoTools[0]!.id,
243 { value: "a" },
244 { onElicitation: "accept-all" },
245 );
246 const afterFirstInvoke = server.sessionCount();
247 expect(afterFirstInvoke).toBe(sessionsAfterAddSource + 1);
248
249 yield* executor.tools.invoke(
250 echoTools[1]!.id,
251 { value: "b" },
252 { onElicitation: "accept-all" },
253 );
254 expect(server.sessionCount()).toBe(afterFirstInvoke + 1);
255 }),
256 );
210});257});
packages/plugins/mcp/src/sdk/invoke.ts+51−11
@@ -35,16 +35,54 @@ const decodeArgsRecord = Schema.decodeUnknownOption(ArgsRecord); · 34 unmodified lines
35const argsRecord = (value: unknown): Record<string, unknown> =>35const argsRecord = (value: unknown): Record<string, unknown> =>
36 Option.getOrElse(decodeArgsRecord(value), () => ({}));36 Option.getOrElse(decodeArgsRecord(value), () => ({}));
37 37
38const connectionCacheKey = (sd: McpStoredSourceData, invokerScope: string): string =>38const stableJson = (value: unknown): string => {
39 sd.transport === "stdio"39 if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`;
40 ? `stdio:${sd.command}`40 if (value && typeof value === "object") {
41 : // Remote sources may resolve per-user secrets (OAuth tokens, header41 return `{${Object.entries(value as Record<string, unknown>)
42 // auth) via scope shadowing, so two users invoking the same source42 .sort(([a], [b]) => a.localeCompare(b))
43 // get different Authorization headers. The connection caches that43 .map(([key, entry]) => `${JSON.stringify(key)}:${stableJson(entry)}`)
44 // header in transport state, so the cache key must include the44 .join(",")}}`;
45 // invoking scope — otherwise user B re-uses user A's connection45 }
46 // (and user A's tokens).46 return JSON.stringify(value);
47 `remote:${invokerScope}:${sd.endpoint}`;47};
48
49const fingerprint = (value: unknown): string => {
50 const input = stableJson(value);
51 let hash = 0x811c9dc5;
52 for (let i = 0; i < input.length; i++) {
53 hash ^= input.charCodeAt(i);
54 hash = Math.imul(hash, 0x01000193);
55 }
56 return (hash >>> 0).toString(16).padStart(8, "0");
57};
58
59const connectionCacheKey = (input: {
60 readonly sourceData: McpStoredSourceData;
61 readonly invokerScope: string;
62 readonly sourceId: string;
63 readonly sourceScope: string;
64}): string => {
65 const sd = input.sourceData;
66 return sd.transport === "stdio"
67 ? `stdio:${fingerprint({
68 sourceId: input.sourceId,
69 sourceScope: input.sourceScope,
70 command: sd.command,
71 args: sd.args ?? [],
72 env: sd.env ?? {},
73 cwd: sd.cwd ?? null,
74 })}`
75 : `remote:${fingerprint({
76 sourceId: input.sourceId,
77 sourceScope: input.sourceScope,
78 invokerScope: input.invokerScope,
79 endpoint: sd.endpoint,
80 remoteTransport: sd.remoteTransport ?? "auto",
81 headers: sd.headers ?? {},
82 queryParams: sd.queryParams ?? {},
83 auth: sd.auth,
84 })}`;
85};
48 86
49// ---------------------------------------------------------------------------87// ---------------------------------------------------------------------------
50// Elicitation bridge — decode incoming MCP ElicitRequest, route through88// Elicitation bridge — decode incoming MCP ElicitRequest, route through
@@ -146,6 +184,8 @@ export interface InvokeMcpToolInput { · 95 unmodified lines
146 readonly toolName: string;184 readonly toolName: string;
147 readonly args: unknown;185 readonly args: unknown;
148 readonly sourceData: McpStoredSourceData;186 readonly sourceData: McpStoredSourceData;
187 readonly sourceId: string;
188 readonly sourceScope: string;
149 /** Innermost executor scope id at invoke time. Mixed into the189 /** Innermost executor scope id at invoke time. Mixed into the
150 * connection cache key so per-user OAuth/secret resolution doesn't190 * connection cache key so per-user OAuth/secret resolution doesn't
151 * collapse multiple users onto one shared connection. */191 * collapse multiple users onto one shared connection. */
@@ -162,7 +202,7 @@ export const invokeMcpTool = ( · 10 unmodified lines
162 const transport: string =202 const transport: string =
163 input.sourceData.transport === "stdio" ? "stdio" : (input.sourceData.remoteTransport ?? "auto");203 input.sourceData.transport === "stdio" ? "stdio" : (input.sourceData.remoteTransport ?? "auto");
164 return Effect.gen(function* () {204 return Effect.gen(function* () {
165 const cacheKey = connectionCacheKey(input.sourceData, input.invokerScope);205 const cacheKey = connectionCacheKey(input);
166 const args = argsRecord(input.args);206 const args = argsRecord(input.args);
167 207
168 // Register the connector for the cache lookup (side-channel pattern208 // Register the connector for the cache lookup (side-channel pattern
packages/plugins/mcp/src/sdk/plugin.ts+2−0
@@ -1581,6 +1581,8 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { · 1580 unmodified lines
1581 toolName: entry.binding.toolName,1581 toolName: entry.binding.toolName,
1582 args,1582 args,
1583 sourceData: sd,1583 sourceData: sd,
1584 sourceId: entry.namespace,
1585 sourceScope: toolScope,
1584 invokerScope: ctx.scopes[0]!.id,1586 invokerScope: ctx.scopes[0]!.id,
1585 resolveConnector: () =>1587 resolveConnector: () =>
1586 resolveConnectorInput(entry.namespace, toolScope, sd, ctx, allowStdio).pipe(1588 resolveConnectorInput(entry.namespace, toolScope, sd, ctx, allowStdio).pipe(