History

Refine paused executor MCP output

Rhys Sullivan committed 8 months ago

Commit eae59191 parent bc100714 files · +207 −64Browse files

Unified · Split

4 files
  1. apps/executor/src/cli/interaction-handling.test.ts +1 −1
  2. packages/executor-mcp/src/index.ts +14 −63
  3. packages/executor-mcp/src/paused-result.test.ts +97 −0
  4. packages/executor-mcp/src/paused-result.ts +95 −0
apps/executor/src/cli/interaction-handling.test.ts+1−1
@@ -3,7 +3,7 @@ import { describe, expect, it } from "@effect/vitest"; · 2 unmodified lines
3import { decideInteractionHandling } from "./interaction-handling";3import { decideInteractionHandling } from "./interaction-handling";
4 4
5describe("interaction-handling", () => {5describe("interaction-handling", () => {
6 it("keeps URL interactions distinct when the terminal is non-interactive", () => {6 it("keeps URL interactions distinct even when the terminal is non-interactive", () => {
7 expect(decideInteractionHandling({7 expect(decideInteractionHandling({
8 parsed: {8 parsed: {
9 mode: "url",9 mode: "url",
packages/executor-mcp/src/index.ts+14−63
@@ -1,7 +1,6 @@
1import type {1import type {
2 ExecutionEnvelope,2 ExecutionEnvelope,
3 ExecutionInteraction,3 ControlPlaneRuntime,
4 SqlControlPlaneRuntime,
5} from "@executor/control-plane";4} from "@executor/control-plane";
6import {5import {
7 EXECUTOR_SOURCES_ADD_HELP_LINES,6 EXECUTOR_SOURCES_ADD_HELP_LINES,
@@ -18,6 +17,11 @@ import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; · 10 unmodified lines
18import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js";17import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js";
19import { z } from "zod/v4";18import { z } from "zod/v4";
20 19
20import {
21 buildPausedResultText,
22 parseInteractionPayload,
23} from "./paused-result";
24
21const pollingIntervalMs = 200;25const pollingIntervalMs = 200;
22 26
23const executeInputSchema = {27const executeInputSchema = {
@@ -34,14 +38,6 @@ const resumeInputSchema = { · 10 unmodified lines
34 }).optional(),38 }).optional(),
35};39};
36 40
37type ParsedInteractionPayload = {
38 mode: "form" | "url";
39 message: string;
40 url?: string;
41 requestedSchema?: Record<string, unknown>;
42 elicitationId?: string;
43};
44
45type ResumePayload = {41type ResumePayload = {
46 executionId: string;42 executionId: string;
47};43};
@@ -62,9 +58,6 @@ export type ExecutorMcpRequestHandler = { · 14 unmodified lines
62 close: () => Promise<void>;58 close: () => Promise<void>;
63};59};
64 60
65const isRecord = (value: unknown): value is Record<string, unknown> =>
66 typeof value === "object" && value !== null && !Array.isArray(value);
67
68const parseJsonValue = (value: string | null): unknown => {61const parseJsonValue = (value: string | null): unknown => {
69 if (value === null) {62 if (value === null) {
70 return null;63 return null;
@@ -93,48 +86,8 @@ const formatResultPreview = (resultJson: string): string => { · 22 unmodified lines
93 return truncateText(resultJson, maxResultPreviewChars);86 return truncateText(resultJson, maxResultPreviewChars);
94 }87 }
95};88};
96
97
98const parseInteractionPayload = (
99 interaction: ExecutionInteraction,
100): ParsedInteractionPayload | null => {
101 try {
102 const parsed = JSON.parse(interaction.payloadJson) as {
103 elicitation?: {
104 mode?: "form" | "url";
105 message?: string;
106 url?: string;
107 requestedSchema?: Record<string, unknown>;
108 elicitationId?: string;
109 id?: string;
110 };
111 };
112
113 if (!parsed.elicitation || typeof parsed.elicitation.message !== "string") {
114 return null;
115 }
116
117 return {
118 mode: parsed.elicitation.mode === "url" ? "url" : "form",
119 message: parsed.elicitation.message,
120 url: parsed.elicitation.url,
121 requestedSchema: isRecord(parsed.elicitation.requestedSchema)
122 ? parsed.elicitation.requestedSchema
123 : undefined,
124 elicitationId:
125 typeof parsed.elicitation.elicitationId === "string"
126 ? parsed.elicitation.elicitationId
127 : typeof parsed.elicitation.id === "string"
128 ? parsed.elicitation.id
129 : undefined,
130 };
131 } catch {
132 return null;
133 }
134};
135
136const runControlPlane = async <A, E, R>(89const runControlPlane = async <A, E, R>(
137 runtime: SqlControlPlaneRuntime,90 runtime: ControlPlaneRuntime,
138 effect: Effect.Effect<A, E, R>,91 effect: Effect.Effect<A, E, R>,
139): Promise<A> => {92): Promise<A> => {
140 const exit = await Effect.runPromiseExit(93 const exit = await Effect.runPromiseExit(
@@ -192,7 +145,7 @@ const buildExecuteWorkflowText = (namespaces: readonly string[] = []): string => · 51 unmodified lines
192 145
193const defaultExecuteDescription = buildExecuteWorkflowText();146const defaultExecuteDescription = buildExecuteWorkflowText();
194 147
195const loadExecuteDescription = (runtime: SqlControlPlaneRuntime): Promise<string> =>148const loadExecuteDescription = (runtime: ControlPlaneRuntime): Promise<string> =>
196 runControlPlane(149 runControlPlane(
197 runtime,150 runtime,
198 Effect.gen(function* () {151 Effect.gen(function* () {
@@ -265,9 +218,7 @@ const buildPausedResult = (envelope: ExecutionEnvelope): ExecutorMcpToolResult = · 66 unmodified lines
265 return {218 return {
266 content: [{219 content: [{
267 type: "text",220 type: "text",
268 text: interaction221 text: buildPausedResultText(envelope),
269 ? `Execution ${envelope.execution.id} paused: ${parsed?.message ?? "Interaction required."}`
270 : `Execution ${envelope.execution.id} is waiting for interaction.`,
271 }],222 }],
272 structuredContent: {223 structuredContent: {
273 executionId: envelope.execution.id,224 executionId: envelope.execution.id,
@@ -305,7 +256,7 @@ const buildToolResult = (envelope: ExecutionEnvelope): ExecutorMcpToolResult => · 31 unmodified lines
305};256};
306 257
307const waitForInteractionProgress = async (input: {258const waitForInteractionProgress = async (input: {
308 runtime: SqlControlPlaneRuntime;259 runtime: ControlPlaneRuntime;
309 workspaceId: string;260 workspaceId: string;
310 executionId: string;261 executionId: string;
311 pendingInteractionId: string;262 pendingInteractionId: string;
@@ -331,7 +282,7 @@ const waitForInteractionProgress = async (input: { · 19 unmodified lines
331};282};
332 283
333const driveExecutionWithElicitation = async (input: {284const driveExecutionWithElicitation = async (input: {
334 runtime: SqlControlPlaneRuntime;285 runtime: ControlPlaneRuntime;
335 workspaceId: string;286 workspaceId: string;
336 accountId: string;287 accountId: string;
337 server: McpServer;288 server: McpServer;
@@ -408,7 +359,7 @@ const driveExecutionWithElicitation = async (input: { · 70 unmodified lines
408};359};
409 360
410const driveExecutionWithoutElicitation = async (input: {361const driveExecutionWithoutElicitation = async (input: {
411 runtime: SqlControlPlaneRuntime;362 runtime: ControlPlaneRuntime;
412 workspaceId: string;363 workspaceId: string;
413 accountId: string;364 accountId: string;
414 executionId: string;365 executionId: string;
@@ -447,7 +398,7 @@ const driveExecutionWithoutElicitation = async (input: { · 32 unmodified lines
447};398};
448 399
449const createExecutorMcpServer = async (config: {400const createExecutorMcpServer = async (config: {
450 runtime: SqlControlPlaneRuntime;401 runtime: ControlPlaneRuntime;
451}): Promise<McpServer> => {402}): Promise<McpServer> => {
452 const executeDescription = await loadExecuteDescription(config.runtime);403 const executeDescription = await loadExecuteDescription(config.runtime);
453 const server = new McpServer(404 const server = new McpServer(
@@ -552,7 +503,7 @@ const jsonErrorResponse = (status: number, code: number, message: string) => · 98 unmodified lines
552 });503 });
553 504
554export const createExecutorMcpRequestHandler = (505export const createExecutorMcpRequestHandler = (
555 runtime: SqlControlPlaneRuntime,506 runtime: ControlPlaneRuntime,
556): ExecutorMcpRequestHandler => {507): ExecutorMcpRequestHandler => {
557 const transports = new Map<string, WebStandardStreamableHTTPServerTransport>();508 const transports = new Map<string, WebStandardStreamableHTTPServerTransport>();
558 const servers = new Map<string, McpServer>();509 const servers = new Map<string, McpServer>();
packages/executor-mcp/src/paused-result.test.ts+97−0
@@ -0,0 +1,97 @@
1import { describe, expect, it } from "@effect/vitest";
2import type { ExecutionEnvelope, ExecutionInteraction } from "@executor/control-plane";
3
4import {
5 buildPausedResultText,
6 parseInteractionPayload,
7} from "./paused-result";
8
9const makeInteraction = (patch: Partial<ExecutionInteraction> = {}): ExecutionInteraction => ({
10 id: "exec_123:source_connect_oauth2:call_1" as never,
11 executionId: "exec_123" as never,
12 status: "pending",
13 kind: "url",
14 purpose: "source_connect_oauth2",
15 payloadJson: JSON.stringify({
16 elicitation: {
17 mode: "url",
18 message: "Finish connecting Axiom",
19 url: "https://mcp.axiom.co/authorize",
20 elicitationId: "oauth-session-123",
21 },
22 }),
23 responseJson: null,
24 createdAt: 1,
25 updatedAt: 1,
26 ...patch,
27});
28
29const makeEnvelope = (
30 patch: Partial<ExecutionEnvelope> = {},
31): ExecutionEnvelope => ({
32 execution: {
33 id: "exec_123" as never,
34 workspaceId: "ws_123" as never,
35 createdByAccountId: "acct_123" as never,
36 status: "waiting_for_interaction",
37 code: "return await tools.executor.sources.add({ endpoint: 'https://mcp.axiom.co/mcp' });",
38 resultJson: null,
39 errorText: null,
40 logsJson: null,
41 startedAt: 1,
42 completedAt: null,
43 createdAt: 1,
44 updatedAt: 1,
45 },
46 pendingInteraction: makeInteraction(),
47 ...patch,
48});
49
50describe("paused-result", () => {
51 it("parses the browser URL from a paused interaction payload", () => {
52 const parsed = parseInteractionPayload(makeInteraction());
53
54 expect(parsed).not.toBeNull();
55 expect(parsed?.mode).toBe("url");
56 expect(parsed?.url).toBe("https://mcp.axiom.co/authorize");
57 expect(parsed?.elicitationId).toBe("oauth-session-123");
58 });
59
60 it("includes URL and resume instructions in the paused text for browser auth", () => {
61 const text = buildPausedResultText(makeEnvelope());
62
63 expect(text).toContain("Finish connecting Axiom");
64 expect(text).toContain("Open this URL in a browser:");
65 expect(text).toContain("https://mcp.axiom.co/authorize");
66 expect(text).toContain("executor.resume");
67 expect(text).toContain('"executionId": "exec_123"');
68 });
69
70 it("includes resume guidance for form interactions", () => {
71 const text = buildPausedResultText(
72 makeEnvelope({
73 pendingInteraction: makeInteraction({
74 kind: "form",
75 purpose: "tool_execution_gate",
76 payloadJson: JSON.stringify({
77 elicitation: {
78 mode: "form",
79 message: "Approve the tool call",
80 requestedSchema: {
81 type: "object",
82 properties: {
83 approve: { type: "boolean" },
84 },
85 required: ["approve"],
86 },
87 },
88 }),
89 }),
90 }),
91 );
92
93 expect(text).toContain("Approve the tool call");
94 expect(text).toContain("include a response object matching the requested schema");
95 expect(text).toContain('"executionId": "exec_123"');
96 });
97});
packages/executor-mcp/src/paused-result.ts+95−0
@@ -0,0 +1,95 @@
1import type {
2 ExecutionEnvelope,
3 ExecutionInteraction,
4} from "@executor/control-plane";
5
6export type ParsedInteractionPayload = {
7 mode: "form" | "url";
8 message: string;
9 url?: string;
10 requestedSchema?: Record<string, unknown>;
11 elicitationId?: string;
12};
13
14const isRecord = (value: unknown): value is Record<string, unknown> =>
15 typeof value === "object" && value !== null && !Array.isArray(value);
16
17export const parseInteractionPayload = (
18 interaction: ExecutionInteraction,
19): ParsedInteractionPayload | null => {
20 try {
21 const parsed = JSON.parse(interaction.payloadJson) as {
22 elicitation?: {
23 mode?: "form" | "url";
24 message?: string;
25 url?: string;
26 requestedSchema?: Record<string, unknown>;
27 elicitationId?: string;
28 id?: string;
29 };
30 };
31
32 if (!parsed.elicitation || typeof parsed.elicitation.message !== "string") {
33 return null;
34 }
35
36 return {
37 mode: parsed.elicitation.mode === "url" ? "url" : "form",
38 message: parsed.elicitation.message,
39 url: parsed.elicitation.url,
40 requestedSchema: isRecord(parsed.elicitation.requestedSchema)
41 ? parsed.elicitation.requestedSchema
42 : undefined,
43 elicitationId:
44 typeof parsed.elicitation.elicitationId === "string"
45 ? parsed.elicitation.elicitationId
46 : typeof parsed.elicitation.id === "string"
47 ? parsed.elicitation.id
48 : undefined,
49 };
50 } catch {
51 return null;
52 }
53};
54
55const formatResumePayload = (executionId: string): string =>
56 JSON.stringify({ executionId }, null, 2);
57
58export const buildPausedResultText = (envelope: ExecutionEnvelope): string => {
59 const interaction = envelope.pendingInteraction;
60 const parsed = interaction ? parseInteractionPayload(interaction) : null;
61
62 if (!interaction) {
63 return [
64 `Execution ${envelope.execution.id} is waiting for interaction.`,
65 "Resume with executor.resume using this resumePayload:",
66 formatResumePayload(envelope.execution.id),
67 ].join("\n");
68 }
69
70 const lines = [
71 `Execution ${envelope.execution.id} paused: ${parsed?.message ?? "Interaction required."}`,
72 ];
73
74 if (parsed?.mode === "url" && typeof parsed.url === "string" && parsed.url.trim().length > 0) {
75 lines.push("Open this URL in a browser:");
76 lines.push(parsed.url.trim());
77 lines.push(
78 "After finishing the browser flow, call executor.resume with this resumePayload if your client does not refresh automatically:",
79 );
80 lines.push(formatResumePayload(envelope.execution.id));
81 return lines.join("\n");
82 }
83
84 if (parsed?.mode === "form") {
85 lines.push(
86 "Resume with executor.resume using this resumePayload and include a response object matching the requested schema:",
87 );
88 lines.push(formatResumePayload(envelope.execution.id));
89 return lines.join("\n");
90 }
91
92 lines.push("Resume with executor.resume using this resumePayload:");
93 lines.push(formatResumePayload(envelope.execution.id));
94 return lines.join("\n");
95};