self-host: browser approval of gated MCP actions

Self-host (and the local app) could pause a gated MCP execution but had no way for a human to approve it in the browser: the in-process MCP store served model-mode only, and the shared resume page fetched paused detail from the session-less /api/executions/:id — a different engine that never holds the MCP pause. - @executor-js/host-mcp: an in-process browser-approval store (the resume long-poll <-> HTTP decision bridge, keyed by executionId), and the in-memory session store now builds browser-mode servers, keeps each session's engine, and serves the session-scoped paused (GET) + resume (POST) endpoints. makeMcpBuildServer returns { mcpServer, engine }. - host-selfhost: mounts the approval endpoints at /api/mcp-sessions/*, gated by a Better Auth session. - react: the shared resume page fetches paused detail session-scoped, so the in-process hosts resolve it from the right engine (mirrors cloud). - apps/local: migrated onto the shared approval store + the new GET handler, dropping its duplicated waiter maps; its browser approval page now works, not just the API-direct path. - e2e: browser-approval scenarios moved to scenarios/ (cross-target) — approve + decline now green on self-host too.

Rhys Sullivan committed 5 months ago

Commit e3c90bb1 parent 7fb922011 files · +380 −87Browse files

Unified · Split

11 files

apps/host-selfhost/src/app.ts +4 −0
apps/host-selfhost/src/mcp/index.ts +41 −1
apps/local/src/mcp.ts +39 −59
apps/local/src/serve.test.ts +5 −0
apps/local/src/serve.ts +5 −1
e2e/scenarios/browser-approval.test.ts +4 −3
packages/core/api/src/server/mcp-build.ts +7 −2
packages/hosts/mcp/package.json +4 −0
packages/hosts/mcp/src/browser-approval-store.ts +71 −0
packages/hosts/mcp/src/in-memory-session-store.ts +155 −19
packages/react/src/routes/resume.$executionId.tsx +45 −2

apps/host-selfhost/src/app.ts+4−0

