import { Duration, Effect, Exit, Result, Scope, ScopedCache } from "effect"; import type { OAuthClientProvider } from "@modelcontextprotocol/sdk/client/auth.js"; import { McpGroup } from "../api/group";import { McpExtensionService, McpHandlers } from "../api/handlers"; import {  SourceDetectionResult,  definePlugin,  resolveSecretBackedMap as resolveSharedSecretBackedMap,  type PluginCtx,  type StorageFailure,  type ToolAnnotations,} from "@executor-js/sdk/core"; import {  makeMcpStore,  mcpSchema,  type McpBindingStore,  type McpStoredSource,} from "./binding-store";import { createMcpConnector, type ConnectorInput, type McpConnection } from "./connection";import { discoverTools } from "./discover";import { McpConnectionError, McpToolDiscoveryError } from "./errors";import { invokeMcpTool } from "./invoke";import { deriveMcpNamespace, type McpToolManifest, type McpToolManifestEntry } from "./manifest";import { probeMcpEndpointShape } from "./probe-shape";import {  McpToolBinding,  type McpConnectionAuth,  type SecretBackedValue,  type McpStoredSourceData,} from "./types"; import {  SECRET_REF_PREFIX,  type ConfigFileSink,  type McpAuthConfig,  type McpRemoteSourceConfig as McpRemoteConfigEntry,  type McpStdioSourceConfig as McpStdioConfigEntry,  type SourceConfig,} from "@executor-js/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, SecretBackedValue>;  readonly headers?: Record<string, SecretBackedValue>;  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// --------------------------------------------------------------------------- // OAuth start/complete/callback moved to the shared// `/scopes/:scopeId/oauth/*` surface in `@executor-js/api` — no// plugin-specific types needed here. 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, SecretBackedValue>;  readonly queryParams?: Record<string, SecretBackedValue>;  readonly auth?: McpConnectionAuth;} export interface McpProbeEndpointInput {  readonly endpoint: string;  readonly headers?: Record<string, SecretBackedValue>;  readonly queryParams?: Record<string, SecretBackedValue>;} // ---------------------------------------------------------------------------// 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,    annotations: entry.annotations,  }); // ---------------------------------------------------------------------------// MCP-SDK OAuth provider adapter — builds the `OAuthClientProvider` the// MCP SDK's StreamableHTTP/SSE transports want, wrapping a pre-resolved// access token.//// Refresh is NOT driven through this provider — `ctx.connections.access// Token` owns that lifecycle at the core level via the canonical// "oauth2" ConnectionProvider. This adapter only injects the current// token into the transport's Authorization header and fails loudly if// the SDK ever tries to initiate a new OAuth flow (which would bypass// our refresh machinery).// ---------------------------------------------------------------------------const makeOAuthProvider = (accessToken: string): OAuthClientProvider => ({  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: () => undefined,  saveClientInformation: () => undefined,  tokens: () => ({ access_token: accessToken, token_type: "Bearer" }),  saveTokens: () => undefined,  redirectToAuthorization: async () => {    throw new Error("MCP OAuth re-authorization required");  },  saveCodeVerifier: () => undefined,  codeVerifier: () => {    throw new Error("No active PKCE verifier");  },  saveDiscoveryState: () => undefined,  discoveryState: () => undefined,}); const remoteConnectionError = (message: string) =>  new McpConnectionError({ transport: "remote", message }); const mcpDiscoveryError = (message: string) =>  new McpToolDiscoveryError({ stage: "list_tools", message }); const resolveSecretBackedMap = (  values: Record<string, SecretBackedValue> | undefined,  ctx: PluginCtx<McpBindingStore>,): Effect.Effect<Record<string, string> | undefined, McpConnectionError | StorageFailure> =>  resolveSharedSecretBackedMap({    values,    getSecret: ctx.secrets.get,    onMissing: (_name, value) =>      remoteConnectionError(`Failed to resolve secret "${value.secretId}"`),    onError: (err, _name, value) =>      "_tag" in err && err._tag === "SecretOwnedByConnectionError"        ? remoteConnectionError(`Failed to resolve secret "${value.secretId}"`)        : err,  }).pipe(    Effect.mapError((err) =>      "_tag" in err && err._tag === "SecretOwnedByConnectionError"        ? remoteConnectionError("Failed to resolve secret")        : err,    ),  ); const plainStringMap = (  values: Record<string, SecretBackedValue> | undefined,): Record<string, string> | undefined => {  if (!values) return undefined;  const entries = Object.entries(values).filter(    (entry): entry is [string, string] => typeof entry[1] === "string",  );  return entries.length > 0 ? Object.fromEntries(entries) : undefined;}; // ---------------------------------------------------------------------------// 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 resolvedHeaders = yield* resolveSecretBackedMap(sd.headers, ctx);    const resolvedQueryParams = yield* resolveSecretBackedMap(sd.queryParams, ctx);    const headers: Record<string, string> = { ...(resolvedHeaders ?? {}) };    let authProvider: OAuthClientProvider | undefined;     const auth = sd.auth;    if (auth.kind === "header") {      const val = yield* ctx.secrets        .get(auth.secretId)        .pipe(          Effect.mapError((err) =>            "_tag" in err && err._tag === "SecretOwnedByConnectionError"              ? remoteConnectionError(`Failed to resolve secret "${auth.secretId}"`)              : err,          ),        );      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 canonical `"oauth2"` ConnectionProvider registered by      // core owns the refresh lifecycle; we just wrap the current      // token for the SDK's transport.      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)              }`,            ),          ),        );      authProvider = makeOAuthProvider(accessToken);    }     return {      transport: "remote" as const,      endpoint: sd.endpoint,      remoteTransport: sd.remoteTransport,      queryParams: resolvedQueryParams,      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.Closeable;} 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.provide(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}`; const authToConfig = (auth: McpConnectionAuth | undefined): McpAuthConfig | undefined => {  if (!auth) return undefined;  if (auth.kind === "none") return { kind: "none" };  if (auth.kind === "header") {    return {      kind: "header",      headerName: auth.headerName,      secret: secretRef(auth.secretId),      prefix: auth.prefix,    };  }  return {    kind: "oauth2",    connectionId: auth.connectionId,  };}; const toMcpConfigEntry = (  namespace: string,  sourceName: string,  config: McpSourceConfig,): SourceConfig => {  if (config.transport === "stdio") {    const entry: McpStdioConfigEntry = {      kind: "mcp",      transport: "stdio",      name: sourceName,      command: config.command,      args: config.args,      env: config.env,      cwd: config.cwd,      namespace,    };    return entry;  }  const entry: McpRemoteConfigEntry = {    kind: "mcp",    transport: "remote",    name: sourceName,    endpoint: config.endpoint,    remoteTransport: config.remoteTransport,    queryParams: plainStringMap(config.queryParams),    headers: plainStringMap(config.headers),    namespace,    auth: authToConfig(config.auth),  };  return entry;}; export const mcpPlugin = definePlugin((options?: McpPluginOptions) => {  const allowStdio = options?.dangerouslyAllowStdioMCP ?? false;  // Per-plugin-instance runtime holder. Captured by closures in  // `extension`, `invokeTool`, and `close`, so all three see the same  // connection cache across a single createExecutor lifecycle.  const runtimeRef: { current: McpRuntime | null } = { current: null };   const ensureRuntime = (): Effect.Effect<McpRuntime, never> =>    runtimeRef.current      ? Effect.succeed(runtimeRef.current)      : makeRuntime().pipe(          Effect.tap((rt) =>            Effect.sync(() => {              runtimeRef.current = rt;            }),          ),        );   return {    id: "mcp" as const,    packageName: "@executor-js/plugin-mcp",    schema: mcpSchema,    storage: (deps): McpBindingStore => makeMcpStore(deps),     extension: (ctx) => {      const probeEndpoint = (input: string | McpProbeEndpointInput) =>        Effect.gen(function* () {          const endpoint = typeof input === "string" ? input : input.endpoint;          const trimmed = endpoint.trim();          if (!trimmed) {            return yield* Effect.fail(remoteConnectionError("Endpoint URL is required"));          }           const name = yield* Effect.try({            try: () => new URL(trimmed).hostname,            catch: () => "mcp",          }).pipe(            Effect.orElseSucceed(() => "mcp"),          );          const namespace = deriveMcpNamespace({ endpoint: trimmed });           const probeHeaders =            typeof input === "string"              ? undefined              : yield* resolveSecretBackedMap(input.headers, ctx);          const probeQueryParams =            typeof input === "string"              ? undefined              : yield* resolveSecretBackedMap(input.queryParams, ctx);           const connector = createMcpConnector({            transport: "remote",            endpoint: trimmed,            headers: probeHeaders,            queryParams: probeQueryParams,          });           const result = yield* discoverTools(connector).pipe(            Effect.map((m) => ({ ok: true as const, manifest: m })),            Effect.catch(() => Effect.succeed({ ok: false as const, manifest: null })),            Effect.withSpan("mcp.plugin.discover_tools"),          );           if (result.ok && result.manifest) {            return {              connected: true,              requiresOAuth: false,              name: result.manifest.server?.name ?? name,              namespace,              toolCount: result.manifest.tools.length,              serverName: result.manifest.server?.name ?? null,            } satisfies McpProbeResult;          }           // Before asking the core OAuth service to look for metadata,          // confirm the endpoint actually speaks MCP. An OAuth-protected          // non-MCP service (e.g. a GraphQL API whose host publishes          // RFC 9728 + 8414 metadata) would otherwise pass the OAuth          // probe and be misclassified as MCP. The shape probe rejects          // anything whose initialize response isn't 2xx or 401+Bearer.          const shape = yield* probeMcpEndpointShape(trimmed, {

Lines 1–500 of 1061. Load more