import type { ExecutionEnvelope, ExecutionInteraction, SqlControlPlaneRuntime,} from "@executor/control-plane";import { EXECUTOR_SOURCES_ADD_HELP_LINES, ExecutionIdSchema, RuntimeExecutionResolverService, createExecution, getExecution, resumeExecution,} from "@executor/control-plane";import * as Effect from "effect/Effect";import * as Cause from "effect/Cause";import * as Exit from "effect/Exit";import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js";import { z } from "zod/v4"; const pollingIntervalMs = 200; const executeInputSchema = { code: z.string().trim().min(1),}; const resumeInputSchema = { resumePayload: z.object({ executionId: z.string().trim().min(1), }), response: z.object({ action: z.enum(["accept", "decline", "cancel"]), content: z.record(z.string(), z.unknown()).optional(), }).optional(),}; type ParsedInteractionPayload = { mode: "form" | "url"; message: string; url?: string; requestedSchema?: Record<string, unknown>; elicitationId?: string;}; type ResumePayload = { executionId: string;}; type ResumeResponseInput = { action: "accept" | "decline" | "cancel"; content?: Record<string, unknown>;}; type ExecutorMcpToolResult = { content: Array<{ type: "text"; text: string }>; structuredContent: Record<string, unknown>; isError?: boolean;}; export type ExecutorMcpRequestHandler = { handleRequest: (request: Request) => Promise<Response>; close: () => Promise<void>;}; const isRecord = (value: unknown): value is Record<string, unknown> => typeof value === "object" && value !== null && !Array.isArray(value); const parseJsonValue = (value: string | null): unknown => { if (value === null) { return null; } try { return JSON.parse(value); } catch { return value; }}; const maxResultPreviewChars = 30_000; const truncateText = (value: string, maxLength: number): string => value.length > maxLength ? `${value.slice(0, maxLength)}\n... [result preview truncated ${value.length - maxLength} chars]` : value; const formatResultPreview = (resultJson: string): string => { try { const parsed = JSON.parse(resultJson) as unknown; const serialized = typeof parsed === "string" ? parsed : JSON.stringify(parsed, null, 2) ?? String(parsed); return truncateText(serialized, maxResultPreviewChars); } catch { return truncateText(resultJson, maxResultPreviewChars); }}; const parseInteractionPayload = ( interaction: ExecutionInteraction,): ParsedInteractionPayload | null => { try { const parsed = JSON.parse(interaction.payloadJson) as { elicitation?: { mode?: "form" | "url"; message?: string; url?: string; requestedSchema?: Record<string, unknown>; elicitationId?: string; id?: string; }; }; if (!parsed.elicitation || typeof parsed.elicitation.message !== "string") { return null; } return { mode: parsed.elicitation.mode === "url" ? "url" : "form", message: parsed.elicitation.message, url: parsed.elicitation.url, requestedSchema: isRecord(parsed.elicitation.requestedSchema) ? parsed.elicitation.requestedSchema : undefined, elicitationId: typeof parsed.elicitation.elicitationId === "string" ? parsed.elicitation.elicitationId : typeof parsed.elicitation.id === "string" ? parsed.elicitation.id : undefined, }; } catch { return null; }}; const runControlPlane = async <A, E, R>( runtime: SqlControlPlaneRuntime, effect: Effect.Effect<A, E, R>,): Promise<A> => { const exit = await Effect.runPromiseExit( effect.pipe(Effect.provide(runtime.runtimeLayer)) as Effect.Effect<A, E, never>, ); if (Exit.isSuccess(exit)) { return exit.value; } const error = Cause.squash(exit.cause); if (error instanceof Error) { // Preserve the original error with its stack trace and message rather // than wrapping it in an opaque FiberFailure. throw error; } throw error;}; const sleep = (ms: number) => new Promise<void>((resolve) => { setTimeout(resolve, ms); }); const supportsManagedElicitation = (server: McpServer): boolean => { const capabilities = server.server.getClientCapabilities(); return Boolean(capabilities?.elicitation?.form) && Boolean(capabilities?.elicitation?.url);}; type CatalogLike = { listNamespaces: (input: { limit: number }) => Effect.Effect< ReadonlyArray<{ namespace: string; displayName?: string }>, unknown >;}; const buildExecuteWorkflowText = (namespaces: readonly string[] = []): string => [ "Execute TypeScript in sandbox; call tools via discovery workflow.", ...(namespaces.length > 0 ? [ "Available namespaces:", ...namespaces.map((namespace) => `- ${namespace}`), ] : []), "Workflow:", '1) const matches = await tools.discover({ query: "<intent>", limit: 12 });', "2) const details = await tools.describe.tool({ path, includeSchemas: true });", "3) Call selected tools.<path>(input).", '4) To connect a source, call tools.executor.sources.add(...) for MCP, OpenAPI, or GraphQL APIs.', ...EXECUTOR_SOURCES_ADD_HELP_LINES, "5) If execution pauses for interaction, resume it with the returned resumePayload or the available resume flow.", "Do not use fetch; use tools.* only.", ].join("\n"); const defaultExecuteDescription = buildExecuteWorkflowText(); const loadExecuteDescription = (runtime: SqlControlPlaneRuntime): Promise<string> => runControlPlane( runtime, Effect.gen(function* () { const resolveExecutionEnvironment = yield* RuntimeExecutionResolverService; const environment = yield* resolveExecutionEnvironment({ workspaceId: runtime.localInstallation.workspaceId, accountId: runtime.localInstallation.accountId, executionId: ExecutionIdSchema.make("exec_mcp_help"), }); const catalog = environment.catalog as CatalogLike | undefined; if (!catalog) { return defaultExecuteDescription; } const namespaces = yield* catalog.listNamespaces({ limit: 200 }).pipe( Effect.map((items) => items.length > 0 ? items.map((item) => item.displayName ?? item.namespace) : ["none discovered yet"], ), Effect.catchAll(() => Effect.succeed(["none discovered yet"])), ); return buildExecuteWorkflowText(namespaces); }).pipe(Effect.catchAll(() => Effect.succeed(defaultExecuteDescription))), ); const summarizeExecution = (execution: ExecutionEnvelope["execution"]): string => { switch (execution.status) { case "completed": { if (execution.resultJson === null) { return `Execution ${execution.id} completed.`; } return `Execution ${execution.id} completed.\nResult:\n${formatResultPreview(execution.resultJson)}`; } case "failed": return execution.errorText ? `Execution ${execution.id} failed: ${execution.errorText}` : `Execution ${execution.id} failed.`; case "waiting_for_interaction": return `Execution ${execution.id} is waiting for interaction.`; default: return `Execution ${execution.id} is ${execution.status}.`; }}; const executionStructuredContent = (envelope: ExecutionEnvelope): Record<string, unknown> => ({ executionId: envelope.execution.id, status: envelope.execution.status, result: parseJsonValue(envelope.execution.resultJson), errorText: envelope.execution.errorText, logs: parseJsonValue(envelope.execution.logsJson),}); const buildFinalResult = ( envelope: ExecutionEnvelope, options: { isError?: boolean } = {},): ExecutorMcpToolResult => ({ content: [{ type: "text", text: summarizeExecution(envelope.execution) }], structuredContent: executionStructuredContent(envelope), ...(options.isError ? { isError: true } : {}),}); const buildPausedResult = (envelope: ExecutionEnvelope): ExecutorMcpToolResult => { const interaction = envelope.pendingInteraction; const parsed = interaction ? parseInteractionPayload(interaction) : null; return { content: [{ type: "text", text: interaction ? `Execution ${envelope.execution.id} paused: ${parsed?.message ?? "Interaction required."}` : `Execution ${envelope.execution.id} is waiting for interaction.`, }], structuredContent: { executionId: envelope.execution.id, status: "waiting_for_interaction", interaction: interaction ? { id: interaction.id, purpose: interaction.purpose, kind: interaction.kind, message: parsed?.message ?? "Interaction required", mode: parsed?.mode ?? (interaction.kind === "url" ? "url" : "form"), url: parsed?.url ?? null, requestedSchema: parsed?.requestedSchema ?? null, } : null, resumePayload: { executionId: envelope.execution.id, } satisfies ResumePayload, }, };}; const buildToolResult = (envelope: ExecutionEnvelope): ExecutorMcpToolResult => { switch (envelope.execution.status) { case "completed": return buildFinalResult(envelope); case "failed": case "cancelled": return buildFinalResult(envelope, { isError: true }); case "waiting_for_interaction": return buildPausedResult(envelope); default: return buildFinalResult(envelope); }}; const waitForInteractionProgress = async (input: { runtime: SqlControlPlaneRuntime; workspaceId: string; executionId: string; pendingInteractionId: string;}): Promise<ExecutionEnvelope> => { while (true) { const next = await runControlPlane( input.runtime, getExecution({ workspaceId: input.workspaceId as never, executionId: input.executionId as never, }), ); if ( next.execution.status !== "waiting_for_interaction" || next.pendingInteraction?.id !== input.pendingInteractionId ) { return next; } await sleep(pollingIntervalMs); }}; const driveExecutionWithElicitation = async (input: { runtime: SqlControlPlaneRuntime; workspaceId: string; accountId: string; server: McpServer; envelope: ExecutionEnvelope;}): Promise<ExecutionEnvelope> => { let current = input.envelope; while (current.execution.status === "waiting_for_interaction") { const pending = current.pendingInteraction; if (pending === null) { return current; } const parsed = parseInteractionPayload(pending); if (!parsed) { return current; } if (parsed.mode === "form") { const response = await input.server.server.elicitInput({ mode: "form", message: parsed.message, requestedSchema: (parsed.requestedSchema ?? { type: "object", properties: {}, }) as never, }); current = await runControlPlane( input.runtime, resumeExecution({ workspaceId: input.workspaceId as never, executionId: current.execution.id as never, payload: { responseJson: JSON.stringify(response), }, resumedByAccountId: input.accountId as never, }), ); continue; } const response = await input.server.server.elicitInput({ mode: "url", message: parsed.message, url: parsed.url ?? "", elicitationId: parsed.elicitationId ?? pending.id, }); if (response.action !== "accept") { current = await runControlPlane( input.runtime, resumeExecution({ workspaceId: input.workspaceId as never, executionId: current.execution.id as never, payload: { responseJson: JSON.stringify(response), }, resumedByAccountId: input.accountId as never, }), ); continue; } current = await waitForInteractionProgress({ runtime: input.runtime, workspaceId: input.workspaceId, executionId: current.execution.id, pendingInteractionId: pending.id, }); } return current;}; const driveExecutionWithoutElicitation = async (input: { runtime: SqlControlPlaneRuntime; workspaceId: string; accountId: string; executionId: string; initialResponse?: ResumeResponseInput;}): Promise<ExecutionEnvelope> => { let current = await runControlPlane( input.runtime, getExecution({ workspaceId: input.workspaceId as never, executionId: input.executionId as never, }), ); let response = input.initialResponse; while (current.execution.status === "waiting_for_interaction") { const pending = current.pendingInteraction; if (pending === null || response === undefined) { return current; } current = await runControlPlane( input.runtime, resumeExecution({ workspaceId: input.workspaceId as never, executionId: current.execution.id as never, payload: { responseJson: JSON.stringify(response), }, resumedByAccountId: input.accountId as never, }), ); response = undefined; } return current;}; const createExecutorMcpServer = async (config: { runtime: SqlControlPlaneRuntime;}): Promise<McpServer> => { const executeDescription = await loadExecuteDescription(config.runtime); const server = new McpServer( { name: "executor", version: "1.0.0" }, { capabilities: { tools: {}, }, }, ); const workspaceId = config.runtime.localInstallation.workspaceId; const accountId = config.runtime.localInstallation.accountId; const executeTool = server.registerTool( "execute", { description: executeDescription, inputSchema: executeInputSchema, }, async ({ code }: { code: string }) => { let created = await runControlPlane( config.runtime, createExecution({ workspaceId, payload: { code }, createdByAccountId: accountId, }), ); if (supportsManagedElicitation(server)) { created = await driveExecutionWithElicitation({ runtime: config.runtime, workspaceId, accountId, server, envelope: created, }); } return buildToolResult(created); }, ); const resumeTool = server.registerTool( "resume", { description: [ "Resume a paused executor execution using the resumePayload returned by execute.", "Never call this without getting approval from the user first unless they explicitly state otherwise.",Lines 1–500 of 655. Load more
import type { ExecutionEnvelope, ExecutionInteraction, SqlControlPlaneRuntime, } from "@executor/control-plane"; import { EXECUTOR_SOURCES_ADD_HELP_LINES, ExecutionIdSchema, RuntimeExecutionResolverService, createExecution, getExecution, resumeExecution, } from "@executor/control-plane"; import * as Effect from "effect/Effect"; import * as Cause from "effect/Cause"; import * as Exit from "effect/Exit"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; import { z } from "zod/v4"; const pollingIntervalMs = 200; const executeInputSchema = { code: z.string().trim().min(1), }; const resumeInputSchema = { resumePayload: z.object({ executionId: z.string().trim().min(1), }), response: z.object({ action: z.enum(["accept", "decline", "cancel"]), content: z.record(z.string(), z.unknown()).optional(), }).optional(), }; type ParsedInteractionPayload = { mode: "form" | "url"; message: string; url?: string; requestedSchema?: Record<string, unknown>; elicitationId?: string; }; type ResumePayload = { executionId: string; }; type ResumeResponseInput = { action: "accept" | "decline" | "cancel"; content?: Record<string, unknown>; }; type ExecutorMcpToolResult = { content: Array<{ type: "text"; text: string }>; structuredContent: Record<string, unknown>; isError?: boolean; }; export type ExecutorMcpRequestHandler = { handleRequest: (request: Request) => Promise<Response>; close: () => Promise<void>; }; const isRecord = (value: unknown): value is Record<string, unknown> => typeof value === "object" && value !== null && !Array.isArray(value); const parseJsonValue = (value: string | null): unknown => { if (value === null) { return null; } try { return JSON.parse(value); } catch { return value; } }; const maxResultPreviewChars = 30_000; const truncateText = (value: string, maxLength: number): string => value.length > maxLength ? `${value.slice(0, maxLength)}\n... [result preview truncated ${value.length - maxLength} chars]` : value; const formatResultPreview = (resultJson: string): string => { try { const parsed = JSON.parse(resultJson) as unknown; const serialized = typeof parsed === "string" ? parsed : JSON.stringify(parsed, null, 2) ?? String(parsed); return truncateText(serialized, maxResultPreviewChars); } catch { return truncateText(resultJson, maxResultPreviewChars); } }; const parseInteractionPayload = ( interaction: ExecutionInteraction, ): ParsedInteractionPayload | null => { try { const parsed = JSON.parse(interaction.payloadJson) as { elicitation?: { mode?: "form" | "url"; message?: string; url?: string; requestedSchema?: Record<string, unknown>; elicitationId?: string; id?: string; }; }; if (!parsed.elicitation || typeof parsed.elicitation.message !== "string") { return null; } return { mode: parsed.elicitation.mode === "url" ? "url" : "form", message: parsed.elicitation.message, url: parsed.elicitation.url, requestedSchema: isRecord(parsed.elicitation.requestedSchema) ? parsed.elicitation.requestedSchema : undefined, elicitationId: typeof parsed.elicitation.elicitationId === "string" ? parsed.elicitation.elicitationId : typeof parsed.elicitation.id === "string" ? parsed.elicitation.id : undefined, }; } catch { return null; } }; const runControlPlane = async <A, E, R>( runtime: SqlControlPlaneRuntime, effect: Effect.Effect<A, E, R>, ): Promise<A> => { const exit = await Effect.runPromiseExit( effect.pipe(Effect.provide(runtime.runtimeLayer)) as Effect.Effect<A, E, never>, ); if (Exit.isSuccess(exit)) { return exit.value; } const error = Cause.squash(exit.cause); if (error instanceof Error) { // Preserve the original error with its stack trace and message rather // than wrapping it in an opaque FiberFailure. throw error; } throw error; }; const sleep = (ms: number) => new Promise<void>((resolve) => { setTimeout(resolve, ms); }); const supportsManagedElicitation = (server: McpServer): boolean => { const capabilities = server.server.getClientCapabilities(); return Boolean(capabilities?.elicitation?.form) && Boolean(capabilities?.elicitation?.url); }; type CatalogLike = { listNamespaces: (input: { limit: number }) => Effect.Effect< ReadonlyArray<{ namespace: string; displayName?: string }>, unknown >; }; const buildExecuteWorkflowText = (namespaces: readonly string[] = []): string => [ "Execute TypeScript in sandbox; call tools via discovery workflow.", ...(namespaces.length > 0 ? [ "Available namespaces:", ...namespaces.map((namespace) => `- ${namespace}`), ] : []), "Workflow:", '1) const matches = await tools.discover({ query: "<intent>", limit: 12 });', "2) const details = await tools.describe.tool({ path, includeSchemas: true });", "3) Call selected tools.<path>(input).", '4) To connect a source, call tools.executor.sources.add(...) for MCP, OpenAPI, or GraphQL APIs.', ...EXECUTOR_SOURCES_ADD_HELP_LINES, "5) If execution pauses for interaction, resume it with the returned resumePayload or the available resume flow.", "Do not use fetch; use tools.* only.", ].join("\n"); const defaultExecuteDescription = buildExecuteWorkflowText(); const loadExecuteDescription = (runtime: SqlControlPlaneRuntime): Promise<string> => runControlPlane( runtime, Effect.gen(function* () { const resolveExecutionEnvironment = yield* RuntimeExecutionResolverService; const environment = yield* resolveExecutionEnvironment({ workspaceId: runtime.localInstallation.workspaceId, accountId: runtime.localInstallation.accountId, executionId: ExecutionIdSchema.make("exec_mcp_help"), }); const catalog = environment.catalog as CatalogLike | undefined; if (!catalog) { return defaultExecuteDescription; } const namespaces = yield* catalog.listNamespaces({ limit: 200 }).pipe( Effect.map((items) => items.length > 0 ? items.map((item) => item.displayName ?? item.namespace) : ["none discovered yet"], ), Effect.catchAll(() => Effect.succeed(["none discovered yet"])), ); return buildExecuteWorkflowText(namespaces); }).pipe(Effect.catchAll(() => Effect.succeed(defaultExecuteDescription))), ); const summarizeExecution = (execution: ExecutionEnvelope["execution"]): string => { switch (execution.status) { case "completed": { if (execution.resultJson === null) { return `Execution ${execution.id} completed.`; } return `Execution ${execution.id} completed.\nResult:\n${formatResultPreview(execution.resultJson)}`; } case "failed": return execution.errorText ? `Execution ${execution.id} failed: ${execution.errorText}` : `Execution ${execution.id} failed.`; case "waiting_for_interaction": return `Execution ${execution.id} is waiting for interaction.`; default: return `Execution ${execution.id} is ${execution.status}.`; } }; const executionStructuredContent = (envelope: ExecutionEnvelope): Record<string, unknown> => ({ executionId: envelope.execution.id, status: envelope.execution.status, result: parseJsonValue(envelope.execution.resultJson), errorText: envelope.execution.errorText, logs: parseJsonValue(envelope.execution.logsJson), }); const buildFinalResult = ( envelope: ExecutionEnvelope, options: { isError?: boolean } = {}, ): ExecutorMcpToolResult => ({ content: [{ type: "text", text: summarizeExecution(envelope.execution) }], structuredContent: executionStructuredContent(envelope), ...(options.isError ? { isError: true } : {}), }); const buildPausedResult = (envelope: ExecutionEnvelope): ExecutorMcpToolResult => { const interaction = envelope.pendingInteraction; const parsed = interaction ? parseInteractionPayload(interaction) : null; return { content: [{ type: "text", text: interaction ? `Execution ${envelope.execution.id} paused: ${parsed?.message ?? "Interaction required."}` : `Execution ${envelope.execution.id} is waiting for interaction.`, }], structuredContent: { executionId: envelope.execution.id, status: "waiting_for_interaction", interaction: interaction ? { id: interaction.id, purpose: interaction.purpose, kind: interaction.kind, message: parsed?.message ?? "Interaction required", mode: parsed?.mode ?? (interaction.kind === "url" ? "url" : "form"), url: parsed?.url ?? null, requestedSchema: parsed?.requestedSchema ?? null, } : null, resumePayload: { executionId: envelope.execution.id, } satisfies ResumePayload, }, }; }; const buildToolResult = (envelope: ExecutionEnvelope): ExecutorMcpToolResult => { switch (envelope.execution.status) { case "completed": return buildFinalResult(envelope); case "failed": case "cancelled": return buildFinalResult(envelope, { isError: true }); case "waiting_for_interaction": return buildPausedResult(envelope); default: return buildFinalResult(envelope); } }; const waitForInteractionProgress = async (input: { runtime: SqlControlPlaneRuntime; workspaceId: string; executionId: string; pendingInteractionId: string; }): Promise<ExecutionEnvelope> => { while (true) { const next = await runControlPlane( input.runtime, getExecution({ workspaceId: input.workspaceId as never, executionId: input.executionId as never, }), ); if ( next.execution.status !== "waiting_for_interaction" || next.pendingInteraction?.id !== input.pendingInteractionId ) { return next; } await sleep(pollingIntervalMs); } }; const driveExecutionWithElicitation = async (input: { runtime: SqlControlPlaneRuntime; workspaceId: string; accountId: string; server: McpServer; envelope: ExecutionEnvelope; }): Promise<ExecutionEnvelope> => { let current = input.envelope; while (current.execution.status === "waiting_for_interaction") { const pending = current.pendingInteraction; if (pending === null) { return current; } const parsed = parseInteractionPayload(pending); if (!parsed) { return current; } if (parsed.mode === "form") { const response = await input.server.server.elicitInput({ mode: "form", message: parsed.message, requestedSchema: (parsed.requestedSchema ?? { type: "object", properties: {}, }) as never, }); current = await runControlPlane( input.runtime, resumeExecution({ workspaceId: input.workspaceId as never, executionId: current.execution.id as never, payload: { responseJson: JSON.stringify(response), }, resumedByAccountId: input.accountId as never, }), ); continue; } const response = await input.server.server.elicitInput({ mode: "url", message: parsed.message, url: parsed.url ?? "", elicitationId: parsed.elicitationId ?? pending.id, }); if (response.action !== "accept") { current = await runControlPlane( input.runtime, resumeExecution({ workspaceId: input.workspaceId as never, executionId: current.execution.id as never, payload: { responseJson: JSON.stringify(response), }, resumedByAccountId: input.accountId as never, }), ); continue; } current = await waitForInteractionProgress({ runtime: input.runtime, workspaceId: input.workspaceId, executionId: current.execution.id, pendingInteractionId: pending.id, }); } return current; }; const driveExecutionWithoutElicitation = async (input: { runtime: SqlControlPlaneRuntime; workspaceId: string; accountId: string; executionId: string; initialResponse?: ResumeResponseInput; }): Promise<ExecutionEnvelope> => { let current = await runControlPlane( input.runtime, getExecution({ workspaceId: input.workspaceId as never, executionId: input.executionId as never, }), ); let response = input.initialResponse; while (current.execution.status === "waiting_for_interaction") { const pending = current.pendingInteraction; if (pending === null || response === undefined) { return current; } current = await runControlPlane( input.runtime, resumeExecution({ workspaceId: input.workspaceId as never, executionId: current.execution.id as never, payload: { responseJson: JSON.stringify(response), }, resumedByAccountId: input.accountId as never, }), ); response = undefined; } return current; }; const createExecutorMcpServer = async (config: { runtime: SqlControlPlaneRuntime; }): Promise<McpServer> => { const executeDescription = await loadExecuteDescription(config.runtime); const server = new McpServer( { name: "executor", version: "1.0.0" }, { capabilities: { tools: {}, }, }, ); const workspaceId = config.runtime.localInstallation.workspaceId; const accountId = config.runtime.localInstallation.accountId; const executeTool = server.registerTool( "execute", { description: executeDescription, inputSchema: executeInputSchema, }, async ({ code }: { code: string }) => { let created = await runControlPlane( config.runtime, createExecution({ workspaceId, payload: { code }, createdByAccountId: accountId, }), ); if (supportsManagedElicitation(server)) { created = await driveExecutionWithElicitation({ runtime: config.runtime, workspaceId, accountId, server, envelope: created, }); } return buildToolResult(created); }, ); const resumeTool = server.registerTool( "resume", { description: [ "Resume a paused executor execution using the resumePayload returned by execute.", "Never call this without getting approval from the user first unless they explicitly state otherwise.",