@@ -85,6 +85,10 @@ export const makeSelfHostApp = async (options: MakeSelfHostAppOptions = {}) => { · 84 unmodified lines
85	85	routes: [
86	86	// Better Auth owns /api/auth/* — the full path reaches it unmodified.
87	87	HttpRouter.add("", "/api/auth/", HttpEffect.fromWebHandler(authHandler)),
	88	+ // Browser approval of paused MCP executions: the console resume page
	89	+ // reads paused detail (GET) and records the decision (POST .../resume),
	90	+ // session-cookie-gated, delegating to the in-process MCP store.
	91	+ HttpRouter.add("", "/api/mcp-sessions/", HttpEffect.fromWebHandler(mcp.approvalHandler)),
88	92	// App-local admin (invite-code) API, served under /api/admin/*.
89	93	makeSelfHostAdminApiLayer({ betterAuth, db: dbHandle, mountPrefix: "/api" }),
90	94	// Public system API: /api/health + /api/setup-status (unauthenticated).

apps/host-selfhost/src/mcp/index.ts+41−1

@@ -1,4 +1,4 @@
1		−import { Layer } from "effect";
	1	+import { Effect, Layer } from "effect";
2	2
3	3	import { IdentityProvider } from "@executor-js/api/server";
4	4	import type { McpAuthProvider, McpErrorReporter, McpSessionStore } from "@executor-js/host-mcp";
@@ -50,10 +50,49 @@ export interface SelfHostMcpSeams { · 45 unmodified lines
50	50	readonly sessions: Layer.Layer<McpSessionStore>;
51	51	/** Route 500 defects through the host's console `ErrorCapture`. */
52	52	readonly reporter: Layer.Layer<McpErrorReporter>;
	53	+ /**
	54	+ * The browser-approval HTTP handler, mounted by the app at
	55	+ * `/api/mcp-sessions/*`: a session-cookie-gated web handler that serves the
	56	+ * paused-execution detail (GET) and records the human's decision (POST
	57	+ * `/resume`) for the console approval page. Browser elicitation mode only.
	58	+ */
	59	+ readonly approvalHandler: (request: Request) => Promise<Response>;
53	60	/** Dispose all live in-process MCP sessions at shutdown (not a seam). */
54	61	readonly close: () => Promise<void>;
55	62	}
56	63
	64	+const jsonResponse = (value: unknown, status: number): Response =>
	65	+ new Response(JSON.stringify(value), { status, headers: { "content-type": "application/json" } });
	66	+
	67	+/**
	68	+ * Gate the browser-approval endpoints behind a valid Better Auth session (the
	69	+ * console page calls them with the user's cookie), then delegate to the
	70	+ * in-process store's paused/resume handlers. Single-tenant: any authenticated
	71	+ * user of the one org may act on a session it still holds — the store confirms
	72	+ * the execution belongs to the addressed session before recording.
	73	+ */
	74	+const makeApprovalHandler =
	75	+ (
	76	+ store: ReturnType<typeof makeSelfHostMcpSessionStore>,
	77	+ betterAuth: BetterAuthHandle,
	78	+ ): ((request: Request) => Promise<Response>) =>
	79	+ async (request) => {
	80	+ // A malformed cookie must read as unauthenticated, not 500.
	81	+ const session = await Effect.runPromise(
	82	+ Effect.tryPromise({
	83	+ try: () => betterAuth.auth.api.getSession({ headers: request.headers }),
	84	+ catch: () => "session lookup failed",
	85	+ }).pipe(Effect.orElseSucceed(() => null)),
	86	+ );
	87	+ if (!session) return jsonResponse({ error: "Unauthorized" }, 401);
	88	+
	89	+ return (
	90	+ (await store.handlePausedRequest(request)) ??
	91	+ (await store.handleApprovalRequest(request)) ??
	92	+ jsonResponse({ error: "Not found" }, 404)
	93	+ );
	94	+ };
	95	+
57	96	/**
58	97	* Build the self-host MCP serving seams over the long-lived DB handle. The auth
59	98	* seam is `selfHostMcpAuth` (Better Auth mcp() OAuth), with the Better Auth
@@ -73,6 +112,7 @@ export const makeSelfHostMcpSeams = ( · 13 unmodified lines
73	112	auth,
74	113	sessions: selfHostMcpSessions(sessionStore),
75	114	reporter: selfHostMcpReporter,
	115	+ approvalHandler: makeApprovalHandler(sessionStore, betterAuth),
76	116	close: sessionStore.close,
77	117	};
78	118	};

apps/local/src/mcp.ts+39−59

@@ -1,4 +1,4 @@
1		−import { Deferred, Effect } from "effect";
	1	+import { Effect } from "effect";
2	2	import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
3	3	import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
4	4	import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js";
@@ -14,7 +14,8 @@ import { · 9 unmodified lines
14	14	formatResumeAcknowledgement,
15	15	readElicitationMode,
16	16	} from "@executor-js/host-mcp/browser-approval";
17		−import type { ResumeResponse } from "@executor-js/execution";
	17	+import { makeInProcessBrowserApprovalStore } from "@executor-js/host-mcp/browser-approval-store";
	18	+import { formatPausedExecution, type ResumeResponse } from "@executor-js/execution";
18	19
19	20	import { startIntegrationsRefresh } from "./integrations";
20	21
@@ -24,6 +25,9 @@ import { startIntegrationsRefresh } from "./integrations"; · 3 unmodified lines
24	25
25	26	export type McpRequestHandler = {
26	27	readonly handleRequest: (request: Request) => Promise<Response>;
	28	+ /** GET `/api/mcp-sessions/:id/executions/:id` — paused detail for the console. */
	29	+ readonly handlePausedRequest: (request: Request) => Promise<Response>;
	30	+ /** POST `/api/mcp-sessions/:id/executions/:id/resume` — record the decision. */
27	31	readonly handleApprovalRequest: (request: Request) => Promise<Response>;
28	32	readonly close: () => Promise<void>;
29	33	};
@@ -52,6 +56,7 @@ const ignoreClose = (close: (() => Promise<void>) \| undefined): Promise<void> => · 22 unmodified lines
52	56	)
53	57	: Promise.resolve();
54	58
	59	+const pausedRequestPattern = /^\/api\/mcp-sessions\/([^/?#]+)\/executions\/([^/?#]+)$/;
55	60	const approvalRequestPattern = /^\/api\/mcp-sessions\/([^/?#]+)\/executions\/([^/?#]+)\/resume$/;
56	61
57	62	const json = (value: unknown, status = 200): Response =>
@@ -77,16 +82,29 @@ const resumeApprovalResult = (executionId: string, response: ResumeResponse) => · 19 unmodified lines
77	82	export const createMcpRequestHandler = (config: ExecutorMcpServerConfig): McpRequestHandler => {
78	83	const transports = new Map<string, WebStandardStreamableHTTPServerTransport>();
79	84	const servers = new Map<string, McpServer>();
80		− const approvalResponses = new Map<string, Map<string, ResumeResponse>>();
81		− const approvalWaiters = new Map<string, Map<string, Deferred.Deferred<ResumeResponse>>>();
	85	+ const approvals = makeInProcessBrowserApprovalStore();
	86	+ // Local runs one shared engine across every MCP session (main.ts builds it and
	87	+ // passes it in), so the paused-execution lookup for browser approval reads it
	88	+ // directly — there is no per-session engine to track.
	89	+ const engine = "engine" in config ? config.engine : null;
	90	+
	91	+ const pausedDetail = (
	92	+ executionId: string,
	93	+ ): Promise<ReturnType<typeof formatPausedExecution> \| null> =>
	94	+ engine
	95	+ ? Effect.runPromise(
	96	+ engine.getPausedExecution(executionId).pipe(
	97	+ Effect.map((paused) => (paused ? formatPausedExecution(paused) : null)),
	98	+ Effect.orElseSucceed(() => null),
	99	+ ),
	100	+ )
	101	+ : Promise.resolve(null);
82	102
83	103	const dispose = async (id: string, opts: { transport?: boolean; server?: boolean } = {}) => {
84	104	const t = transports.get(id);
85	105	const s = servers.get(id);
86	106	transports.delete(id);
87	107	servers.delete(id);
88		− approvalResponses.delete(id);
89		− approvalWaiters.delete(id);
90	108	if (opts.transport) await ignoreClose(t ? () => t.close() : undefined);
91	109	if (opts.server) await ignoreClose(s ? () => s.close() : undefined);
92	110	};
@@ -125,48 +143,7 @@ export const createMcpRequestHandler = (config: ExecutorMcpServerConfig): McpReq · 32 unmodified lines
125	143	created = await Effect.runPromise(
126	144	createExecutorMcpServer({
127	145	...config,
128		− browserApprovalStore: {
129		− takeResponse: (executionId) =>
130		− Effect.sync(() => {
131		− if (!createdSessionId) return null;
132		− const sessionApprovals = approvalResponses.get(createdSessionId);
133		− const response = sessionApprovals?.get(executionId) ?? null;
134		− sessionApprovals?.delete(executionId);
135		− return response;
136		− }),
137		− waitForResponse: (executionId) =>
138		− Effect.gen(function* () {
139		− if (!createdSessionId) return null;
140		− const sessionApprovals = approvalResponses.get(createdSessionId);
141		− const response = sessionApprovals?.get(executionId) ?? null;
142		− if (response) {
143		− sessionApprovals?.delete(executionId);
144		− return response;
145		− }
146		−
147		− const sessionWaiters =
148		− approvalWaiters.get(createdSessionId) ??
149		− new Map<string, Deferred.Deferred<ResumeResponse>>();
150		− const waiter =
151		− sessionWaiters.get(executionId) ?? (yield* Deferred.make<ResumeResponse>());
152		− sessionWaiters.set(executionId, waiter);
153		− approvalWaiters.set(createdSessionId, sessionWaiters);
154		−
155		− yield* Deferred.await(waiter).pipe(
156		− Effect.ensuring(
157		− Effect.sync(() => {
158		− if (sessionWaiters.get(executionId) === waiter) {
159		− sessionWaiters.delete(executionId);
160		− }
161		− }),
162		− ),
163		− );
164		− const approvals = approvalResponses.get(createdSessionId);
165		− const approved = approvals?.get(executionId) ?? null;
166		− approvals?.delete(executionId);
167		− return approved;
168		− }),
169		− },
	146	+ browserApprovalStore: approvals.store,
170	147	elicitationMode:
171	148	elicitationMode === "browser"
172	149	? {
@@ -197,26 +174,29 @@ export const createMcpRequestHandler = (config: ExecutorMcpServerConfig): McpReq · 24 unmodified lines
197	174	}
198	175	},
199	176
	177	+ handlePausedRequest: async (request) => {
	178	+ const match = pausedRequestPattern.exec(new URL(request.url).pathname);
	179	+ if (!match) return json({ error: "Not found" }, 404);
	180	+ if (request.method !== "GET") return json({ error: "Method not allowed" }, 405);
	181	+
	182	+ const paused = await pausedDetail(decodeURIComponent(match[2]));
	183	+ if (!paused) return json({ error: "Paused execution not found" }, 404);
	184	+ return json({ text: paused.text, structured: paused.structured });
	185	+ },
	186	+
200	187	handleApprovalRequest: async (request) => {
201		− const url = new URL(request.url);
202		− const match = approvalRequestPattern.exec(url.pathname);
	188	+ const match = approvalRequestPattern.exec(new URL(request.url).pathname);
203	189	if (!match) return json({ error: "Not found" }, 404);
204	190	if (request.method !== "POST") return json({ error: "Method not allowed" }, 405);
205	191
206		− const sessionId = decodeURIComponent(match[1]);
207	192	const executionId = decodeURIComponent(match[2]);
208		− if (!servers.has(sessionId)) return json({ error: "MCP session not found" }, 404);
	193	+ // The shared engine must still hold the paused execution — guards stale ids.
	194	+ if (!(await pausedDetail(executionId))) return json({ error: "MCP session not found" }, 404);
209	195
210	196	const response = await readResumeResponse(request);
211	197	if (!response) return json({ error: "Invalid approval response" }, 400);
212	198
213		− const sessionApprovals =
214		− approvalResponses.get(sessionId) ?? new Map<string, ResumeResponse>();
215		− sessionApprovals.set(executionId, response);
216		− approvalResponses.set(sessionId, sessionApprovals);
217		− const waiter = approvalWaiters.get(sessionId)?.get(executionId);
218		− if (waiter) await Effect.runPromise(Deferred.succeed(waiter, response));
219		−
	199	+ await Effect.runPromise(approvals.recordResponse(executionId, response));
220	200	return json(resumeApprovalResult(executionId, response));
221	201	},
222	202

apps/local/src/serve.test.ts+5−0

@@ -21,6 +21,7 @@ const startTestServer = async (): Promise<string> => { · 20 unmodified lines
21	21	mcp: {
22	22	handleRequest: async () => new Response("ok"),
23	23	handleApprovalRequest: async () => new Response("ok"),
	24	+ handlePausedRequest: async () => new Response("ok"),
24	25	close: async () => {},
25	26	},
26	27	},
@@ -80,6 +81,7 @@ describe("startServer network bind auth", () => { · 53 unmodified lines
80	81	mcp: {
81	82	handleRequest: async () => new Response("ok"),
82	83	handleApprovalRequest: async () => new Response("ok"),
	84	+ handlePausedRequest: async () => new Response("ok"),
83	85	close: async () => {},
84	86	},
85	87	},
@@ -101,6 +103,7 @@ describe("startServer network bind auth", () => { · 15 unmodified lines
101	103	mcp: {
102	104	handleRequest: async () => new Response("ok"),
103	105	handleApprovalRequest: async () => new Response("ok"),
	106	+ handlePausedRequest: async () => new Response("ok"),
104	107	close: async () => {},
105	108	},
106	109	},
@@ -131,6 +134,7 @@ describe("startServer network bind auth", () => { · 24 unmodified lines
131	134	mcp: {
132	135	handleRequest: async () => new Response("ok"),
133	136	handleApprovalRequest: async () => new Response("ok"),
	137	+ handlePausedRequest: async () => new Response("ok"),
134	138	close: async () => {},
135	139	},
136	140	},
@@ -166,6 +170,7 @@ describe("startServer network bind auth", () => { · 29 unmodified lines
166	170	mcp: {
167	171	handleRequest: async () => new Response("ok"),
168	172	handleApprovalRequest: async () => new Response("ok"),
	173	+ handlePausedRequest: async () => new Response("ok"),
169	174	close: async () => {},
170	175	},
171	176	},

apps/local/src/serve.ts+5−1

@@ -326,7 +326,11 @@ export async function startServer(opts: StartServerOptions = {}): Promise<Server · 325 unmodified lines
326	326	}
327	327
328	328	if (url.pathname.startsWith("/api/mcp-sessions/")) {
329		− return maybeWithCorsHeaders(await handlers.mcp.handleApprovalRequest(req));
	329	+ const handler =
	330	+ req.method === "GET"
	331	+ ? handlers.mcp.handlePausedRequest
	332	+ : handlers.mcp.handleApprovalRequest;
	333	+ return maybeWithCorsHeaders(await handler(req));
330	334	}
331	335
332	336	// OAuth result polling — local-only, served outside the typed API

e2e/scenarios/browser-approval.test.tsrenamed from e2e/cloud/browser-approval.test.ts+4−3

@@ -12,9 +12,10 @@ · 11 unmodified lines
12	12	// The policy is removed in an `ensuring` finalizer — a leaked require_approval
13	13	// gate on a shared built-in tool would pause unrelated scenarios.
14	14	//
15		−// Lives under cloud/ for now because cloud is the only host wired for browser
16		−// approval; it moves to scenarios/ (cross-target) as self-host and Cloudflare
17		−// gain the feature.
	15	+// Cross-target: runs on every host that wires browser approval (cloud's Durable
	16	+// Object, self-host's in-process store, Cloudflare's DO). The host differences —
	17	+// where the approval URL points, which engine holds the pause — are invisible
	18	+// here; the scenario only drives the rendered console page.
18	19	import { expect } from "@effect/vitest";
19	20	import { Effect } from "effect";
20	21	import { composePluginApi } from "@executor-js/api/server";

packages/core/api/src/server/mcp-build.ts+7−2

@@ -4,6 +4,7 @@ import { McpErrorReporter, type Principal } from "@executor-js/host-mcp"; · 3 unmodified lines
4	4	import {
5	5	McpEngineBuildError,
6	6	type McpBuildServer,
	7	+ type McpBuildServerOptions,
7	8	} from "@executor-js/host-mcp/in-memory-session-store";
8	9	import { createExecutorMcpServer } from "@executor-js/host-mcp/tool-server";
9	10
@@ -37,7 +38,7 @@ export type McpExecutionStackLayer = Layer.Layer< · 27 unmodified lines
37	38	*/
38	39	export const makeMcpBuildServer =
39	40	(executionStack: McpExecutionStackLayer): McpBuildServer =>
40		− (principal: Principal) =>
	41	+ (principal: Principal, options?: McpBuildServerOptions) =>
41	42	makeExecutionStack(
42	43	principal.accountId,
43	44	principal.organizationId,
@@ -46,7 +47,11 @@ export const makeMcpBuildServer = · 2 unmodified lines
46	47	Effect.map(({ engine }) => engine),
47	48	Effect.provide(executionStack),
48	49	Effect.mapError((cause) => new McpEngineBuildError({ cause })),
49		− Effect.flatMap((engine) => createExecutorMcpServer({ engine })),
	50	+ Effect.flatMap((engine) =>
	51	+ createExecutorMcpServer({ engine, ...(options ?? {}) }).pipe(
	52	+ Effect.map((mcpServer) => ({ mcpServer, engine })),
	53	+ ),
	54	+ ),
50	55	);
51	56
52	57	/**

packages/hosts/mcp/package.json+4−0

@@ -19,6 +19,10 @@ · 18 unmodified lines
19	19	"./browser-approval": {
20	20	"types": "./src/browser-approval.ts",
21	21	"default": "./src/browser-approval.ts"
	22	+ },
	23	+ "./browser-approval-store": {
	24	+ "types": "./src/browser-approval-store.ts",
	25	+ "default": "./src/browser-approval-store.ts"
22	26	}
23	27	},
24	28	"scripts": {

packages/hosts/mcp/src/browser-approval-store.ts+71−0

@@ -0,0 +1,71 @@
	1	+// ---------------------------------------------------------------------------
	2	+// In-process browser-approval store — the single-process equivalent of the
	3	+// Durable Object's persisted approval responses (apps/cloud, host-cloudflare).
	4	+//
	5	+// It is the bridge between the two halves of a browser approval:
	6	+// - the MCP `resume` tool long-polls `store.waitForResponse(executionId)`,
	7	+// - the HTTP approval endpoint records the human's decision via
	8	+// `recordResponse(executionId, response)`, which wakes that waiter.
	9	+//
	10	+// Keyed by executionId alone — execution ids are unique per execution, so one
	11	+// store serves every session in the process. The in-memory MCP session store
	12	+// and the local app both build on it.
	13	+// ---------------------------------------------------------------------------
	14	+
	15	+import { Deferred, Effect } from "effect";
	16	+
	17	+import type { ResumeResponse } from "@executor-js/execution";
	18	+
	19	+import type { BrowserApprovalStore } from "./tool-server";
	20	+
	21	+export interface InProcessBrowserApprovalStore {
	22	+ /** The store the MCP server awaits a decision on (browser elicitation mode). */
	23	+ readonly store: BrowserApprovalStore;
	24	+ /** Record a human's decision, waking any in-flight `waitForResponse`. */
	25	+ readonly recordResponse: (executionId: string, response: ResumeResponse) => Effect.Effect<void>;
	26	+ /** Drop a pending decision/waiter (e.g. when its session is torn down). */
	27	+ readonly forget: (executionId: string) => void;
	28	+}
	29	+
	30	+export const makeInProcessBrowserApprovalStore = (): InProcessBrowserApprovalStore => {
	31	+ const responses = new Map<string, ResumeResponse>();
	32	+ const waiters = new Map<string, Deferred.Deferred<ResumeResponse>>();
	33	+
	34	+ const take = (executionId: string): Effect.Effect<ResumeResponse \| null> =>
	35	+ Effect.sync(() => {
	36	+ const response = responses.get(executionId) ?? null;
	37	+ if (response) responses.delete(executionId);
	38	+ return response;
	39	+ });
	40	+
	41	+ const waitFor = (executionId: string): Effect.Effect<ResumeResponse \| null> =>
	42	+ Effect.gen(function* () {
	43	+ const existing = yield* take(executionId);
	44	+ if (existing) return existing;
	45	+
	46	+ const waiter = waiters.get(executionId) ?? (yield* Deferred.make<ResumeResponse>());
	47	+ waiters.set(executionId, waiter);
	48	+ yield* Deferred.await(waiter).pipe(
	49	+ Effect.ensuring(
	50	+ Effect.sync(() => {
	51	+ if (waiters.get(executionId) === waiter) waiters.delete(executionId);
	52	+ }),
	53	+ ),
	54	+ );
	55	+ return yield* take(executionId);
	56	+ });
	57	+
	58	+ return {
	59	+ store: { takeResponse: take, waitForResponse: waitFor },
	60	+ recordResponse: (executionId, response) =>
	61	+ Effect.gen(function* () {
	62	+ responses.set(executionId, response);
	63	+ const waiter = waiters.get(executionId);
	64	+ if (waiter) yield* Deferred.succeed(waiter, response);
	65	+ }),
	66	+ forget: (executionId) => {
	67	+ responses.delete(executionId);
	68	+ waiters.delete(executionId);
	69	+ },
	70	+ };
	71	+};

packages/hosts/mcp/src/in-memory-session-store.ts+155−19

@@ -1,7 +1,19 @@
1		−import { Data, Effect, Layer } from "effect";
	1	+import { Cause, Data, Effect, Layer } from "effect";
2	2	import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
3	3	import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js";
4	4
	5	+import { formatPausedExecution, type ExecutionEngine } from "@executor-js/execution";
	6	+
	7	+import {
	8	+ approvalUrlForRequest,
	9	+ decodeResumeResponse,
	10	+ formatResumeAcknowledgement,
	11	+ readElicitationMode,
	12	+} from "./browser-approval";
	13	+import {
	14	+ makeInProcessBrowserApprovalStore,
	15	+ type InProcessBrowserApprovalStore,
	16	+} from "./browser-approval-store";
5	17	import { jsonRpcErrorBody } from "./envelope";
6	18	import {
7	19	McpSessionStore,
@@ -10,23 +22,26 @@ import { · 2 unmodified lines
10	22	type McpDispatchResult,
11	23	type Principal,
12	24	} from "./seams";
	25	+import type { BrowserApprovalStore } from "./tool-server";
13	26
14	27	// ---------------------------------------------------------------------------
15	28	// In-process McpSessionStore — the single-node serving store, shared by every
16		−// host that has no cross-isolate session backend (self-host, the Cloudflare
17		−// QuickJS host). Cloud's Durable Object store is the cross-isolate variant of
18		−// the same `McpSessionStore` seam.
	29	+// host that has no cross-isolate session backend (self-host, the local app).
	30	+// Cloud's Durable Object store is the cross-isolate variant of the same
	31	+// `McpSessionStore` seam.
19	32	//
20	33	// In the two-seam envelope the store owns the ENTIRE session lifecycle via
21	34	// `dispatch`: create (no session id + POST initialize), forward (session id
22		−// present), and ownership (cross-bearer). Three Maps keyed by mcp-session-id —
23		−// transports, servers, owners — hold the live in-process sessions. Closing a
24		−// session is just closing its transport + server.
	35	+// present), and ownership (cross-bearer). Maps keyed by mcp-session-id hold the
	36	+// live in-process sessions: transports, servers, owners, and — for the browser
	37	+// approval flow — the per-session engines.
25	38	//
26		−// The engine is a store implementation detail, not an envelope seam: the store
27		−// builds each per-session `McpServer` through the host-supplied `buildServer`
28		−// (the host's execution stack over its own DB + code substrate). The two-seam
29		−// envelope has no engine seam — the store owns engine construction.
	39	+// Browser approval: when the create request carries `?elicitation_mode=browser`,
	40	+// the store builds the session's server in browser mode (an `approvalUrl` + the
	41	+// shared in-process approval store) and keeps the session's engine so the HTTP
	42	+// approval endpoints (`handlePausedRequest` / `handleApprovalRequest`) can read
	43	+// the paused execution and record the human's decision. The Durable Object
	44	+// hosts do the equivalent with `ctx.storage`.
30	45	//
31	46	// `dispatch` returns the transport `Response` to pass through, or:
32	47	// - "not-found" (unknown session id) -> envelope renders 404 -32001
@@ -38,14 +53,42 @@ export class McpEngineBuildError extends Data.TaggedError("McpEngineBuildError") · 5 unmodified lines
38	53	readonly cause: unknown;
39	54	}> {}
40	55
41		−/** Build the per-session `McpServer` for a principal (the host's engine + tools). */
	56	+/** The connected MCP server plus the engine the approval endpoints drive. */
	57	+export interface BuiltMcpServer {
	58	+ readonly mcpServer: McpServer;
	59	+ readonly engine: ExecutionEngine<Cause.YieldableError>;
	60	+}
	61	+
	62	+/** The browser-mode wiring the store hands a build call when a session opts in. */
	63	+export interface McpBuildServerOptions {
	64	+ readonly elicitationMode?:
	65	+ \| { readonly mode: "browser"; readonly approvalUrl: (executionId: string) => string }
	66	+ \| { readonly mode: "model" }
	67	+ \| { readonly mode: "native" };
	68	+ readonly browserApprovalStore?: BrowserApprovalStore;
	69	+}
	70	+
	71	+/** Build the per-session `McpServer` + engine for a principal (the host's engine + tools). */
42	72	export type McpBuildServer = (
43	73	principal: Principal,
44		−) => Effect.Effect<McpServer, McpEngineBuildError>;
	74	+ options?: McpBuildServerOptions,
	75	+) => Effect.Effect<BuiltMcpServer, McpEngineBuildError>;
45	76
46	77	export interface InMemoryMcpSessionStore {
47	78	/** The `McpSessionStore` seam value to hand to `inMemoryMcpSessionsLayer`. */
48	79	readonly store: McpSessionStore["Service"];
	80	+ /**
	81	+ * Serve `GET /api/mcp-sessions/:sessionId/executions/:executionId` — the
	82	+ * paused-execution detail the console approval page renders. Returns the
	83	+ * paused `{ text, structured }` or a 404. Null if the path does not match.
	84	+ */
	85	+ readonly handlePausedRequest: (request: Request) => Promise<Response \| null>;
	86	+ /**
	87	+ * Serve `POST /api/mcp-sessions/:sessionId/executions/:executionId/resume` —
	88	+ * record the human's decision and wake the long-polling `resume` tool call.
	89	+ * Null if the path does not match.
	90	+ */
	91	+ readonly handleApprovalRequest: (request: Request) => Promise<Response \| null>;
49	92	/** Dispose every live session — wire into the host's shutdown (not a seam). */
50	93	readonly close: () => Promise<void>;
51	94	}
@@ -65,6 +108,12 @@ const formatBoundaryError = (error: unknown): unknown => · 13 unmodified lines
65	108	const jsonRpcError = (status: number, code: number, message: string): Response =>
66	109	jsonRpcErrorBody(status, code, message, { cors: false });
67	110
	111	+const json = (value: unknown, status = 200): Response =>
	112	+ new Response(JSON.stringify(value), { status, headers: { "content-type": "application/json" } });
	113	+
	114	+const PAUSED_PATH = /^\/api\/mcp-sessions\/([^/?#]+)\/executions\/([^/?#]+)$/;
	115	+const RESUME_PATH = /^\/api\/mcp-sessions\/([^/?#]+)\/executions\/([^/?#]+)\/resume$/;
	116	+
68	117	/**
69	118	* Build the in-process session store plus an explicit `close()` that disposes
70	119	* all live sessions. `close()` is not part of the seam — it is the host lifetime
@@ -77,6 +126,8 @@ export const makeInMemoryMcpSessionStore = ( · 6 unmodified lines
77	126	const transports = new Map<string, WebStandardStreamableHTTPServerTransport>();
78	127	const servers = new Map<string, McpServer>();
79	128	const owners = new Map<string, Principal>();
	129	+ const engines = new Map<string, ExecutionEngine<Cause.YieldableError>>();
	130	+ const approvals: InProcessBrowserApprovalStore = makeInProcessBrowserApprovalStore();
80	131
81	132	const dispose = async (id: string, opts: { transport?: boolean; server?: boolean } = {}) => {
82	133	const transport = transports.get(id);
@@ -84,6 +135,7 @@ export const makeInMemoryMcpSessionStore = ( · 1 unmodified line
84	135	transports.delete(id);
85	136	servers.delete(id);
86	137	owners.delete(id);
	138	+ engines.delete(id);
87	139	if (opts.transport) await ignoreClose(transport ? () => transport.close() : undefined);
88	140	if (opts.server) await ignoreClose(server ? () => server.close() : undefined);
89	141	};
@@ -126,18 +178,44 @@ export const makeInMemoryMcpSessionStore = ( · 36 unmodified lines
126	178	return runHandleRequest(transport, request);
127	179	};
128	180
	181	+ /**
	182	+ * The browser-mode wiring for a create request: when the client asks for
	183	+ * `elicitation_mode=browser`, build the server with an `approvalUrl` (anchored
	184	+ * at the request origin + the session id, minted on initialize) and the shared
	185	+ * approval store. Otherwise pass the bare model/native mode through.
	186	+ */
	187	+ const buildOptionsFor = (
	188	+ request: Request,
	189	+ sessionId: () => string \| null,
	190	+ ): McpBuildServerOptions => {
	191	+ if (readElicitationMode(request) !== "browser") return { elicitationMode: { mode: "model" } };
	192	+ return {
	193	+ elicitationMode: {
	194	+ mode: "browser",
	195	+ approvalUrl: (executionId) => approvalUrlForRequest(request, executionId, sessionId()),
	196	+ },
	197	+ browserApprovalStore: approvals.store,
	198	+ };
	199	+ };
	200	+
129	201	/** Open a new session: build the server, connect a transport, drive the request. */
130		− const create = (principal: Principal, request: Request): Effect.Effect<McpDispatchResult> =>
131		− buildServer(principal).pipe(
132		− Effect.flatMap((server) =>
	202	+ const create = (principal: Principal, request: Request): Effect.Effect<McpDispatchResult> => {
	203	+ let createdSessionId: string \| null = null;
	204	+ return buildServer(
	205	+ principal,
	206	+ buildOptionsFor(request, () => createdSessionId),
	207	+ ).pipe(
	208	+ Effect.flatMap(({ mcpServer, engine }) =>
133	209	Effect.gen(function* () {
134	210	const transport = new WebStandardStreamableHTTPServerTransport({
135	211	sessionIdGenerator: () => crypto.randomUUID(),
136	212	enableJsonResponse: true,
137	213	onsessioninitialized: (sid) => {
	214	+ createdSessionId = sid;
138	215	transports.set(sid, transport);
139		− servers.set(sid, server);
	216	+ servers.set(sid, mcpServer);
140	217	owners.set(sid, principal);
	218	+ engines.set(sid, engine);
141	219	},
142	220	onsessionclosed: (sid) => void dispose(sid, { server: true }),
143	221	});
@@ -145,12 +223,12 @@ export const makeInMemoryMcpSessionStore = ( · 1 unmodified line
145	223	const sid = transport.sessionId;
146	224	if (sid) void dispose(sid, { server: true });
147	225	};
148		− yield* Effect.promise(() => server.connect(transport));
	226	+ yield* Effect.promise(() => mcpServer.connect(transport));
149	227	// The session id is minted on the first (initialize) request, so we
150	228	// drive `handleRequest` here; if no id results we close eagerly.
151	229	return yield* runHandleRequest(transport, request, () => {
152	230	void ignoreClose(() => transport.close());
153		− void ignoreClose(() => server.close());
	231	+ void ignoreClose(() => mcpServer.close());
154	232	});
155	233	}),
156	234	),
@@ -159,6 +237,7 @@ export const makeInMemoryMcpSessionStore = ( · 2 unmodified lines
159	237	Effect.succeed(jsonRpcError(500, -32603, "Internal server error")),
160	238	),
161	239	);
	240	+ };
162	241
163	242	const store: McpSessionStore["Service"] = {
164	243	dispatch: ({ request, principal, sessionId }: McpDispatchInput) =>
@@ -167,8 +246,65 @@ export const makeInMemoryMcpSessionStore = ( · 2 unmodified lines
167	246	Effect.promise(() => dispose(sessionId, { transport: true, server: true })),
168	247	};
169	248
	249	+ /** Resolve a paused execution from the session that owns it, for HTTP approval. */
	250	+ const pausedFromSession = (
	251	+ sessionId: string,
	252	+ executionId: string,
	253	+ ): Promise<ReturnType<typeof formatPausedExecution> \| null> => {
	254	+ const engine = engines.get(sessionId);
	255	+ if (!engine) return Promise.resolve(null);
	256	+ return Effect.runPromise(
	257	+ engine.getPausedExecution(executionId).pipe(
	258	+ Effect.map((paused) => (paused ? formatPausedExecution(paused) : null)),
	259	+ Effect.orElseSucceed(() => null),
	260	+ ),
	261	+ );
	262	+ };
	263	+
	264	+ const handlePausedRequest = async (request: Request): Promise<Response \| null> => {
	265	+ const match = PAUSED_PATH.exec(new URL(request.url).pathname);
	266	+ if (!match) return null;
	267	+ if (request.method !== "GET") return json({ error: "Method not allowed" }, 405);
	268	+ const paused = await pausedFromSession(
	269	+ decodeURIComponent(match[1]!),
	270	+ decodeURIComponent(match[2]!),
	271	+ );
	272	+ if (!paused) return json({ error: "Paused execution not found" }, 404);
	273	+ return json({ text: paused.text, structured: paused.structured });
	274	+ };
	275	+
	276	+ const handleApprovalRequest = async (request: Request): Promise<Response \| null> => {
	277	+ const match = RESUME_PATH.exec(new URL(request.url).pathname);
	278	+ if (!match) return null;
	279	+ if (request.method !== "POST") return json({ error: "Method not allowed" }, 405);
	280	+
	281	+ const sessionId = decodeURIComponent(match[1]!);
	282	+ const executionId = decodeURIComponent(match[2]!);
	283	+ // The session must still hold the paused execution — guards stale ids and
	284	+ // confirms the execution belongs to this session before recording.
	285	+ const paused = await pausedFromSession(sessionId, executionId);
	286	+ if (!paused) return json({ error: "Paused execution not found" }, 404);
	287	+
	288	+ const raw = await Effect.runPromise(
	289	+ Effect.tryPromise({ try: () => request.json(), catch: () => null }).pipe(
	290	+ Effect.orElseSucceed(() => null),
	291	+ ),
	292	+ );
	293	+ const response = raw === null ? null : decodeResumeResponse(raw);
	294	+ if (!response) return json({ error: "Invalid approval response" }, 400);
	295	+
	296	+ await Effect.runPromise(approvals.recordResponse(executionId, response));
	297	+ return json({
	298	+ status: "completed",
	299	+ ...formatResumeAcknowledgement(executionId, response),
	300	+ isError: false,
	301	+ });
	302	+ };
	303	+
170	304	return {
171	305	store,
	306	+ handlePausedRequest,
	307	+ handleApprovalRequest,
172	308	close: async () => {
173	309	const ids = new Set([...transports.keys(), ...servers.keys()]);
174	310	await Promise.all([...ids].map((id) => dispose(id, { transport: true, server: true })));

packages/react/src/routes/resume.$executionId.tsx+45−2

@@ -5,7 +5,6 @@ import * as Atom from "effect/unstable/reactivity/Atom"; · 4 unmodified lines
5	5	import { createFileRoute } from "@tanstack/react-router";
6	6
7	7	import { ResumeApprovalPage, ResumeApprovalPageView } from "../pages/resume-approval";
8		−import { pausedExecutionAtom } from "../api/atoms";
9	8	import type { ElicitationAction } from "../components/elicitation-approval";
10	9
11	10	const SearchParams = Schema.toStandardSchemaV1(
@@ -31,6 +30,48 @@ class LocalMcpResumeError extends Data.TaggedError("LocalMcpResumeError")<{ · 19 unmodified lines
31	30	readonly message: string;
32	31	}> {}
33	32
	33	+const McpPausedExecutionInfo = Schema.Struct({
	34	+ text: Schema.String,
	35	+ structured: Schema.Unknown,
	36	+});
	37	+const decodeMcpPausedExecutionInfo = Schema.decodeUnknownOption(McpPausedExecutionInfo);
	38	+
	39	+// Paused-execution detail for the in-process / Cloudflare hosts, fetched
	40	+// session-scoped: the MCP paused execution lives in its session's engine, so it
	41	+// is resolved through `/api/mcp-sessions/:id/...`, not the session-less
	42	+// `/api/executions/:id` (which hits a different engine and can't see it). Cloud
	43	+// serves the equivalent through its own route + Durable Object RPC.
	44	+const mcpPausedExecutionAtom = Atom.family(
	45	+ (key: { readonly mcpSessionId: string; readonly executionId: string }) =>
	46	+ Atom.make(
	47	+ Effect.gen(function* () {
	48	+ const response = yield* Effect.tryPromise({
	49	+ try: () =>
	50	+ fetch(
	51	+ `/api/mcp-sessions/${encodeURIComponent(key.mcpSessionId)}/executions/${encodeURIComponent(key.executionId)}`,
	52	+ ),
	53	+ catch: () => new LocalMcpResumeError({ message: "Failed to load the paused execution." }),
	54	+ });
	55	+ if (!response.ok) {
	56	+ return yield* new LocalMcpResumeError({
	57	+ message: `Paused execution unavailable (${response.status}).`,
	58	+ });
	59	+ }
	60	+ const body = yield* Effect.tryPromise({
	61	+ try: () => response.json(),
	62	+ catch: () => new LocalMcpResumeError({ message: "Paused response was not valid JSON." }),
	63	+ });
	64	+ const decoded = decodeMcpPausedExecutionInfo(body);
	65	+ if (Option.isNone(decoded)) {
	66	+ return yield* new LocalMcpResumeError({
	67	+ message: "Paused response had an unexpected shape.",
	68	+ });
	69	+ }
	70	+ return decoded.value;
	71	+ }),
	72	+ ),
	73	+);
	74	+
34	75	type LocalMcpResumeInput = {
35	76	readonly mcpSessionId: string;
36	77	readonly executionId: string;
@@ -96,7 +137,9 @@ function RouteComponent() { · 59 unmodified lines
96	137	}
97	138
98	139	function LocalMcpResumeApproval(props: { executionId: string; mcpSessionId: string }) {
99		− const paused = useAtomValue(pausedExecutionAtom(props.executionId));
	140	+ const paused = useAtomValue(
	141	+ mcpPausedExecutionAtom({ mcpSessionId: props.mcpSessionId, executionId: props.executionId }),
	142	+ );
100	143	const doResume = useAtomSet(resumeLocalMcpExecution, { mode: "promiseExit" });
101	144	const resume = useCallback(
102	145	(executionId: string, action: ElicitationAction, content?: Record<string, unknown>) =>

self-host: browser approval of gated MCP actions

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code