import { Duration, Effect, Exit, Schema, Scope, ScopedCache } from "effect"; import { refreshAuthorization, type OAuthClientProvider, type OAuthDiscoveryState,} from "@modelcontextprotocol/sdk/client/auth.js";import type { OAuthClientInformationMixed, OAuthMetadata, OAuthTokens,} from "@modelcontextprotocol/sdk/shared/auth.js"; import { ConnectionId, ConnectionRefreshError, CreateConnectionInput, ScopeId, SecretId, SourceDetectionResult, TokenMaterial, definePlugin, type ConnectionProvider, type ConnectionRefreshInput, type ConnectionRefreshResult, type PluginCtx, type StorageFailure,} from "@executor/sdk"; import { makeMcpStore, mcpSchema, type McpBindingStore, type McpStoredSource,} from "./binding-store";import { createMcpConnector, type ConnectorInput, type McpConnection,} from "./connection";import { discoverTools } from "./discover";import { McpConnectionError, McpOAuthError, McpToolDiscoveryError,} from "./errors";import { invokeMcpTool } from "./invoke";import { deriveMcpNamespace, type McpToolManifestEntry,} from "./manifest";import { exchangeMcpOAuthCode, startMcpOAuthAuthorization } from "./oauth";import { McpToolBinding, McpJsonObject, type McpConnectionAuth, type McpStoredSourceData,} from "./types"; import { SECRET_REF_PREFIX, type ConfigFileSink, type McpAuthConfig, type McpRemoteSourceConfig as McpRemoteConfigEntry, type McpStdioSourceConfig as McpStdioConfigEntry, type SourceConfig,} from "@executor/config"; // ---------------------------------------------------------------------------// Plugin config — discriminated union on transport// --------------------------------------------------------------------------- /** * Executor scope id that owns a newly-added MCP source row. Must be one * of the executor's configured scopes. Admins adding a shared server at * org scope pin here; per-user stdio sources can pin at the inner * scope. */type McpSourceScopeField = { readonly scope: string }; export interface McpRemoteSourceConfig extends McpSourceScopeField { readonly transport: "remote"; readonly name: string; readonly endpoint: string; readonly remoteTransport?: "streamable-http" | "sse" | "auto"; readonly queryParams?: Record<string, string>; readonly headers?: Record<string, string>; readonly namespace?: string; readonly auth?: McpConnectionAuth;} export interface McpStdioSourceConfig extends McpSourceScopeField { readonly transport: "stdio"; readonly name: string; readonly command: string; readonly args?: string[]; readonly env?: Record<string, string>; readonly cwd?: string; readonly namespace?: string;} export type McpSourceConfig = McpRemoteSourceConfig | McpStdioSourceConfig; // ---------------------------------------------------------------------------// Extension types// --------------------------------------------------------------------------- export interface McpOAuthStartInput { readonly endpoint: string; readonly redirectUrl: string; readonly queryParams?: Record<string, string> | null; /** * Executor scope id where the minted Connection will land. Defaults * to `ctx.scopes[0].id` (innermost) — for a per-user stack * `[user, org]` that pins the connection to the user scope so the * source's stored `connectionId` resolves per-user via shadowing. */ readonly tokenScope?: string; /** * Pre-decided SDK connection id the exchange will mint. Stable per * source (typically `mcp-oauth2-${namespace}`) so multiple users * signing in against the same MCP source all write to the same id at * their own scope — `ctx.connections.accessToken(id)` then resolves * innermost-first. */ readonly connectionId: string; /** * Source-level OAuth state captured by a previous user's flow. Pass * the values stored on the source's auth config to skip Dynamic * Client Registration — the new user's flow re-uses the same * client_id and discovery results. */ readonly clientInformation?: Record<string, unknown> | null; readonly authorizationServerUrl?: string | null; readonly resourceMetadataUrl?: string | null;} export interface McpOAuthStartResponse { readonly sessionId: string; readonly authorizationUrl: string;} export interface McpOAuthCompleteInput { readonly state: string; readonly code?: string; readonly error?: string;} export interface McpOAuthCompleteResponse { /** Id of the SDK Connection the exchange minted. The caller stores it * on the source's `oauth2` auth field and resolves tokens via * `ctx.connections`. */ readonly connectionId: string; readonly tokenType: string; readonly expiresAt: number | null; readonly scope: string | null; /** DCR client + discovery URLs captured during the flow. The UI * stores them on the source's auth config so subsequent users can * skip DCR and re-discovery. */ readonly clientInformation: Record<string, unknown> | null; readonly authorizationServerUrl: string | null; readonly resourceMetadataUrl: string | null;} export interface McpProbeResult { readonly connected: boolean; readonly requiresOAuth: boolean; readonly name: string; readonly namespace: string; readonly toolCount: number | null; readonly serverName: string | null;} export interface McpUpdateSourceInput { readonly name?: string; readonly endpoint?: string; readonly headers?: Record<string, string>; readonly queryParams?: Record<string, string>; readonly auth?: McpConnectionAuth;} // ---------------------------------------------------------------------------// Helpers// --------------------------------------------------------------------------- const toStoredSourceData = (config: McpSourceConfig): McpStoredSourceData => { if (config.transport === "stdio") { return { transport: "stdio", command: config.command, args: config.args, env: config.env, cwd: config.cwd, }; } return { transport: "remote", endpoint: config.endpoint, remoteTransport: config.remoteTransport ?? "auto", queryParams: config.queryParams, headers: config.headers, auth: config.auth ?? { kind: "none" }, };}; const normalizeNamespace = (config: McpSourceConfig): string => config.namespace ?? deriveMcpNamespace({ name: config.name, endpoint: config.transport === "remote" ? config.endpoint : undefined, command: config.transport === "stdio" ? config.command : undefined, }); const toBinding = (entry: McpToolManifestEntry): McpToolBinding => new McpToolBinding({ toolId: entry.toolId, toolName: entry.toolName, description: entry.description, inputSchema: entry.inputSchema, outputSchema: entry.outputSchema, }); interface OAuthProviderInputs { readonly accessToken: string; readonly tokenType: string; /** Source-level state — same for every user. Surfaced to the MCP * transport so internal DCR/discovery lookups find a populated * provider, but never used to mint tokens: refresh is owned by the * SDK via `ctx.connections.accessToken`. */ readonly clientInformation?: OAuthClientInformationMixed | null; readonly authorizationServerUrl?: string | null; readonly resourceMetadataUrl?: string | null; readonly endpoint: string;} const makeOAuthProvider = (inputs: OAuthProviderInputs): OAuthClientProvider => { const currentTokens: OAuthTokens = { access_token: inputs.accessToken, token_type: inputs.tokenType, }; let clientInformation: OAuthClientInformationMixed | undefined = inputs.clientInformation ?? undefined; let discoveryState: OAuthDiscoveryState | undefined = inputs.authorizationServerUrl || inputs.resourceMetadataUrl ? { authorizationServerUrl: inputs.authorizationServerUrl ?? new URL("/", inputs.endpoint).toString(), resourceMetadataUrl: inputs.resourceMetadataUrl ?? undefined, } : undefined; return { get redirectUrl() { return "http://localhost/oauth/callback"; }, get clientMetadata() { return { redirect_uris: ["http://localhost/oauth/callback"], grant_types: ["authorization_code", "refresh_token"] as string[], response_types: ["code"] as string[], token_endpoint_auth_method: "none" as const, client_name: "Executor", }; }, clientInformation: () => clientInformation, saveClientInformation: (ci) => { clientInformation = ci; }, tokens: () => currentTokens, // No-op: token refresh is driven by the SDK through // `ctx.connections.accessToken`, and the access token resolved at // `resolveConnectorInput` time is the only one this provider ever // hands to the transport. A 401 from the server flips the // invokeMcpTool retry path, which invalidates the cached connection // and re-resolves — picking up a freshly refreshed token. saveTokens: () => undefined, redirectToAuthorization: async () => { throw new Error("MCP OAuth re-authorization required"); }, saveCodeVerifier: () => {}, codeVerifier: () => { throw new Error("No active PKCE verifier"); }, saveDiscoveryState: (s) => { discoveryState = s; }, discoveryState: () => discoveryState, };}; // ---------------------------------------------------------------------------// Connection `provider_state` shape for mcp-oauth2.//// Carries everything the SDK hands to the refresh handler so it can// re-hit the token endpoint without re-reading plugin storage. The DCR// client info is NOT sensitive on its own — the MCP SDK's DCR flow uses// `token_endpoint_auth_method: "none"` (PKCE), so there's no// client_secret to protect. Having it here lets the refresh handler// skip discovery entirely.// --------------------------------------------------------------------------- const MCP_OAUTH2_PROVIDER_KEY = "mcp:oauth2" as const; const OAuth2ProviderState = Schema.Struct({ endpoint: Schema.String, tokenType: Schema.String, // Nullable to accommodate early-onboarded sources where DCR client // registration was never persisted back onto the source. Refresh fails // fast on null and surfaces a "re-sign-in required" error, which the UI // turns into a Reconnect prompt. clientInformation: Schema.NullOr(McpJsonObject), authorizationServerUrl: Schema.NullOr(Schema.String), authorizationServerMetadata: Schema.NullOr(McpJsonObject), resourceMetadataUrl: Schema.NullOr(Schema.String), resourceMetadata: Schema.NullOr(McpJsonObject),});type OAuth2ProviderState = typeof OAuth2ProviderState.Type; const encodeProviderState = Schema.encodeSync(OAuth2ProviderState);const decodeProviderState = Schema.decodeUnknownSync(OAuth2ProviderState); const toProviderStateRecord = ( state: OAuth2ProviderState,): Record<string, unknown> => encodeProviderState(state) as unknown as Record<string, unknown>; const remoteConnectionError = (message: string) => new McpConnectionError({ transport: "remote", message }); const mcpOAuthError = (message: string) => new McpOAuthError({ message }); const mcpDiscoveryError = (message: string) => new McpToolDiscoveryError({ stage: "list_tools", message }); // ---------------------------------------------------------------------------// Shared connector resolution — reads secrets, builds stdio/remote input// --------------------------------------------------------------------------- const resolveConnectorInput = ( sd: McpStoredSourceData, ctx: PluginCtx<McpBindingStore>, allowStdio: boolean,): Effect.Effect<ConnectorInput, McpConnectionError | StorageFailure> => { if (sd.transport === "stdio") { if (!allowStdio) { return Effect.fail( new McpConnectionError({ transport: "stdio", message: "MCP stdio transport is disabled. Enable it by passing `dangerouslyAllowStdioMCP: true` to mcpPlugin() — only safe for trusted local contexts.", }), ); } return Effect.succeed({ transport: "stdio" as const, command: sd.command, args: sd.args, env: sd.env, cwd: sd.cwd, }); } return Effect.gen(function* () { const headers: Record<string, string> = { ...sd.headers }; let authProvider: OAuthClientProvider | undefined; const auth = sd.auth; if (auth.kind === "header") { const val = yield* ctx.secrets.get(auth.secretId); if (val === null) { return yield* Effect.fail( remoteConnectionError(`Failed to resolve secret "${auth.secretId}"`), ); } headers[auth.headerName] = auth.prefix ? `${auth.prefix}${val}` : val; } else if (auth.kind === "oauth2") { // `accessToken(id)` handles refresh internally — by the time we // hand the value to the MCP transport it's guaranteed fresh. The // connection row carries the DCR client + discovery metadata in // `providerState`, but we read it off the connection here only // for the `OAuthClientProvider` surface; the actual token // material is what matters at the wire. const accessToken = yield* ctx.connections .accessToken(auth.connectionId) .pipe( Effect.mapError((err) => remoteConnectionError( `Failed to resolve OAuth connection "${auth.connectionId}": ${ "message" in err ? (err as { message: string }).message : String(err) }`, ), ), ); const connection = yield* ctx.connections.get(auth.connectionId); const providerState: OAuth2ProviderState | null = connection?.providerState ? yield* Effect.sync(() => { try { return decodeProviderState(connection.providerState); } catch { return null; } }) : null; authProvider = makeOAuthProvider({ accessToken, tokenType: providerState?.tokenType ?? "Bearer", clientInformation: (providerState?.clientInformation as OAuthClientInformationMixed | undefined) ?? null, authorizationServerUrl: providerState?.authorizationServerUrl ?? null, resourceMetadataUrl: providerState?.resourceMetadataUrl ?? null, endpoint: sd.endpoint, }); } return { transport: "remote" as const, endpoint: sd.endpoint, remoteTransport: sd.remoteTransport, queryParams: sd.queryParams, headers: Object.keys(headers).length > 0 ? headers : undefined, authProvider, }; });}; // ---------------------------------------------------------------------------// Connection cache — kept as plugin-module state so both invokeTool and// the close hook see the same ScopedCache instance. The ScopedCache's// lookup key is the stringified stored source data identity.// --------------------------------------------------------------------------- interface McpRuntime { readonly connectionCache: ScopedCache.ScopedCache< string, McpConnection, McpConnectionError >; readonly pendingConnectors: Map< string, Effect.Effect<McpConnection, McpConnectionError> >; readonly cacheScope: Scope.CloseableScope;} const makeRuntime = (): Effect.Effect<McpRuntime, never> => Effect.gen(function* () { const cacheScope = yield* Scope.make(); const pendingConnectors = new Map< string, Effect.Effect<McpConnection, McpConnectionError> >(); const connectionCache = yield* ScopedCache.make({ lookup: (key: string) => Effect.acquireRelease( Effect.suspend(() => { const connector = pendingConnectors.get(key); if (!connector) { return Effect.fail( new McpConnectionError({ transport: "auto", message: `No pending connector for key: ${key}`, }), ); } return connector; }), (connection) => Effect.promise(() => connection.close().catch(() => {})), ), capacity: 64, timeToLive: Duration.minutes(5), }).pipe(Scope.extend(cacheScope)); return { connectionCache, pendingConnectors, cacheScope }; }); // ---------------------------------------------------------------------------// Plugin factory// --------------------------------------------------------------------------- export interface McpPluginOptions { /** * Allow configuring stdio-transport MCP sources. Off by default. * * Stdio sources spawn a local subprocess that inherits the parent * `process.env`. Only enable for trusted single-user contexts. */ readonly dangerouslyAllowStdioMCP?: boolean; /** If provided, source add/remove is mirrored to executor.jsonc * (best-effort — file errors are logged, not raised). */ readonly configFile?: ConfigFileSink;} const secretRef = (id: string): string => `${SECRET_REF_PREFIX}${id}`;Lines 1–500 of 1433. Load more
import { Duration, Effect, Exit, Schema, Scope, ScopedCache } from "effect"; import { refreshAuthorization, type OAuthClientProvider, type OAuthDiscoveryState, } from "@modelcontextprotocol/sdk/client/auth.js"; import type { OAuthClientInformationMixed, OAuthMetadata, OAuthTokens, } from "@modelcontextprotocol/sdk/shared/auth.js"; import { ConnectionId, ConnectionRefreshError, CreateConnectionInput, ScopeId, SecretId, SourceDetectionResult, TokenMaterial, definePlugin, type ConnectionProvider, type ConnectionRefreshInput, type ConnectionRefreshResult, type PluginCtx, type StorageFailure, } from "@executor/sdk"; import { makeMcpStore, mcpSchema, type McpBindingStore, type McpStoredSource, } from "./binding-store"; import { createMcpConnector, type ConnectorInput, type McpConnection, } from "./connection"; import { discoverTools } from "./discover"; import { McpConnectionError, McpOAuthError, McpToolDiscoveryError, } from "./errors"; import { invokeMcpTool } from "./invoke"; import { deriveMcpNamespace, type McpToolManifestEntry, } from "./manifest"; import { exchangeMcpOAuthCode, startMcpOAuthAuthorization } from "./oauth"; import { McpToolBinding, McpJsonObject, type McpConnectionAuth, type McpStoredSourceData, } from "./types"; import { SECRET_REF_PREFIX, type ConfigFileSink, type McpAuthConfig, type McpRemoteSourceConfig as McpRemoteConfigEntry, type McpStdioSourceConfig as McpStdioConfigEntry, type SourceConfig, } from "@executor/config"; // --------------------------------------------------------------------------- // Plugin config — discriminated union on transport // --------------------------------------------------------------------------- /** * Executor scope id that owns a newly-added MCP source row. Must be one * of the executor's configured scopes. Admins adding a shared server at * org scope pin here; per-user stdio sources can pin at the inner * scope. */ type McpSourceScopeField = { readonly scope: string }; export interface McpRemoteSourceConfig extends McpSourceScopeField { readonly transport: "remote"; readonly name: string; readonly endpoint: string; readonly remoteTransport?: "streamable-http" | "sse" | "auto"; readonly queryParams?: Record<string, string>; readonly headers?: Record<string, string>; readonly namespace?: string; readonly auth?: McpConnectionAuth; } export interface McpStdioSourceConfig extends McpSourceScopeField { readonly transport: "stdio"; readonly name: string; readonly command: string; readonly args?: string[]; readonly env?: Record<string, string>; readonly cwd?: string; readonly namespace?: string; } export type McpSourceConfig = McpRemoteSourceConfig | McpStdioSourceConfig; // --------------------------------------------------------------------------- // Extension types // --------------------------------------------------------------------------- export interface McpOAuthStartInput { readonly endpoint: string; readonly redirectUrl: string; readonly queryParams?: Record<string, string> | null; /** * Executor scope id where the minted Connection will land. Defaults * to `ctx.scopes[0].id` (innermost) — for a per-user stack * `[user, org]` that pins the connection to the user scope so the * source's stored `connectionId` resolves per-user via shadowing. */ readonly tokenScope?: string; /** * Pre-decided SDK connection id the exchange will mint. Stable per * source (typically `mcp-oauth2-${namespace}`) so multiple users * signing in against the same MCP source all write to the same id at * their own scope — `ctx.connections.accessToken(id)` then resolves * innermost-first. */ readonly connectionId: string; /** * Source-level OAuth state captured by a previous user's flow. Pass * the values stored on the source's auth config to skip Dynamic * Client Registration — the new user's flow re-uses the same * client_id and discovery results. */ readonly clientInformation?: Record<string, unknown> | null; readonly authorizationServerUrl?: string | null; readonly resourceMetadataUrl?: string | null; } export interface McpOAuthStartResponse { readonly sessionId: string; readonly authorizationUrl: string; } export interface McpOAuthCompleteInput { readonly state: string; readonly code?: string; readonly error?: string; } export interface McpOAuthCompleteResponse { /** Id of the SDK Connection the exchange minted. The caller stores it * on the source's `oauth2` auth field and resolves tokens via * `ctx.connections`. */ readonly connectionId: string; readonly tokenType: string; readonly expiresAt: number | null; readonly scope: string | null; /** DCR client + discovery URLs captured during the flow. The UI * stores them on the source's auth config so subsequent users can * skip DCR and re-discovery. */ readonly clientInformation: Record<string, unknown> | null; readonly authorizationServerUrl: string | null; readonly resourceMetadataUrl: string | null; } export interface McpProbeResult { readonly connected: boolean; readonly requiresOAuth: boolean; readonly name: string; readonly namespace: string; readonly toolCount: number | null; readonly serverName: string | null; } export interface McpUpdateSourceInput { readonly name?: string; readonly endpoint?: string; readonly headers?: Record<string, string>; readonly queryParams?: Record<string, string>; readonly auth?: McpConnectionAuth; } // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- const toStoredSourceData = (config: McpSourceConfig): McpStoredSourceData => { if (config.transport === "stdio") { return { transport: "stdio", command: config.command, args: config.args, env: config.env, cwd: config.cwd, }; } return { transport: "remote", endpoint: config.endpoint, remoteTransport: config.remoteTransport ?? "auto", queryParams: config.queryParams, headers: config.headers, auth: config.auth ?? { kind: "none" }, }; }; const normalizeNamespace = (config: McpSourceConfig): string => config.namespace ?? deriveMcpNamespace({ name: config.name, endpoint: config.transport === "remote" ? config.endpoint : undefined, command: config.transport === "stdio" ? config.command : undefined, }); const toBinding = (entry: McpToolManifestEntry): McpToolBinding => new McpToolBinding({ toolId: entry.toolId, toolName: entry.toolName, description: entry.description, inputSchema: entry.inputSchema, outputSchema: entry.outputSchema, }); interface OAuthProviderInputs { readonly accessToken: string; readonly tokenType: string; /** Source-level state — same for every user. Surfaced to the MCP * transport so internal DCR/discovery lookups find a populated * provider, but never used to mint tokens: refresh is owned by the * SDK via `ctx.connections.accessToken`. */ readonly clientInformation?: OAuthClientInformationMixed | null; readonly authorizationServerUrl?: string | null; readonly resourceMetadataUrl?: string | null; readonly endpoint: string; } const makeOAuthProvider = (inputs: OAuthProviderInputs): OAuthClientProvider => { const currentTokens: OAuthTokens = { access_token: inputs.accessToken, token_type: inputs.tokenType, }; let clientInformation: OAuthClientInformationMixed | undefined = inputs.clientInformation ?? undefined; let discoveryState: OAuthDiscoveryState | undefined = inputs.authorizationServerUrl || inputs.resourceMetadataUrl ? { authorizationServerUrl: inputs.authorizationServerUrl ?? new URL("/", inputs.endpoint).toString(), resourceMetadataUrl: inputs.resourceMetadataUrl ?? undefined, } : undefined; return { get redirectUrl() { return "http://localhost/oauth/callback"; }, get clientMetadata() { return { redirect_uris: ["http://localhost/oauth/callback"], grant_types: ["authorization_code", "refresh_token"] as string[], response_types: ["code"] as string[], token_endpoint_auth_method: "none" as const, client_name: "Executor", }; }, clientInformation: () => clientInformation, saveClientInformation: (ci) => { clientInformation = ci; }, tokens: () => currentTokens, // No-op: token refresh is driven by the SDK through // `ctx.connections.accessToken`, and the access token resolved at // `resolveConnectorInput` time is the only one this provider ever // hands to the transport. A 401 from the server flips the // invokeMcpTool retry path, which invalidates the cached connection // and re-resolves — picking up a freshly refreshed token. saveTokens: () => undefined, redirectToAuthorization: async () => { throw new Error("MCP OAuth re-authorization required"); }, saveCodeVerifier: () => {}, codeVerifier: () => { throw new Error("No active PKCE verifier"); }, saveDiscoveryState: (s) => { discoveryState = s; }, discoveryState: () => discoveryState, }; }; // --------------------------------------------------------------------------- // Connection `provider_state` shape for mcp-oauth2. // // Carries everything the SDK hands to the refresh handler so it can // re-hit the token endpoint without re-reading plugin storage. The DCR // client info is NOT sensitive on its own — the MCP SDK's DCR flow uses // `token_endpoint_auth_method: "none"` (PKCE), so there's no // client_secret to protect. Having it here lets the refresh handler // skip discovery entirely. // --------------------------------------------------------------------------- const MCP_OAUTH2_PROVIDER_KEY = "mcp:oauth2" as const; const OAuth2ProviderState = Schema.Struct({ endpoint: Schema.String, tokenType: Schema.String, // Nullable to accommodate early-onboarded sources where DCR client // registration was never persisted back onto the source. Refresh fails // fast on null and surfaces a "re-sign-in required" error, which the UI // turns into a Reconnect prompt. clientInformation: Schema.NullOr(McpJsonObject), authorizationServerUrl: Schema.NullOr(Schema.String), authorizationServerMetadata: Schema.NullOr(McpJsonObject), resourceMetadataUrl: Schema.NullOr(Schema.String), resourceMetadata: Schema.NullOr(McpJsonObject), }); type OAuth2ProviderState = typeof OAuth2ProviderState.Type; const encodeProviderState = Schema.encodeSync(OAuth2ProviderState); const decodeProviderState = Schema.decodeUnknownSync(OAuth2ProviderState); const toProviderStateRecord = ( state: OAuth2ProviderState, ): Record<string, unknown> => encodeProviderState(state) as unknown as Record<string, unknown>; const remoteConnectionError = (message: string) => new McpConnectionError({ transport: "remote", message }); const mcpOAuthError = (message: string) => new McpOAuthError({ message }); const mcpDiscoveryError = (message: string) => new McpToolDiscoveryError({ stage: "list_tools", message }); // --------------------------------------------------------------------------- // Shared connector resolution — reads secrets, builds stdio/remote input // --------------------------------------------------------------------------- const resolveConnectorInput = ( sd: McpStoredSourceData, ctx: PluginCtx<McpBindingStore>, allowStdio: boolean, ): Effect.Effect<ConnectorInput, McpConnectionError | StorageFailure> => { if (sd.transport === "stdio") { if (!allowStdio) { return Effect.fail( new McpConnectionError({ transport: "stdio", message: "MCP stdio transport is disabled. Enable it by passing `dangerouslyAllowStdioMCP: true` to mcpPlugin() — only safe for trusted local contexts.", }), ); } return Effect.succeed({ transport: "stdio" as const, command: sd.command, args: sd.args, env: sd.env, cwd: sd.cwd, }); } return Effect.gen(function* () { const headers: Record<string, string> = { ...sd.headers }; let authProvider: OAuthClientProvider | undefined; const auth = sd.auth; if (auth.kind === "header") { const val = yield* ctx.secrets.get(auth.secretId); if (val === null) { return yield* Effect.fail( remoteConnectionError(`Failed to resolve secret "${auth.secretId}"`), ); } headers[auth.headerName] = auth.prefix ? `${auth.prefix}${val}` : val; } else if (auth.kind === "oauth2") { // `accessToken(id)` handles refresh internally — by the time we // hand the value to the MCP transport it's guaranteed fresh. The // connection row carries the DCR client + discovery metadata in // `providerState`, but we read it off the connection here only // for the `OAuthClientProvider` surface; the actual token // material is what matters at the wire. const accessToken = yield* ctx.connections .accessToken(auth.connectionId) .pipe( Effect.mapError((err) => remoteConnectionError( `Failed to resolve OAuth connection "${auth.connectionId}": ${ "message" in err ? (err as { message: string }).message : String(err) }`, ), ), ); const connection = yield* ctx.connections.get(auth.connectionId); const providerState: OAuth2ProviderState | null = connection?.providerState ? yield* Effect.sync(() => { try { return decodeProviderState(connection.providerState); } catch { return null; } }) : null; authProvider = makeOAuthProvider({ accessToken, tokenType: providerState?.tokenType ?? "Bearer", clientInformation: (providerState?.clientInformation as OAuthClientInformationMixed | undefined) ?? null, authorizationServerUrl: providerState?.authorizationServerUrl ?? null, resourceMetadataUrl: providerState?.resourceMetadataUrl ?? null, endpoint: sd.endpoint, }); } return { transport: "remote" as const, endpoint: sd.endpoint, remoteTransport: sd.remoteTransport, queryParams: sd.queryParams, headers: Object.keys(headers).length > 0 ? headers : undefined, authProvider, }; }); }; // --------------------------------------------------------------------------- // Connection cache — kept as plugin-module state so both invokeTool and // the close hook see the same ScopedCache instance. The ScopedCache's // lookup key is the stringified stored source data identity. // --------------------------------------------------------------------------- interface McpRuntime { readonly connectionCache: ScopedCache.ScopedCache< string, McpConnection, McpConnectionError >; readonly pendingConnectors: Map< string, Effect.Effect<McpConnection, McpConnectionError> >; readonly cacheScope: Scope.CloseableScope; } const makeRuntime = (): Effect.Effect<McpRuntime, never> => Effect.gen(function* () { const cacheScope = yield* Scope.make(); const pendingConnectors = new Map< string, Effect.Effect<McpConnection, McpConnectionError> >(); const connectionCache = yield* ScopedCache.make({ lookup: (key: string) => Effect.acquireRelease( Effect.suspend(() => { const connector = pendingConnectors.get(key); if (!connector) { return Effect.fail( new McpConnectionError({ transport: "auto", message: `No pending connector for key: ${key}`, }), ); } return connector; }), (connection) => Effect.promise(() => connection.close().catch(() => {})), ), capacity: 64, timeToLive: Duration.minutes(5), }).pipe(Scope.extend(cacheScope)); return { connectionCache, pendingConnectors, cacheScope }; }); // --------------------------------------------------------------------------- // Plugin factory // --------------------------------------------------------------------------- export interface McpPluginOptions { /** * Allow configuring stdio-transport MCP sources. Off by default. * * Stdio sources spawn a local subprocess that inherits the parent * `process.env`. Only enable for trusted single-user contexts. */ readonly dangerouslyAllowStdioMCP?: boolean; /** If provided, source add/remove is mirrored to executor.jsonc * (best-effort — file errors are logged, not raised). */ readonly configFile?: ConfigFileSink; } const secretRef = (id: string): string => `${SECRET_REF_PREFIX}${id}`;