diff --git a/apps/cloud/src/env-augment.d.ts b/apps/cloud/src/env-augment.d.ts index 4ef130bfa..7a5d86075 100644 --- a/apps/cloud/src/env-augment.d.ts +++ b/apps/cloud/src/env-augment.d.ts @@ -67,6 +67,8 @@ declare global { MCP_RESOURCE_ORIGIN?: string; MCP_SESSION_TIMEOUT_MS?: string; MCP_PAUSED_SESSION_IDLE_TIMEOUT_MS?: string; + /** HMAC key for MCP 2026-07-28 continuation state (32+ byte secret). */ + MCP_REQUEST_STATE_KEY?: string; NODE_ENV?: string; // Shared with frontend diff --git a/apps/cloud/src/mcp/agent-handler.ts b/apps/cloud/src/mcp/agent-handler.ts index 7738e79c6..b1ec9ad6e 100644 --- a/apps/cloud/src/mcp/agent-handler.ts +++ b/apps/cloud/src/mcp/agent-handler.ts @@ -9,6 +9,7 @@ import { type AuthOutcome, type McpResource, } from "@executor-js/host-mcp"; +import { requestBodyFromRequest } from "@executor-js/host-mcp/tool-server-v2"; import { currentPropagationHeaders, readArtifactsEnabled, @@ -16,26 +17,21 @@ import { withVerifiedIdentityHeaders, } from "@executor-js/cloudflare/mcp/do-headers"; import type { McpSessionProps } from "@executor-js/cloudflare/mcp/agent-durable-object"; +import { + classifyMcpProtocolEra, + makeMcpModernRequestRouter, + mcpCorsPreflightResponse, + requireMcpRequestStateKey, +} from "@executor-js/cloudflare/mcp/modern-request-router"; +import { mcpExecutionOwnerDirectoryFromNamespace } from "@executor-js/cloudflare/mcp/execution-owner-directory"; import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; import { wrapMcpSseResponse } from "../observability/memory-metrics"; import { WorkerTelemetryLive } from "../observability/telemetry"; import { cloudMcpAuth } from "./auth-provider"; -import { McpSessionDOSqlite } from "./session-durable-object"; +import { McpSessionDOSqlite, makeCloudModernMcpServerBuilder } from "./session-durable-object"; import { parseTraceparent } from "./traceparent"; -const corsPreflightResponse = (): Response => - new Response(null, { - status: 204, - headers: { - "access-control-allow-origin": "*", - "access-control-allow-methods": "GET, POST, DELETE, OPTIONS", - "access-control-allow-headers": - "content-type, authorization, mcp-session-id, accept, mcp-protocol-version", - "access-control-expose-headers": "mcp-session-id, WWW-Authenticate", - }, - }); - const jsonRpcResponse = ( status: number, code: number, @@ -141,6 +137,7 @@ const propsForPrincipal = ( }); export const makeCloudMcpAgentHandler = () => { + const modern = makeMcpModernRequestRouter(); const serveOptions = { binding: "MCP_SESSION", transport: "streamable-http", @@ -158,7 +155,9 @@ export const makeCloudMcpAgentHandler = () => { const ALLOWED_METHODS = new Set(["GET", "POST", "DELETE", "OPTIONS"]); return async (request: Request, env: Env, ctx: ExecutionContext): Promise => { - if (request.method === "OPTIONS") return corsPreflightResponse(); + if (request.method === "OPTIONS") { + return mcpCorsPreflightResponse(request.headers.get("access-control-request-headers")); + } // The old envelope (packages/hosts/mcp/src/envelope.ts) answered anything // outside GET/POST/DELETE/OPTIONS with a JSON-RPC 405; the agents SDK // handler only understands its own transport verbs and falls through to @@ -188,6 +187,36 @@ export const makeCloudMcpAgentHandler = () => { return renderAuthError(auth, request, outcome); } + const parsedBody = await Effect.runPromise(requestBodyFromRequest(request)); + const era = await classifyMcpProtocolEra(request, parsedBody); + if (era === "modern") { + const resource = resourceFromPath(request); + const props = await runTraced( + request, + propsForPrincipal(request, outcome.principal, resource), + ); + (ctx as ExecutionContext & { props?: McpSessionProps }).props = props; + const forwarded = withVerifiedIdentityHeaders( + request, + { + accountId: outcome.principal.accountId, + organizationId: outcome.principal.organizationId, + }, + resource, + ); + return modern.fetch({ + request: forwarded, + parsedBody, + principal: outcome.principal, + resource, + props, + requestStateSigningKey: requireMcpRequestStateKey(env.MCP_REQUEST_STATE_KEY), + builder: makeCloudModernMcpServerBuilder(props.session), + sessions: env.MCP_SESSION, + executionOwners: mcpExecutionOwnerDirectoryFromNamespace(env.MCP_EXECUTION_OWNER), + }); + } + if (!sessionId && request.method === "DELETE") { // Matches the old envelope's contract (@modelcontextprotocol/sdk's // `WebStandardStreamableHTTPServerTransport.handleDeleteRequest`): 200, diff --git a/apps/cloud/src/mcp/session-durable-object.ts b/apps/cloud/src/mcp/session-durable-object.ts index 31d6d4bf6..ee3450fc2 100644 --- a/apps/cloud/src/mcp/session-durable-object.ts +++ b/apps/cloud/src/mcp/session-durable-object.ts @@ -16,6 +16,7 @@ import { env } from "cloudflare:workers"; import { Data, Effect, Layer } from "effect"; import type { Cause } from "effect"; +import type * as Tracer from "effect/Tracer"; import * as OtelTracer from "@effect/opentelemetry/Tracer"; import { drizzle } from "drizzle-orm/postgres-js"; import postgres, { type Sql } from "postgres"; @@ -23,7 +24,11 @@ import postgres, { type Sql } from "postgres"; import { PAUSED_APPROVAL_TIMEOUT_MS, createExecutorMcpServer, + type PausedExecutionHooks, + type ResumeFallbackOutcome, } from "@executor-js/host-mcp/tool-server"; +import { buildMcpServerV2 } from "@executor-js/host-mcp/tool-server-v2"; +import type { McpModernServerBuilder, Principal } from "@executor-js/host-mcp"; import { buildResumeApprovalUrl } from "@executor-js/host-mcp/browser-approval"; import { artifactUrlFor } from "@executor-js/host-mcp/create-artifact"; import { makeAssetsShellHtmlLoader } from "@executor-js/mcp-apps-shell/worker"; @@ -31,18 +36,20 @@ import { smokeRenderArtifact } from "@executor-js/mcp-apps-shell/smoke-render"; import { McpAgentSessionDOBase, type BuiltMcpServer, + type BuiltModernMcpRuntime, type IncomingTraceHeaders, type McpApprovalOwner, type McpSessionModelResumeResult, type McpSessionInit, type SessionMeta, } from "@executor-js/cloudflare/mcp/agent-durable-object"; +import { requireMcpRequestStateKey } from "@executor-js/cloudflare/mcp/modern-request-router"; import { mcpExecutionOwnerDirectoryFromNamespace, type McpExecutionOwnerDirectory, type McpExecutionOwnerRoute, } from "@executor-js/cloudflare/mcp/execution-owner-directory"; -import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; +import { mcpSessionStubForOwner } from "@executor-js/cloudflare/mcp/session-stub"; import { buildExecuteDescription, type ResumeResponse } from "@executor-js/execution"; // The DO meters executions just like the HTTP `/api/*` plane: it builds its @@ -117,6 +124,10 @@ class McpModelResumeForwardError extends Data.TaggedError("McpModelResumeForward readonly cause: unknown; }> {} +class CloudModernMcpBuildError extends Data.TaggedError("CloudModernMcpBuildError")<{ + readonly cause: unknown; +}> {} + /** * The DO keeps one postgres.js client for the MCP session runtime. postgres.js * closes idle sockets quickly, while the runtime object stays alive so the MCP @@ -168,6 +179,127 @@ const loadAppShellHtml = makeAssetsShellHtmlLoader({ import("virtual:executor-mcp-apps-shell-dev-html").then((mod) => mod.devShellHtml), }); +const resolveCloudSessionMeta = (token: McpSessionInit, dbHandle: CloudSessionDbHandle) => + Effect.gen(function* () { + const org = yield* resolveOrganization(token.organizationId); + if (!org) { + return yield* new OrganizationNotFoundError({ organizationId: token.organizationId }); + } + return { + organizationId: org.id, + organizationName: org.name, + organizationSlug: org.slug, + userId: token.userId, + resource: token.resource, + elicitationMode: token.elicitationMode, + artifactsEnabled: token.artifactsEnabled, + webOrigin: token.webOrigin, + } satisfies SessionMeta; + }).pipe(Effect.provide(makeSessionServices(dbHandle))); + +const makeCloudExecutionRuntime = (sessionMeta: SessionMeta, dbHandle: CloudSessionDbHandle) => + Effect.gen(function* () { + yield* Effect.promise(() => preloadQuickJs()); + const { executor, engine } = yield* makeExecutionStack( + sessionMeta.userId, + sessionMeta.organizationId, + sessionMeta.organizationName, + { mcpResource: sessionMeta.resource }, + ).pipe( + Effect.provide(CloudMeteredExecutionStackLayer.pipe(Layer.provide(AutumnService.Default))), + Effect.withSpan("McpSessionDOSqlite.makeExecutionStack"), + ); + const description = yield* buildExecuteDescription(executor).pipe( + Effect.withSpan("mcp.execute.description.build"), + ); + return { executor, engine, description }; + }).pipe(Effect.provide(makeSessionServices(dbHandle))); + +type CloudExecutionRuntime = Effect.Success>; + +type CloudModernLifecycle = { + readonly pausedExecutionHooks?: PausedExecutionHooks; + readonly resumeFallback?: ( + executionId: string, + response: ResumeResponse, + ) => Effect.Effect; + readonly parentSpan?: () => Tracer.AnySpan | undefined; +}; + +const makeCloudModernRuntime = ( + sessionMeta: SessionMeta, + runtime: CloudExecutionRuntime, + lifecycle: CloudModernLifecycle = {}, +): BuiltModernMcpRuntime => ({ + engine: runtime.engine, + buildServer: (options) => + buildMcpServerV2({ + engine: runtime.engine, + description: runtime.description, + artifacts: runtime.executor.artifacts, + connections: runtime.executor.connections, + artifactsEnabled: sessionMeta.artifactsEnabled ?? true, + loadAppShellHtml, + smokeRenderArtifact, + artifactUrl: artifactUrlFor( + env.VITE_PUBLIC_SITE_URL ?? "https://executor.sh", + sessionMeta.organizationSlug, + ), + debug: env.EXECUTOR_MCP_DEBUG === "true", + elicitationMode: { mode: "native" }, + ...(lifecycle.parentSpan ? { parentSpan: lifecycle.parentSpan } : {}), + ...(lifecycle.pausedExecutionHooks + ? { + pausedExecutionHooks: lifecycle.pausedExecutionHooks, + pausedExecutionLeaseMs: PAUSED_APPROVAL_TIMEOUT_MS, + } + : {}), + ...(lifecycle.resumeFallback ? { resumeFallback: lifecycle.resumeFallback } : {}), + ...options, + }), +}); + +const closeModernServerWithDb = Promise }>( + server: Server, + dbHandle: CloudSessionDbHandle, +): Server => { + const closeServer = server.close.bind(server); + server.close = () => + Effect.runPromise( + Effect.promise(closeServer).pipe(Effect.ensuring(Effect.promise(() => dbHandle.end()))), + ); + return server; +}; + +/** Build one worker-side stateless SDK v2 server over a fresh cloud runtime. */ +export const makeCloudModernMcpServerBuilder = ( + session: McpSessionInit, +): McpModernServerBuilder["Service"] => ({ + build: (principal: Principal, options) => { + const dbHandle = makeEphemeralDb(); + const { resource, ...requestOptions } = options; + const token: McpSessionInit = { + ...session, + userId: principal.accountId, + organizationId: principal.organizationId, + resource, + }; + return resolveCloudSessionMeta(token, dbHandle).pipe( + Effect.flatMap((sessionMeta) => + makeCloudExecutionRuntime(sessionMeta, dbHandle).pipe( + Effect.map((runtime) => ({ runtime, sessionMeta })), + ), + ), + Effect.flatMap(({ runtime, sessionMeta }) => + makeCloudModernRuntime(sessionMeta, runtime).buildServer(requestOptions), + ), + Effect.map((server) => closeModernServerWithDb(server, dbHandle)), + Effect.tapCause(() => Effect.promise(() => dbHandle.end())), + Effect.mapError((cause) => new CloudModernMcpBuildError({ cause })), + ); + }, +}); + // --------------------------------------------------------------------------- // Durable Object // --------------------------------------------------------------------------- @@ -195,7 +327,7 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase { return Effect.tryPromise({ try: () => - mcpSessionStub(env.MCP_SESSION, owner.sessionId).resumeExecutionForModel( + mcpSessionStubForOwner(env.MCP_SESSION, owner).resumeExecutionForModel( executionId, identity, response, @@ -213,23 +345,8 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase { const dbHandle = makeEphemeralDb(); - return Effect.gen(function* () { - const org = yield* resolveOrganization(token.organizationId); - if (!org) { - return yield* new OrganizationNotFoundError({ organizationId: token.organizationId }); - } - return { - organizationId: org.id, - organizationName: org.name, - organizationSlug: org.slug, - userId: token.userId, - resource: token.resource, - elicitationMode: token.elicitationMode, - artifactsEnabled: token.artifactsEnabled, - } satisfies SessionMeta; - }).pipe( + return resolveCloudSessionMeta(token, dbHandle).pipe( Effect.withSpan("McpSessionDOSqlite.resolveSessionMeta"), - Effect.provide(makeSessionServices(dbHandle)), Effect.ensuring(Effect.promise(() => dbHandle.end())), // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: a vanished org is a defect; the worker already verified the bearer Effect.orDie, @@ -242,34 +359,13 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase { const self = this; return Effect.gen(function* () { - // QuickJS-WASM must be loaded before anything asks for a sandbox: the - // default variant cannot fetch its own `.wasm` on Workers. Cloud runs - // user `execute` code on the dynamic-worker runtime, but the artifact - // smoke render is a QuickJS sandbox on every host — without this it fails - // open on each create and the check silently does nothing. - // Idempotent per isolate. - yield* Effect.promise(() => preloadQuickJs()); - const { executor, engine } = yield* makeExecutionStack( - sessionMeta.userId, - sessionMeta.organizationId, - sessionMeta.organizationName, - { mcpResource: sessionMeta.resource }, - ).pipe( - // The metered stack tracks each execution to Autumn. It requires - // `AutumnService | DbService`; `AutumnService.Default` is provided here - // (it only reads `env`, no further deps), and `DbService` flows from the - // outer `makeSessionServices`. When `AUTUMN_SECRET_KEY` is unset the - // billing service degrades to a no-op tracker, so this stays inert in - // cloud dev/preview environments that run without a billing backend. - Effect.provide(CloudMeteredExecutionStackLayer.pipe(Layer.provide(AutumnService.Default))), - Effect.withSpan("McpSessionDOSqlite.makeExecutionStack"), - ); - // Build the description here so `executor.connections.list()` stays under - // the DO startup span and the MCP SDK receives a concrete string instead - // of invoking `engine.getDescription` across its async boundary. - const description = yield* buildExecuteDescription(executor).pipe( - Effect.withSpan("mcp.execute.description.build"), - ); + const runtime = yield* makeCloudExecutionRuntime(sessionMeta, dbHandle); + const { executor, engine, description } = runtime; + const modernRuntime = makeCloudModernRuntime(sessionMeta, runtime, { + pausedExecutionHooks: self.modernPausedExecutionHooks, + resumeFallback: self.modernModelResumeFallback, + parentSpan: () => self.currentParentSpan(), + }); const sessionElicitationMode = sessionMeta.elicitationMode ?? "model"; const mcpServer = yield* createExecutorMcpServer({ engine, @@ -310,15 +406,37 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase { + const self = this; + return makeCloudExecutionRuntime(sessionMeta, dbHandle).pipe( + Effect.map((runtime) => + makeCloudModernRuntime(sessionMeta, runtime, { + pausedExecutionHooks: self.modernPausedExecutionHooks, + resumeFallback: self.modernModelResumeFallback, + parentSpan: () => self.currentParentSpan(), + }), + ), + Effect.withSpan("McpSessionDOSqlite.buildModernMcpRuntime"), + // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: runtime-build failures surface through the base RPC cleanup path + Effect.orDie, + ); + } + + protected override modernRequestStateSigningKey(): string { + return requireMcpRequestStateKey(env.MCP_REQUEST_STATE_KEY); + } + protected override withTelemetry( effect: Effect.Effect, incoming?: IncomingTraceHeaders, diff --git a/apps/cloud/src/mcp/telemetry-modern.test.ts b/apps/cloud/src/mcp/telemetry-modern.test.ts new file mode 100644 index 000000000..66db34036 --- /dev/null +++ b/apps/cloud/src/mcp/telemetry-modern.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect } from "effect"; +import type * as Tracer from "effect/Tracer"; + +import { annotateMcpRequest } from "./telemetry"; + +const makeRecordingTracer = (): { + readonly tracer: Tracer.Tracer; + readonly requestAttributes: () => ReadonlyMap | undefined; +} => { + const recorded: Array<{ + readonly name: string; + readonly attributes: Map; + }> = []; + const tracer: Tracer.Tracer = { + span: (options) => { + const attributes = new Map(); + recorded.push({ name: options.name, attributes }); + let status: Tracer.SpanStatus = { _tag: "Started", startTime: options.startTime }; + return { + _tag: "Span", + name: options.name, + spanId: `span-${recorded.length}`, + traceId: "trace-modern", + parent: options.parent, + annotations: options.annotations, + get status() { + return status; + }, + attributes, + links: options.links, + sampled: options.sampled, + kind: options.kind, + end: (endTime, exit) => { + status = { _tag: "Ended", startTime: options.startTime, endTime, exit }; + }, + attribute: (key, value) => { + attributes.set(key, value); + }, + event: () => undefined, + addLinks: () => undefined, + }; + }, + }; + return { + tracer, + requestAttributes: () => recorded.find(({ name }) => name === "mcp.request")?.attributes, + }; +}; + +describe("annotateMcpRequest modern envelope", () => { + it.effect("records the 2026 protocol version outside initialize", () => { + const { tracer, requestAttributes } = makeRecordingTracer(); + const request = new Request("https://executor.sh/mcp", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "tools/list", + params: { + _meta: { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientCapabilities": {}, + }, + }, + }), + }); + + return Effect.gen(function* () { + yield* annotateMcpRequest(request, { token: null, parseBody: true }); + const attributes = requestAttributes(); + expect(attributes?.get("mcp.rpc.method")).toBe("tools/list"); + expect(attributes?.get("mcp.client.protocol_version")).toBe("2026-07-28"); + }).pipe(Effect.withSpan("mcp.request"), Effect.withTracer(tracer)); + }); +}); diff --git a/apps/cloud/src/mcp/telemetry.ts b/apps/cloud/src/mcp/telemetry.ts index 5f4c370f7..1e2d63133 100644 --- a/apps/cloud/src/mcp/telemetry.ts +++ b/apps/cloud/src/mcp/telemetry.ts @@ -102,6 +102,14 @@ const InitializeParams = Schema.Struct({ capabilities: Schema.optional(UnknownRecord), }); +const ModernEnvelopeParams = Schema.Struct({ + _meta: Schema.optional( + Schema.Struct({ + "io.modelcontextprotocol/protocolVersion": Schema.optional(Schema.String), + }), + ), +}); + const NamedParams = Schema.Struct({ name: Schema.optional(Schema.String) }); const UriParams = Schema.Struct({ uri: Schema.optional(Schema.String) }); @@ -119,6 +127,7 @@ const decodeJsonRpcEnvelopeString = Schema.decodeUnknownOption( Schema.fromJsonString(JsonRpcEnvelope), ); const decodeInitializeParams = Schema.decodeUnknownOption(InitializeParams); +const decodeModernEnvelopeParams = Schema.decodeUnknownOption(ModernEnvelopeParams); const decodeNamedParams = Schema.decodeUnknownOption(NamedParams); const decodeUriParams = Schema.decodeUnknownOption(UriParams); const decodeCancelledParams = Schema.decodeUnknownOption(CancelledParams); @@ -136,7 +145,14 @@ const readJsonRpcEnvelope = (request: Request): Effect.Effect => { const params = envelope.params ?? {}; - return Match.value(envelope.method).pipe( + const protocolAttrs = Option.match(decodeModernEnvelopeParams(params), { + onNone: () => ({}), + onSome: (modern) => { + const protocolVersion = modern._meta?.["io.modelcontextprotocol/protocolVersion"]; + return protocolVersion ? { "mcp.client.protocol_version": protocolVersion } : {}; + }, + }); + const methodSpecific = Match.value(envelope.method).pipe( Match.when("initialize", () => Option.match(decodeInitializeParams(params), { onNone: () => ({}) as Record, @@ -181,6 +197,7 @@ const methodAttrs = (envelope: JsonRpcEnvelope): Record => { Match.option, Option.getOrElse(() => ({}) as Record), ); + return { ...protocolAttrs, ...methodSpecific }; }; const replyAttrs = (envelope: JsonRpcEnvelope): Record => { diff --git a/apps/cloud/wrangler.jsonc b/apps/cloud/wrangler.jsonc index dfd9bbd80..121306674 100644 --- a/apps/cloud/wrangler.jsonc +++ b/apps/cloud/wrangler.jsonc @@ -96,6 +96,10 @@ "binding": "LOADER", }, ], + // DEPLOYMENT PREREQUISITE: MCP 2026-07-28 requestState is shared between + // stateless Worker isolates and session DOs. Configure the same 32+ byte + // secret for both with `wrangler secret put MCP_REQUEST_STATE_KEY`. + // It must never be placed in `vars` or generated independently per isolate. "vars": { "VITE_PUBLIC_SITE_URL": "https://executor.sh", "VITE_PUBLIC_POSTHOG_KEY": "phc_nNLrNMALpRsfrEkZovUkfMxYbcJvHnsJHeoSPavprgLL", diff --git a/apps/host-cloudflare/src/config.ts b/apps/host-cloudflare/src/config.ts index c397c4ef8..84499be5f 100644 --- a/apps/host-cloudflare/src/config.ts +++ b/apps/host-cloudflare/src/config.ts @@ -47,6 +47,8 @@ export interface CloudflareEnv { readonly SELF_HOSTED_ORG_SLUG?: string; /** At-rest secret-encryption key (a `wrangler secret`, NOT a var). */ readonly EXECUTOR_SECRET_KEY?: string; + /** HMAC key for MCP 2026-07-28 continuation state (32+ byte secret). */ + readonly MCP_REQUEST_STATE_KEY?: string; readonly ALLOW_LOCAL_NETWORK?: string; readonly VITE_PUBLIC_SITE_URL?: string; /** diff --git a/apps/host-cloudflare/src/mcp/agent-handler.ts b/apps/host-cloudflare/src/mcp/agent-handler.ts index 5ec09fc1b..b1345ddf6 100644 --- a/apps/host-cloudflare/src/mcp/agent-handler.ts +++ b/apps/host-cloudflare/src/mcp/agent-handler.ts @@ -7,6 +7,7 @@ import { type AuthOutcome, type Principal, } from "@executor-js/host-mcp"; +import { requestBodyFromRequest } from "@executor-js/host-mcp/tool-server-v2"; import { currentPropagationHeaders, readArtifactsEnabled, @@ -14,23 +15,18 @@ import { withVerifiedIdentityHeaders, } from "@executor-js/cloudflare/mcp/do-headers"; import type { McpSessionProps } from "@executor-js/cloudflare/mcp/agent-durable-object"; +import { + classifyMcpProtocolEra, + makeMcpModernRequestRouter, + mcpCorsPreflightResponse, + requireMcpRequestStateKey, +} from "@executor-js/cloudflare/mcp/modern-request-router"; +import { mcpExecutionOwnerDirectoryFromNamespace } from "@executor-js/cloudflare/mcp/execution-owner-directory"; import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; import type { CloudflareConfig, CloudflareEnv } from "../config"; import { cloudflareAccessMcpAuth } from "./auth"; -import { McpSessionDO } from "./session-durable-object"; - -const corsPreflightResponse = (): Response => - new Response(null, { - status: 204, - headers: { - "access-control-allow-origin": "*", - "access-control-allow-methods": "GET, POST, DELETE, OPTIONS", - "access-control-allow-headers": - "content-type, authorization, mcp-session-id, accept, mcp-protocol-version", - "access-control-expose-headers": "mcp-session-id, WWW-Authenticate", - }, - }); +import { McpSessionDO, makeCloudflareModernMcpServerBuilder } from "./session-durable-object"; const jsonRpcResponse = ( status: number, @@ -91,13 +87,16 @@ const propsForPrincipal = ( }); export const makeCloudflareMcpAgentHandler = (config: CloudflareConfig) => { + const modern = makeMcpModernRequestRouter(); const serve = McpSessionDO.serve("/mcp", { binding: "MCP_SESSION", transport: "streamable-http", }); return async (request: Request, env: CloudflareEnv, ctx: ExecutionContext): Promise => { - if (request.method === "OPTIONS") return corsPreflightResponse(); + if (request.method === "OPTIONS") { + return mcpCorsPreflightResponse(request.headers.get("access-control-request-headers")); + } const sessionId = request.headers.get("mcp-session-id"); const { auth, outcome } = await Effect.runPromise(authenticate(request, config)); @@ -114,6 +113,32 @@ export const makeCloudflareMcpAgentHandler = (config: CloudflareConfig) => { return renderAuthError(auth, request, outcome); } + const parsedBody = await Effect.runPromise(requestBodyFromRequest(request)); + const era = await classifyMcpProtocolEra(request, parsedBody); + if (era === "modern") { + const props = await Effect.runPromise(propsForPrincipal(request, outcome.principal)); + (ctx as ExecutionContext & { props?: McpSessionProps }).props = props; + const forwarded = withVerifiedIdentityHeaders( + request, + { + accountId: outcome.principal.accountId, + organizationId: outcome.principal.organizationId, + }, + defaultMcpResource, + ); + return modern.fetch({ + request: forwarded, + parsedBody, + principal: outcome.principal, + resource: defaultMcpResource, + props, + requestStateSigningKey: requireMcpRequestStateKey(env.MCP_REQUEST_STATE_KEY), + builder: makeCloudflareModernMcpServerBuilder(env, config, props.session), + sessions: env.MCP_SESSION, + executionOwners: mcpExecutionOwnerDirectoryFromNamespace(env.MCP_EXECUTION_OWNER), + }); + } + if (!sessionId && request.method === "DELETE") { return new Response(null, { status: 204, headers: { "access-control-allow-origin": "*" } }); } diff --git a/apps/host-cloudflare/src/mcp/session-durable-object.ts b/apps/host-cloudflare/src/mcp/session-durable-object.ts index e75199dc0..4195de098 100644 --- a/apps/host-cloudflare/src/mcp/session-durable-object.ts +++ b/apps/host-cloudflare/src/mcp/session-durable-object.ts @@ -3,7 +3,11 @@ import { Data, Effect } from "effect"; import { PAUSED_APPROVAL_TIMEOUT_MS, createExecutorMcpServer, + type PausedExecutionHooks, + type ResumeFallbackOutcome, } from "@executor-js/host-mcp/tool-server"; +import { buildMcpServerV2 } from "@executor-js/host-mcp/tool-server-v2"; +import type { McpModernServerBuilder, Principal } from "@executor-js/host-mcp"; import { buildResumeApprovalUrl } from "@executor-js/host-mcp/browser-approval"; import { artifactUrlFor } from "@executor-js/host-mcp/create-artifact"; import { makeAssetsShellHtmlLoader } from "@executor-js/mcp-apps-shell/worker"; @@ -12,18 +16,20 @@ import type { ExecutorDbHandle } from "@executor-js/api/server"; import { McpAgentSessionDOBase, type BuiltMcpServer, + type BuiltModernMcpRuntime, type McpApprovalOwner, type McpSessionModelResumeResult, type McpSessionInit, type SessionMeta, } from "@executor-js/cloudflare/mcp/agent-durable-object"; +import { requireMcpRequestStateKey } from "@executor-js/cloudflare/mcp/modern-request-router"; import { mcpExecutionOwnerDirectoryFromNamespace, type McpExecutionOwnerDirectory, type McpExecutionOwnerRoute, } from "@executor-js/cloudflare/mcp/execution-owner-directory"; -import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; -import type { ResumeResponse } from "@executor-js/execution"; +import { mcpSessionStubForOwner } from "@executor-js/cloudflare/mcp/session-stub"; +import { buildExecuteDescription, type ResumeResponse } from "@executor-js/execution"; import { loadConfig, type CloudflareConfig, type CloudflareEnv } from "../config"; import { createD1ExecutorDb } from "../db/d1"; @@ -54,6 +60,124 @@ class McpModelResumeForwardError extends Data.TaggedError("McpModelResumeForward readonly cause: unknown; }> {} +class CloudflareModernMcpBuildError extends Data.TaggedError("CloudflareModernMcpBuildError")<{ + readonly cause: unknown; +}> {} + +const makeCloudflareExecutionRuntime = ( + sessionMeta: SessionMeta, + dbHandle: CfSessionDbHandle, + config: CloudflareConfig, +) => + Effect.gen(function* () { + yield* Effect.promise(() => preloadQuickJs()); + const { engine, executor } = yield* makeExecutionStack( + sessionMeta.userId, + sessionMeta.organizationId, + sessionMeta.organizationName, + { mcpResource: sessionMeta.resource }, + ).pipe(Effect.provide(makeCloudflareExecutionStackLayer(config, dbHandle))); + const description = yield* buildExecuteDescription(executor); + return { engine, executor, description }; + }); + +type CloudflareExecutionRuntime = Effect.Success>; + +type CloudflareModernLifecycle = { + readonly pausedExecutionHooks?: PausedExecutionHooks; + readonly resumeFallback?: ( + executionId: string, + response: ResumeResponse, + ) => Effect.Effect; +}; + +const makeCloudflareModernRuntime = ( + sessionMeta: SessionMeta, + runtime: CloudflareExecutionRuntime, + loadAppShellHtml: () => Promise, + config: CloudflareConfig, + lifecycle: CloudflareModernLifecycle = {}, +): BuiltModernMcpRuntime => { + const artifactOrigin = sessionMeta.webOrigin ?? config.webBaseUrl; + return { + engine: runtime.engine, + buildServer: (options) => + buildMcpServerV2({ + engine: runtime.engine, + description: runtime.description, + artifacts: runtime.executor.artifacts, + connections: runtime.executor.connections, + artifactsEnabled: sessionMeta.artifactsEnabled ?? true, + loadAppShellHtml, + smokeRenderArtifact, + ...(artifactOrigin + ? { artifactUrl: artifactUrlFor(artifactOrigin, sessionMeta.organizationSlug) } + : {}), + elicitationMode: { mode: "native" }, + ...(lifecycle.pausedExecutionHooks + ? { + pausedExecutionHooks: lifecycle.pausedExecutionHooks, + pausedExecutionLeaseMs: PAUSED_APPROVAL_TIMEOUT_MS, + } + : {}), + ...(lifecycle.resumeFallback ? { resumeFallback: lifecycle.resumeFallback } : {}), + ...options, + }), + }; +}; + +const closeModernServerWithDb = Promise }>( + server: Server, + dbHandle: CfSessionDbHandle, +): Server => { + const closeServer = server.close.bind(server); + server.close = () => + Effect.runPromise( + Effect.promise(closeServer).pipe(Effect.ensuring(Effect.promise(() => dbHandle.end()))), + ); + return server; +}; + +/** Build the worker-side SDK v2 server over a fresh D1 execution runtime. */ +export const makeCloudflareModernMcpServerBuilder = ( + env: CloudflareEnv, + config: CloudflareConfig, + session: McpSessionInit, +): McpModernServerBuilder["Service"] => ({ + build: (principal: Principal, options) => + Effect.promise(async () => { + const handle = await createD1ExecutorDb(env.DB, env.BLOBS); + return { ...handle, end: () => handle.close() } satisfies CfSessionDbHandle; + }).pipe( + Effect.flatMap((dbHandle) => { + const { resource, ...requestOptions } = options; + const sessionMeta: SessionMeta = { + organizationId: principal.organizationId, + organizationName: config.organizationName, + organizationSlug: config.organizationSlug, + userId: principal.accountId, + resource, + elicitationMode: session.elicitationMode, + artifactsEnabled: session.artifactsEnabled, + webOrigin: session.webOrigin, + }; + return makeCloudflareExecutionRuntime(sessionMeta, dbHandle, config).pipe( + Effect.flatMap((runtime) => + makeCloudflareModernRuntime( + sessionMeta, + runtime, + makeAssetsShellHtmlLoader({ assets: env.ASSETS }), + config, + ).buildServer(requestOptions), + ), + Effect.map((server) => closeModernServerWithDb(server, dbHandle)), + Effect.tapCause(() => Effect.promise(() => dbHandle.end())), + Effect.mapError((cause) => new CloudflareModernMcpBuildError({ cause })), + ); + }), + ), +}); + export class McpSessionDO extends McpAgentSessionDOBase { private readonly cfEnv: CloudflareEnv; private readonly cfConfig: CloudflareConfig; @@ -88,7 +212,7 @@ export class McpSessionDO extends McpAgentSessionDOBase { return Effect.tryPromise({ try: () => - mcpSessionStub(this.cfEnv.MCP_SESSION, owner.sessionId).resumeExecutionForModel( + mcpSessionStubForOwner(this.cfEnv.MCP_SESSION, owner).resumeExecutionForModel( executionId, identity, response, @@ -123,15 +247,18 @@ export class McpSessionDO extends McpAgentSessionDOBase preloadQuickJs()); - const { engine, executor } = yield* makeExecutionStack( - sessionMeta.userId, - sessionMeta.organizationId, - sessionMeta.organizationName, - { mcpResource: sessionMeta.resource }, - ).pipe(Effect.provide(makeCloudflareExecutionStackLayer(config, dbHandle))); + const runtime = yield* makeCloudflareExecutionRuntime(sessionMeta, dbHandle, config); + const { engine, executor, description } = runtime; + const modernRuntime = makeCloudflareModernRuntime( + sessionMeta, + runtime, + self.loadAppShellHtml, + config, + { + pausedExecutionHooks: self.modernPausedExecutionHooks, + resumeFallback: self.modernModelResumeFallback, + }, + ); // Browser elicitation mode (the base owns the approval store + the HTTP // approval RPCs): a gated execution pauses and returns an approvalUrl into // the console resume page. The URL origin is the create request's origin @@ -143,6 +270,7 @@ export class McpSessionDO extends McpAgentSessionDOBase { + const self = this; + return makeCloudflareExecutionRuntime(sessionMeta, dbHandle, this.cfConfig).pipe( + Effect.map((runtime) => + makeCloudflareModernRuntime(sessionMeta, runtime, self.loadAppShellHtml, self.cfConfig, { + pausedExecutionHooks: self.modernPausedExecutionHooks, + resumeFallback: self.modernModelResumeFallback, + }), + ), + Effect.withSpan("McpSessionDO.buildModernMcpRuntime"), + // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: runtime-build failures surface through the base RPC cleanup path + Effect.orDie, + ); + } + + protected override modernRequestStateSigningKey(): string { + return requireMcpRequestStateKey(this.cfEnv.MCP_REQUEST_STATE_KEY); + } } diff --git a/apps/host-cloudflare/wrangler.jsonc b/apps/host-cloudflare/wrangler.jsonc index f97e19507..a2d6bf913 100644 --- a/apps/host-cloudflare/wrangler.jsonc +++ b/apps/host-cloudflare/wrangler.jsonc @@ -67,9 +67,11 @@ ], // Cloudflare Access is the entire auth layer. ACCESS_TEAM_DOMAIN, ACCESS_AUD, // and ADMIN_EMAILS are installation-specific live vars, set after the first - // deploy and preserved by keep_vars. EXECUTOR_SECRET_KEY (the at-rest - // secret-encryption key) is a SECRET, set it with - // `wrangler secret put EXECUTOR_SECRET_KEY`, never in vars. + // deploy and preserved by keep_vars. EXECUTOR_SECRET_KEY (at-rest encryption) + // and MCP_REQUEST_STATE_KEY (MCP 2026-07-28 continuation signing, 32+ bytes) + // are SECRETS, set with `wrangler secret put `, never in vars. The + // MCP key is a deployment prerequisite shared by Worker and session DOs; + // never generate it independently per isolate. "vars": { "ACCESS_NAME_CLAIM": "name", "ACCESS_GROUPS_CLAIM": "groups", diff --git a/bun.lock b/bun.lock index f48fc7e9c..f89cbd072 100644 --- a/bun.lock +++ b/bun.lock @@ -685,6 +685,7 @@ "@executor-js/host-mcp": "workspace:*", "@executor-js/sdk": "workspace:*", "@modelcontextprotocol/sdk": "^1.29.0", + "@modelcontextprotocol/server": "2.0.0", "agents": "^0.17.3", "effect": "catalog:", }, diff --git a/packages/hosts/cloudflare/package.json b/packages/hosts/cloudflare/package.json index be3627759..420c9ad36 100644 --- a/packages/hosts/cloudflare/package.json +++ b/packages/hosts/cloudflare/package.json @@ -23,6 +23,10 @@ "./mcp/session-stub": { "types": "./src/mcp/session-stub.ts", "default": "./src/mcp/session-stub.ts" + }, + "./mcp/modern-request-router": { + "types": "./src/mcp/modern-request-router.ts", + "default": "./src/mcp/modern-request-router.ts" } }, "scripts": { @@ -36,6 +40,7 @@ "@executor-js/host-mcp": "workspace:*", "@executor-js/sdk": "workspace:*", "@modelcontextprotocol/sdk": "^1.29.0", + "@modelcontextprotocol/server": "2.0.0", "agents": "^0.17.3", "effect": "catalog:" }, diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts index 4ccc51965..f8377cf86 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts @@ -1,8 +1,14 @@ -import { Cause, Deferred, Effect, Exit, Option, Schema } from "effect"; +import { Cause, Data, Deferred, Effect, Exit, Option, Schema } from "effect"; import type * as Tracer from "effect/Tracer"; import type { Connection, ConnectionContext } from "agents"; import { McpAgent } from "agents/mcp"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; +import { + createMcpHandler, + type McpHttpHandler, + type McpRequestContext, + type McpServer as ModernMcpServer, +} from "@modelcontextprotocol/server"; import { RequestOrgSlug, RequestWebOrigin } from "@executor-js/api/server"; import { @@ -18,13 +24,28 @@ import { type PausedExecutionHooks, type ResumeFallbackOutcome, } from "@executor-js/host-mcp/tool-server"; -import { defaultMcpResource, type McpResource } from "@executor-js/host-mcp"; +import { + defaultMcpResource, + jsonRpcErrorBody, + mcpResourceKey, + type McpResource, +} from "@executor-js/host-mcp"; +import { + appsEnabledForClientCapabilities, + clientCapabilitiesFromRequestBody, + mcpRequestStatePrincipal, +} from "@executor-js/host-mcp/tool-server-v2"; -import type { IncomingPropagationHeaders, McpElicitationMode } from "./do-headers"; -import type { - McpExecutionOwnerDirectory, - McpExecutionOwnerRecord, - McpExecutionOwnerRoute, +import { + verifiedMcpRequestHeaders, + type IncomingPropagationHeaders, + type McpElicitationMode, +} from "./do-headers"; +import { + modernMcpExecutionOwnerRoute, + type McpExecutionOwnerDirectory, + type McpExecutionOwnerRecord, + type McpExecutionOwnerRoute, } from "./execution-owner-directory"; import { MAX_PAUSED_SESSION_IDLE_MS, @@ -125,6 +146,23 @@ export interface SessionMeta { export interface BuiltMcpServer { readonly mcpServer: McpServer; readonly engine: ExecutionEngine; + /** Modern per-request server factory sharing this legacy runtime's engine. */ + readonly modernRuntime?: BuiltModernMcpRuntime; +} + +/** Request-specific inputs added to a DO-local SDK v2 server. */ +export interface ModernMcpServerRequestOptions { + readonly appsEnabled: boolean; + readonly requestStateSigningKey: Uint8Array | string; + readonly requestStatePrincipal: string; +} + +/** Long-lived DO execution runtime shared by per-request SDK v2 servers. */ +export interface BuiltModernMcpRuntime { + readonly engine: ExecutionEngine; + readonly buildServer: ( + options: ModernMcpServerRequestOptions, + ) => Effect.Effect; } export interface BrowserApprovalStore { @@ -132,8 +170,15 @@ export interface BrowserApprovalStore { readonly waitForResponse: (executionId: string) => Effect.Effect; } +type ModernRuntimeAccess = + | { readonly status: "ok"; readonly runtime: BuiltModernMcpRuntime } + | { readonly status: "forbidden" }; + +class ModernMcpRuntimeNotConfigured extends Data.TaggedError("ModernMcpRuntimeNotConfigured") {} + const SESSION_META_KEY = "session-meta"; const LAST_ACTIVITY_KEY = "last-activity-ms"; +const MODERN_SESSION_KEY = "modern-session"; const PARTYSERVER_NAME_KEY = "__ps_name"; /** The agents SDK's durable "condemned" marker (`_cf_scheduleDestroy`). */ const AGENTS_DESTROY_PENDING_KEY = "cf_agents_destroy_pending"; @@ -229,6 +274,12 @@ export abstract class McpAgentSessionDOBase< private engine: ExecutionEngine | null = null; private dbHandle: TDbHandle | null = null; private sessionMeta: SessionMeta | null = null; + private modernRuntime: BuiltModernMcpRuntime | null = null; + private modernRuntimePromise: Promise | null = null; + private modernHandler: McpHttpHandler | null = null; + private modernRunningRequestCount = 0; + private modernRequestBodies = new WeakMap(); + private modernRequestPropagation = new WeakMap(); private initialized = false; private onStartPromise: Promise | null = null; private lastActivityMs = 0; @@ -245,6 +296,20 @@ export abstract class McpAgentSessionDOBase< dbHandle: TDbHandle, ): Effect.Effect; + /** Build the engine and per-request SDK v2 server factory for a modern-only DO. */ + protected buildModernMcpRuntime( + _sessionMeta: SessionMeta, + _dbHandle: TDbHandle, + ): Effect.Effect { + return Effect.fail(new ModernMcpRuntimeNotConfigured()); + } + + /** Read and validate the deployment-provided modern request-state signing key. */ + protected modernRequestStateSigningKey(): string { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- composition boundary: subclasses serving modern MCP must provide a shared deployment key + throw new Error("Modern MCP request-state signing is not configured"); + } + protected withTelemetry( effect: Effect.Effect, _incoming?: IncomingTraceHeaders, @@ -293,6 +358,16 @@ export abstract class McpAgentSessionDOBase< return { sessionId: this.sessionId }; } + private modernExecutionOwnerRoute(): McpExecutionOwnerRoute { + return this.ctx.id.name + ? this.executionOwnerRoute() + : modernMcpExecutionOwnerRoute(this.ctx.id.toString()); + } + + private runtimeOwnerId(): string { + return this.ctx.id.name ? this.sessionId : this.modernExecutionOwnerRoute().sessionId; + } + protected sameExecutionOwnerRoute(a: McpExecutionOwnerRoute, b: McpExecutionOwnerRoute): boolean { return a.sessionId === b.sessionId; } @@ -320,6 +395,12 @@ export abstract class McpAgentSessionDOBase< ): Effect.Effect => this.resumeFromExecutionOwnerDirectory(executionId, response); + protected readonly modernModelResumeFallback = ( + executionId: string, + response: ResumeResponse, + ): Effect.Effect => + this.resumeFromExecutionOwnerDirectory(executionId, response, this.modernExecutionOwnerRoute()); + protected readonly pausedExecutionHooks: PausedExecutionHooks = { onExecutionPaused: (executionId, deadline) => Effect.sync(() => { @@ -329,6 +410,17 @@ export abstract class McpAgentSessionDOBase< onResumeSettled: (executionId) => this.finishPendingApprovalResume(executionId), }; + /** + * Modern pause hooks await the directory write before the `input_required` + * result leaves the DO, so its signed continuation is immediately routable. + */ + protected readonly modernPausedExecutionHooks: PausedExecutionHooks = { + onExecutionPaused: (executionId, deadline) => + this.startPendingApprovalLease(executionId, deadline, this.modernExecutionOwnerRoute()), + onResumeStarted: (executionId) => this.beginPendingApprovalResume(executionId), + onResumeSettled: (executionId) => this.finishPendingApprovalResume(executionId), + }; + override async onConnect(conn: Connection, context: ConnectionContext): Promise { const requestIds = readActivePostRequestIds(context.request); if (requestIds.length === 0) { @@ -428,7 +520,7 @@ export abstract class McpAgentSessionDOBase< for (const requestIds of rows.values()) { if (Array.isArray(requestIds)) count += requestIds.length; } - return count; + return count + this.modernRunningRequestCount; } private closeActiveStreams(): void { @@ -481,7 +573,7 @@ export abstract class McpAgentSessionDOBase< console.info( JSON.stringify({ event: "mcp_session_idle_runtime_dispose", - sessionId: this.sessionId, + sessionId: this.runtimeOwnerId(), idleMs: input.idleMs, pausedExecutionCount: input.pausedExecutionCount, }), @@ -540,7 +632,7 @@ export abstract class McpAgentSessionDOBase< event: "mcp_execution_owner_directory_error", operation: input.operation, executionId: input.executionId, - sessionId: self.sessionId, + sessionId: self.runtimeOwnerId(), exceptionType: first?.name ?? "Error", exceptionMessage: first?.message ?? "unknown", cause: Cause.pretty(input.cause), @@ -565,7 +657,7 @@ export abstract class McpAgentSessionDOBase< JSON.stringify({ event: "mcp_model_resume_forward_error", executionId: input.executionId, - sessionId: self.sessionId, + sessionId: self.runtimeOwnerId(), ownerSessionId: input.owner.sessionId, exceptionType: first?.name ?? "Error", exceptionMessage: first?.message ?? "unknown", @@ -591,7 +683,7 @@ export abstract class McpAgentSessionDOBase< event: "mcp_model_resume_forward_error", reason: "timeout", executionId: input.executionId, - sessionId: self.sessionId, + sessionId: self.runtimeOwnerId(), ownerSessionId: input.owner.sessionId, timeoutMs: input.timeoutMs, }), @@ -619,6 +711,120 @@ export abstract class McpAgentSessionDOBase< : built; } + private buildModernRuntime(sessionMeta: SessionMeta, dbHandle: TDbHandle) { + const built = sessionMeta.organizationSlug + ? this.buildModernMcpRuntime(sessionMeta, dbHandle).pipe( + Effect.provideService(RequestOrgSlug, { slug: sessionMeta.organizationSlug }), + ) + : this.buildModernMcpRuntime(sessionMeta, dbHandle); + return sessionMeta.webOrigin + ? built.pipe(Effect.provideService(RequestWebOrigin, { origin: sessionMeta.webOrigin })) + : built; + } + + private modernPropsOwnSession(sessionMeta: SessionMeta, props: McpSessionProps): boolean { + return ( + props.session.userId === sessionMeta.userId && + props.session.organizationId === sessionMeta.organizationId && + mcpResourceKey(props.session.resource) === mcpResourceKey(sessionMeta.resource) + ); + } + + private startModernRuntime(props: McpSessionProps): Promise { + if (this.modernRuntimePromise) return this.modernRuntimePromise; + + const self = this; + const program = Effect.gen(function* () { + yield* self.prepareErrorCaptureScope(); + const stored = yield* self.loadSessionMeta(); + if (stored && !self.modernPropsOwnSession(stored, props)) { + return { status: "forbidden" as const }; + } + const sessionMeta = stored ?? (yield* self.resolveAndStoreSessionMeta(props.session)); + if (self.modernRuntime && self.engine) { + yield* Effect.promise(() => self.markActivity()); + return { status: "ok" as const, runtime: self.modernRuntime }; + } + + const dbHandle = self.dbHandle ?? (yield* self.openSessionDbHandle()); + self.dbHandle = dbHandle; + const runtime = yield* self.buildModernRuntime(sessionMeta, dbHandle); + self.modernRuntime = runtime; + self.engine = runtime.engine; + yield* Effect.promise(() => + Promise.all([self.ctx.storage.put(MODERN_SESSION_KEY, true), self.markActivity()]).then( + () => undefined, + ), + ); + return { status: "ok" as const, runtime }; + }).pipe( + Effect.tapCause((cause) => + Effect.gen(function* () { + console.error("[mcp-session] modern runtime init failed:", Cause.pretty(cause)); + yield* self.captureCauseEffect(cause); + yield* self.recordCauseOnSpan(cause); + yield* self.closeRuntime(); + }), + ), + Effect.withSpan("McpSessionDO.startModernRuntime", { + attributes: { "mcp.auth.organization_id": props.session.organizationId }, + }), + (effect) => self.withTelemetry(effect, props.propagation), + // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: Durable Object RPC methods can only reject their Promise + Effect.orDie, + (effect) => self.withSpanFlush(effect), + ); + + const starting = Effect.runPromise(program); + this.modernRuntimePromise = starting; + starting.then( + () => { + if (this.modernRuntimePromise === starting) this.modernRuntimePromise = null; + }, + () => { + if (this.modernRuntimePromise === starting) this.modernRuntimePromise = null; + }, + ); + return starting; + } + + private modernHandlerForRuntime(): McpHttpHandler { + if (this.modernHandler) return this.modernHandler; + const self = this; + this.modernHandler = createMcpHandler( + (context: McpRequestContext) => { + const request = context.requestInfo; + const runtime = self.modernRuntime; + const sessionMeta = self.sessionMeta; + if (!request || !runtime || !sessionMeta || !self.modernRequestBodies.has(request)) { + // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: the third-party factory Promise has no typed failure channel; absent DO request context is an SDK defect + return Effect.runPromise(Effect.die("Modern MCP Durable Object has no request runtime")); + } + const parsedBody = self.modernRequestBodies.get(request); + const propagation = self.modernRequestPropagation.get(request); + const capabilities = clientCapabilitiesFromRequestBody(parsedBody); + return Effect.runPromise( + runtime + .buildServer({ + appsEnabled: appsEnabledForClientCapabilities(capabilities), + requestStateSigningKey: self.modernRequestStateSigningKey(), + requestStatePrincipal: mcpRequestStatePrincipal({ + accountId: sessionMeta.userId, + organizationId: sessionMeta.organizationId, + }), + }) + .pipe( + (effect) => self.withTelemetry(effect, propagation), + // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: the third-party factory Promise can only reject + Effect.orDie, + ), + ); + }, + { legacy: "reject" }, + ); + return this.modernHandler; + } + private closeRuntime(options: { readonly closeStreams?: boolean } = {}): Effect.Effect { const self = this; return Effect.gen(function* () { @@ -631,8 +837,16 @@ export abstract class McpAgentSessionDOBase< delete (self as { server?: McpServer }).server; yield* Effect.promise(() => server.close()).pipe(Effect.ignore); } + if (self.modernHandler) { + const handler = self.modernHandler; + self.modernHandler = null; + yield* Effect.promise(() => handler.close()).pipe(Effect.ignore); + } Reflect.set(self, "_transport", undefined); self.engine = null; + self.modernRuntime = null; + self.modernRequestBodies = new WeakMap(); + self.modernRequestPropagation = new WeakMap(); if (self.dbHandle) { const dbHandle = self.dbHandle; self.dbHandle = null; @@ -709,10 +923,11 @@ export abstract class McpAgentSessionDOBase< yield* self.prepareErrorCaptureScope(); const sessionMeta = yield* self.resolveAndStoreSessionMeta(props.session); const dbHandle = yield* self.openSessionDbHandle(); - const { mcpServer, engine } = yield* self.buildRuntime(sessionMeta, dbHandle); + const { mcpServer, engine, modernRuntime } = yield* self.buildRuntime(sessionMeta, dbHandle); self.dbHandle = dbHandle; self.server = mcpServer; self.engine = engine; + self.modernRuntime = modernRuntime ?? null; self.initialized = true; yield* Effect.promise(() => self.markActivity()).pipe( Effect.withSpan("McpSessionDO.markActivity"), @@ -747,6 +962,50 @@ export abstract class McpAgentSessionDOBase< ); } + /** + * Serve one authenticated modern request without entering the legacy + * `McpAgent` streamable-HTTP transport. + */ + async serveModernMcp( + request: Request, + props: McpSessionProps, + parsedBody: unknown, + ): Promise { + this.modernRequestStateSigningKey(); + const verified = verifiedMcpRequestHeaders(request); + if ( + !verified || + verified.accountId !== props.session.userId || + verified.organizationId !== props.session.organizationId || + verified.resourceKey !== mcpResourceKey(props.session.resource) + ) { + return jsonRpcErrorBody(403, -32003, "Invalid MCP Durable Object identity", { + cors: false, + }); + } + const access = await this.startModernRuntime(props); + const sessionMeta = this.sessionMeta; + if ( + access.status === "forbidden" || + !sessionMeta || + !this.modernPropsOwnSession(sessionMeta, props) + ) { + return jsonRpcErrorBody(403, -32003, "MCP session does not belong to the current bearer", { + cors: false, + }); + } + + this.modernRequestBodies.set(request, parsedBody); + this.modernRequestPropagation.set(request, props.propagation); + this.modernRunningRequestCount += 1; + // oxlint-disable-next-line executor/no-try-catch-or-throw -- adapter boundary: the RPC must decrement its in-memory running lease on both handler resolution and rejection + try { + return await this.modernHandlerForRuntime().fetch(request, { parsedBody }); + } finally { + this.modernRunningRequestCount = Math.max(0, this.modernRunningRequestCount - 1); + } + } + async validateMcpSessionOwner( identity: McpApprovalOwner, ): Promise<"ok" | "not_found" | "forbidden" | "terminated"> { @@ -928,7 +1187,8 @@ export abstract class McpAgentSessionDOBase< } override async alarm(): Promise { - if (!(await this.hasPartyServerName())) { + const isModernSession = (await this.ctx.storage.get(MODERN_SESSION_KEY)) === true; + if (!isModernSession && !(await this.hasPartyServerName())) { await this.cleanupUnaddressableSessionAlarm(); return; } @@ -947,15 +1207,21 @@ export abstract class McpAgentSessionDOBase< }); if (decision.kind === "idle_within_timeout") { + if (isModernSession) { + await this.ctx.storage.setAlarm(Date.now() + Math.max(1, this.sessionTimeoutMs() - idleMs)); + return; + } await super.alarm(); return; } + const ownerId = isModernSession ? this.modernExecutionOwnerRoute().sessionId : this.sessionId; + if (decision.kind === "extend_paused_lease") { console.info( JSON.stringify( pausedLeaseExtensionLog({ - sessionId: this.sessionId, + sessionId: ownerId, pausedExecutionCount, idleMs, leaseMs: decision.leaseMs, @@ -970,7 +1236,7 @@ export abstract class McpAgentSessionDOBase< console.info( JSON.stringify( runningLeaseExtensionLog({ - sessionId: this.sessionId, + sessionId: ownerId, runningExecutionCount, activeStreamCount, idleMs, @@ -1030,6 +1296,7 @@ export abstract class McpAgentSessionDOBase< private writeExecutionOwnerEntry( executionId: string, deadline: PausedExecutionDeadline | undefined, + owner: McpExecutionOwnerRoute = this.executionOwnerRoute(), ): Effect.Effect { const directory = this.executionOwnerDirectory(); if (!directory || !deadline) return Effect.void; @@ -1039,7 +1306,7 @@ export abstract class McpAgentSessionDOBase< if (!sessionMeta) return; const record: McpExecutionOwnerRecord = { executionId, - owner: self.executionOwnerRoute(), + owner, accountId: sessionMeta.userId, organizationId: sessionMeta.organizationId, expiresAt: deadline.expiresAt, @@ -1082,6 +1349,7 @@ export abstract class McpAgentSessionDOBase< private resumeFromExecutionOwnerDirectory( executionId: string, response: ResumeResponse, + currentOwner: McpExecutionOwnerRoute = this.executionOwnerRoute(), ): Effect.Effect { const directory = this.executionOwnerDirectory(); if (!directory) return Effect.succeed(null); @@ -1108,7 +1376,7 @@ export abstract class McpAgentSessionDOBase< return { status: "execution_forbidden" } as const; } - if (self.sameExecutionOwnerRoute(record.owner, self.executionOwnerRoute())) { + if (self.sameExecutionOwnerRoute(record.owner, currentOwner)) { yield* self.deleteExecutionOwnerEntry(executionId); return { status: "execution_expired", ttlMs: record.ttlMs } as const; } @@ -1155,6 +1423,7 @@ export abstract class McpAgentSessionDOBase< private startPendingApprovalLease( executionId: string, deadline: PausedExecutionDeadline | undefined, + owner: McpExecutionOwnerRoute = this.executionOwnerRoute(), ): Effect.Effect { const self = this; return Effect.gen(function* () { @@ -1177,7 +1446,7 @@ export abstract class McpAgentSessionDOBase< self.queuePendingApprovalLeaseExpiration(executionId); }, PAUSED_APPROVAL_TIMEOUT_MS); self.pendingApprovalLeases.set(executionId, { disposeKeepAlive, timeout, expiring: false }); - yield* self.writeExecutionOwnerEntry(executionId, deadline); + yield* self.writeExecutionOwnerEntry(executionId, deadline, owner); }).pipe( Effect.withSpan("McpSessionDO.pending_approval_lease.start", { attributes: { "mcp.execution.id": executionId }, diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-modern.test.ts b/packages/hosts/cloudflare/src/mcp/agent-session-modern.test.ts new file mode 100644 index 000000000..2bb12b13f --- /dev/null +++ b/packages/hosts/cloudflare/src/mcp/agent-session-modern.test.ts @@ -0,0 +1,331 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Cause, Effect } from "effect"; + +import type { + ExecutionEngine, + ExecutionResult, + PausedExecution, + ResumeResponse, +} from "@executor-js/execution"; +import { defaultMcpResource } from "@executor-js/host-mcp"; +import { PAUSED_APPROVAL_TIMEOUT_MS } from "@executor-js/host-mcp/tool-server"; +import { buildMcpServerV2 } from "@executor-js/host-mcp/tool-server-v2"; +import { FormElicitation, ToolAddress } from "@executor-js/sdk"; + +import { + McpAgentSessionDOBase, + type BuiltMcpServer, + type BuiltModernMcpRuntime, + type McpSessionInit, + type McpSessionProps, + type ModernMcpServerRequestOptions, + type SessionMeta, +} from "./agent-session-durable-object"; +import { + modernMcpExecutionOwnerRoute, + type McpExecutionOwnerDirectory, + type McpExecutionOwnerRecord, + type McpExecutionOwnerRoute, +} from "./execution-owner-directory"; + +const REQUEST_STATE_KEY = "0123456789abcdef0123456789abcdef"; +const EXECUTION_ID = "exec-modern-pause"; + +class MemoryStorage { + private readonly values = new Map(); + alarm: number | undefined; + + async get(key: string): Promise { + return this.values.get(key) as T | undefined; + } + + async put(key: string, value: unknown): Promise { + this.values.set(key, value); + } + + async delete(key: string | readonly string[]): Promise { + if (typeof key === "string") { + this.values.delete(key); + return; + } + for (const entry of key) this.values.delete(entry); + } + + async list(options: { readonly prefix?: string } = {}): Promise> { + return new Map( + Array.from(this.values.entries()) + .filter(([key]) => !options.prefix || key.startsWith(options.prefix)) + .map(([key, value]) => [key, value as T]), + ); + } + + async setAlarm(time: number | Date): Promise { + this.alarm = typeof time === "number" ? time : time.getTime(); + } + + async deleteAlarm(): Promise { + this.alarm = undefined; + } +} + +class MemoryContext { + readonly storage = new MemoryStorage(); + readonly id = { + name: undefined, + toString: () => "modern-do-id", + }; + readonly waitUntilPromises: Promise[] = []; + + waitUntil(promise: Promise): void { + this.waitUntilPromises.push(promise); + } +} + +class MemoryDirectory implements McpExecutionOwnerDirectory { + readonly records = new Map(); + + put(record: McpExecutionOwnerRecord): Effect.Effect { + return Effect.sync(() => { + this.records.set(record.executionId, record); + }); + } + + get(executionId: string): Effect.Effect { + return Effect.sync(() => this.records.get(executionId) ?? null); + } + + delete(executionId: string): Effect.Effect { + return Effect.sync(() => { + this.records.delete(executionId); + }); + } +} + +type Harness = { + approvalResponses: Map; + approvalWaiters: Map; + beginPendingApprovalResume: (executionId: string) => Effect.Effect; + buildMcpServer: () => Effect.Effect; + buildModernMcpRuntime: () => Effect.Effect; + ctx: MemoryContext; + dbHandle: { readonly end: () => void } | null; + engine: ExecutionEngine | null; + executionOwnerDirectory: () => McpExecutionOwnerDirectory; + finishPendingApprovalResume: (executionId: string) => Effect.Effect; + initialized: boolean; + keepAlive: () => Promise<() => void>; + lastActivityMs: number; + modernHandler: null; + modernPausedExecutionHooks: { + readonly onExecutionPaused: ( + executionId: string, + deadline: { readonly expiresAt: string; readonly ttlMs: number } | undefined, + ) => Effect.Effect; + readonly onResumeStarted: (executionId: string) => Effect.Effect; + readonly onResumeSettled: (executionId: string) => Effect.Effect; + }; + modernRequestBodies: WeakMap; + modernRequestPropagation: WeakMap; + modernRequestStateSigningKey: () => string; + modernRunningRequestCount: number; + modernRuntime: BuiltModernMcpRuntime | null; + modernRuntimePromise: Promise | null; + onStartPromise: Promise | null; + openSessionDb: () => { readonly end: () => void }; + pendingApprovalLeases: Map; + resolveSessionMeta: (token: McpSessionInit) => Effect.Effect; + serveModernMcp: ( + request: Request, + props: McpSessionProps, + parsedBody: unknown, + ) => Promise; + server?: never; + sessionMeta: SessionMeta | null; + startPendingApprovalLease: ( + executionId: string, + deadline: { readonly expiresAt: string; readonly ttlMs: number } | undefined, + owner: McpExecutionOwnerRoute, + ) => Effect.Effect; +}; + +const makeEngine = (): { + readonly engine: ExecutionEngine; + readonly resumeCalls: ResumeResponse[]; +} => { + const paused = new Map(); + const resumeCalls: ResumeResponse[] = []; + const execution: PausedExecution = { + id: EXECUTION_ID, + elicitationContext: { + address: ToolAddress.make("tools.test.org.main.confirm"), + args: {}, + request: FormElicitation.make({ message: "Confirm?", requestedSchema: {} }), + }, + }; + const engine: ExecutionEngine = { + execute: () => Effect.succeed({ result: "unused" }), + executeWithPause: () => + Effect.sync(() => { + paused.set(execution.id, execution); + return { status: "paused" as const, execution }; + }), + resume: (executionId, response) => + Effect.sync((): ExecutionResult | null => { + if (!paused.delete(executionId)) return null; + resumeCalls.push(response); + return { status: "completed", result: { result: response.content?.approved } }; + }), + isExecutionSettled: () => Effect.succeed(false), + getPausedExecution: (executionId) => Effect.sync(() => paused.get(executionId) ?? null), + pausedExecutionCount: () => Effect.sync(() => paused.size), + hasPausedExecutions: () => Effect.sync(() => paused.size > 0), + getDescription: Effect.succeed("test engine"), + }; + return { engine, resumeCalls }; +}; + +const makeHarness = () => { + const ctx = new MemoryContext(); + const directory = new MemoryDirectory(); + const { engine, resumeCalls } = makeEngine(); + const session = Object.create(McpAgentSessionDOBase.prototype) as Harness; + session.ctx = ctx; + session.engine = null; + session.dbHandle = null; + session.sessionMeta = null; + session.modernRuntime = null; + session.modernRuntimePromise = null; + session.modernHandler = null; + session.modernRunningRequestCount = 0; + session.modernRequestBodies = new WeakMap(); + session.modernRequestPropagation = new WeakMap(); + session.initialized = false; + session.onStartPromise = null; + session.lastActivityMs = 0; + session.approvalResponses = new Map(); + session.approvalWaiters = new Map(); + session.pendingApprovalLeases = new Map(); + session.openSessionDb = () => ({ end: () => undefined }); + session.keepAlive = () => Promise.resolve(() => undefined); + session.executionOwnerDirectory = () => directory; + session.modernRequestStateSigningKey = () => REQUEST_STATE_KEY; + session.resolveSessionMeta = (token) => + Effect.succeed({ + organizationId: token.organizationId, + organizationName: "Test Org", + userId: token.userId, + resource: token.resource, + elicitationMode: token.elicitationMode, + artifactsEnabled: token.artifactsEnabled, + webOrigin: token.webOrigin, + }); + session.buildMcpServer = () => Effect.die("legacy build is not used by this harness"); + session.modernPausedExecutionHooks = { + onExecutionPaused: (executionId, deadline) => + session.startPendingApprovalLease( + executionId, + deadline, + modernMcpExecutionOwnerRoute(ctx.id.toString()), + ), + onResumeStarted: (executionId) => session.beginPendingApprovalResume(executionId), + onResumeSettled: (executionId) => session.finishPendingApprovalResume(executionId), + }; + session.buildModernMcpRuntime = () => + Effect.succeed({ + engine, + buildServer: (options: ModernMcpServerRequestOptions) => + buildMcpServerV2({ + engine, + elicitationMode: { mode: "native" }, + pausedExecutionHooks: session.modernPausedExecutionHooks, + pausedExecutionLeaseMs: PAUSED_APPROVAL_TIMEOUT_MS, + ...options, + }), + }); + return { session, directory, resumeCalls }; +}; + +const requestBody = (input?: { readonly requestState?: string }) => ({ + jsonrpc: "2.0", + id: input?.requestState ? 2 : 1, + method: "tools/call", + params: { + name: "execute", + arguments: { code: "await tools.test.confirm()" }, + ...(input?.requestState + ? { + requestState: input.requestState, + inputResponses: { + elicitation: { action: "accept", content: { approved: true } }, + }, + } + : {}), + _meta: { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientCapabilities": { elicitation: { form: {} } }, + }, + }, +}); + +const requestFor = (body: ReturnType): Request => + new Request("https://executor.test/mcp", { + method: "POST", + headers: { + "content-type": "application/json", + "mcp-protocol-version": "2026-07-28", + "mcp-method": "tools/call", + "mcp-name": "execute", + "x-executor-mcp-account-id": "acct_1", + "x-executor-mcp-organization-id": "org_1", + "x-executor-mcp-resource-key": "default", + }, + body: JSON.stringify(body), + }); + +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +const requestStateFromResponse = (value: unknown): string | null => { + if (!isRecord(value) || !isRecord(value.result)) return null; + return typeof value.result.requestState === "string" ? value.result.requestState : null; +}; + +describe("McpAgentSessionDOBase modern entry", () => { + it("serves a pause, registers modern ownership, and resumes in the same DO", async () => { + const { session, directory, resumeCalls } = makeHarness(); + const props: McpSessionProps = { + session: { + organizationId: "org_1", + userId: "acct_1", + elicitationMode: "native", + resource: defaultMcpResource, + webOrigin: "https://executor.test", + }, + }; + + const firstBody = requestBody(); + const first = await session.serveModernMcp(requestFor(firstBody), props, firstBody); + const firstPayload: unknown = await first.json(); + const requestState = requestStateFromResponse(firstPayload); + + expect(first.status).toBe(200); + expect(requestState).not.toBeNull(); + expect(directory.records.get(EXECUTION_ID)).toMatchObject({ + executionId: EXECUTION_ID, + owner: { sessionId: "modern:modern-do-id" }, + accountId: "acct_1", + organizationId: "org_1", + ttlMs: PAUSED_APPROVAL_TIMEOUT_MS, + }); + if (!requestState) return; + + const secondBody = requestBody({ requestState }); + const second = await session.serveModernMcp(requestFor(secondBody), props, secondBody); + const secondText = JSON.stringify(await second.json()); + + expect(second.status).toBe(200); + expect(secondText).toContain("true"); + expect(resumeCalls).toEqual([{ action: "accept", content: { approved: true } }]); + expect(directory.records.has(EXECUTION_ID)).toBe(false); + }); +}); diff --git a/packages/hosts/cloudflare/src/mcp/do-headers.ts b/packages/hosts/cloudflare/src/mcp/do-headers.ts index 996cfedb5..36e62b2d4 100644 --- a/packages/hosts/cloudflare/src/mcp/do-headers.ts +++ b/packages/hosts/cloudflare/src/mcp/do-headers.ts @@ -25,6 +25,21 @@ export type VerifiedTokenHeaders = { readonly organizationId: string; }; +/** Parsed worker-stamped identity and resource received by a session DO. */ +export type VerifiedMcpRequestHeaders = VerifiedTokenHeaders & { + readonly resourceKey: string; +}; + +/** Parse the complete worker-stamped modern identity header set. */ +export const verifiedMcpRequestHeaders = (request: Request): VerifiedMcpRequestHeaders | null => { + const accountId = request.headers.get(INTERNAL_ACCOUNT_ID_HEADER); + const organizationId = request.headers.get(INTERNAL_ORGANIZATION_ID_HEADER); + const resourceKey = request.headers.get(INTERNAL_RESOURCE_KEY_HEADER); + return accountId && organizationId && resourceKey + ? { accountId, organizationId, resourceKey } + : null; +}; + // Worker and DO run in separate isolates with independent WebSdk tracer // providers. Neither one can see the other's OTEL context, so the DO used // to emit a brand-new root trace on every stub call. Ferry the worker span @@ -89,7 +104,7 @@ export const withVerifiedIdentityHeaders = ( export const withMcpResponseHeaders = (response: Response): Response => { const headers = new Headers(response.headers); headers.set("access-control-allow-origin", "*"); - headers.set("access-control-expose-headers", "mcp-session-id"); + headers.set("access-control-expose-headers", "mcp-session-id, mcp-protocol-version"); return new Response(response.body, { status: response.status, statusText: response.statusText, diff --git a/packages/hosts/cloudflare/src/mcp/execution-owner-directory.ts b/packages/hosts/cloudflare/src/mcp/execution-owner-directory.ts index 54583f2e5..02451f304 100644 --- a/packages/hosts/cloudflare/src/mcp/execution-owner-directory.ts +++ b/packages/hosts/cloudflare/src/mcp/execution-owner-directory.ts @@ -5,6 +5,21 @@ export type McpExecutionOwnerRoute = { readonly sessionId: string; }; +/** Prefix distinguishing a modern unique DO id from a legacy Agent session id. */ +export const MODERN_MCP_EXECUTION_OWNER_PREFIX = "modern:"; + +/** Encode a unique modern session DO id in the existing owner route slot. */ +export const modernMcpExecutionOwnerRoute = (durableObjectId: string): McpExecutionOwnerRoute => ({ + sessionId: `${MODERN_MCP_EXECUTION_OWNER_PREFIX}${durableObjectId}`, +}); + +/** Decode the unique DO id from a modern owner route, or return null for legacy owners. */ +export const modernMcpDurableObjectId = (route: McpExecutionOwnerRoute): string | null => { + if (!route.sessionId.startsWith(MODERN_MCP_EXECUTION_OWNER_PREFIX)) return null; + const id = route.sessionId.slice(MODERN_MCP_EXECUTION_OWNER_PREFIX.length); + return id.length > 0 ? id : null; +}; + export type McpExecutionOwnerRecord = { readonly executionId: string; readonly owner: McpExecutionOwnerRoute; diff --git a/packages/hosts/cloudflare/src/mcp/modern-request-router.test.ts b/packages/hosts/cloudflare/src/mcp/modern-request-router.test.ts new file mode 100644 index 000000000..93f80c592 --- /dev/null +++ b/packages/hosts/cloudflare/src/mcp/modern-request-router.test.ts @@ -0,0 +1,381 @@ +import { describe, expect, it } from "@effect/vitest"; +import { createRequestStateCodec } from "@modelcontextprotocol/server"; +import { Effect } from "effect"; + +import type { ExecutionEngine } from "@executor-js/execution"; +import { + defaultMcpResource, + type McpModernServerBuilder, + type Principal, +} from "@executor-js/host-mcp"; +import { buildMcpServerV2, mcpRequestStatePrincipal } from "@executor-js/host-mcp/tool-server-v2"; + +import type { McpSessionProps } from "./agent-session-durable-object"; +import { + modernMcpExecutionOwnerRoute, + type McpExecutionOwnerDirectory, + type McpExecutionOwnerRecord, +} from "./execution-owner-directory"; +import { + classifyMcpProtocolEra, + makeMcpModernRequestRouter, + mcpCorsPreflightResponse, + requireMcpRequestStateKey, + type McpModernSessionNamespace, + type McpModernSessionStub, +} from "./modern-request-router"; + +const REQUEST_STATE_KEY = "0123456789abcdef0123456789abcdef"; + +const principal: Principal = { + accountId: "acct_1", + organizationId: "org_1", + organizationName: "Org 1", + email: "user@example.test", + name: "Test User", + avatarUrl: null, + roles: [], +}; + +const props: McpSessionProps = { + session: { + organizationId: principal.organizationId, + userId: principal.accountId, + elicitationMode: "native", + resource: defaultMcpResource, + webOrigin: "https://executor.test", + }, +}; + +const engine: ExecutionEngine = { + execute: (code) => Effect.succeed({ result: code }), + executeWithPause: (code) => + Effect.succeed({ status: "completed" as const, result: { result: code } }), + resume: () => Effect.succeed(null), + isExecutionSettled: () => Effect.succeed(false), + getPausedExecution: () => Effect.succeed(null), + pausedExecutionCount: () => Effect.succeed(0), + hasPausedExecutions: () => Effect.succeed(false), + getDescription: Effect.succeed("test engine"), +}; + +const modernBody = (input: { + readonly method: string; + readonly name?: string; + readonly arguments?: Record; + readonly requestState?: string; +}) => ({ + jsonrpc: "2.0", + id: 1, + method: input.method, + params: { + ...(input.name ? { name: input.name } : {}), + ...(input.arguments ? { arguments: input.arguments } : {}), + ...(input.requestState ? { requestState: input.requestState } : {}), + _meta: { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientCapabilities": {}, + }, + }, +}); + +const modernRequest = (body: ReturnType): Request => + new Request("https://executor.test/mcp", { + method: "POST", + headers: { + "content-type": "application/json", + "mcp-protocol-version": "2026-07-28", + "mcp-method": body.method, + ...(typeof body.params.name === "string" ? { "mcp-name": body.params.name } : {}), + }, + body: JSON.stringify(body), + }); + +class MemoryDirectory implements McpExecutionOwnerDirectory { + readonly records = new Map(); + + put(record: McpExecutionOwnerRecord): Effect.Effect { + return Effect.sync(() => { + this.records.set(record.executionId, record); + }); + } + + get(executionId: string): Effect.Effect { + return Effect.sync(() => this.records.get(executionId) ?? null); + } + + delete(executionId: string): Effect.Effect { + return Effect.sync(() => { + this.records.delete(executionId); + }); + } +} + +type ForwardedRequest = { + readonly id: string; + readonly body: unknown; +}; + +class MemorySessions implements McpModernSessionNamespace { + readonly forwarded: ForwardedRequest[] = []; + uniqueIds = 0; + + newUniqueId(): string { + this.uniqueIds += 1; + return `unique-${this.uniqueIds}`; + } + + idFromName(name: string): string { + return `name:${name}`; + } + + idFromString(id: string): string { + return `id:${id}`; + } + + get(id: string): McpModernSessionStub { + return { + serveModernMcp: async (_request, _props, parsedBody) => { + this.forwarded.push({ id, body: parsedBody }); + return new Response(JSON.stringify({ id }), { + headers: { "content-type": "application/json" }, + }); + }, + }; + } +} + +const makeBuilder = (builds: { count: number }): McpModernServerBuilder["Service"] => ({ + build: (_principal, options) => { + builds.count += 1; + const { resource: _resource, ...requestOptions } = options; + return buildMcpServerV2({ + engine, + elicitationMode: { mode: "native" }, + ...requestOptions, + }); + }, +}); + +const dispatch = async (input: { + readonly body: ReturnType; + readonly sessions: MemorySessions; + readonly directory: MemoryDirectory; + readonly builder: McpModernServerBuilder["Service"]; +}) => { + const request = modernRequest(input.body); + return makeMcpModernRequestRouter().fetch({ + request, + parsedBody: input.body, + principal, + resource: defaultMcpResource, + props, + requestStateSigningKey: REQUEST_STATE_KEY, + builder: input.builder, + sessions: input.sessions, + executionOwners: input.directory, + }); +}; + +const mintRequestState = async (executionId: string, ttlSeconds = 60): Promise => { + const binding = `tools/call\u0000${mcpRequestStatePrincipal(principal)}`; + const codec = createRequestStateCodec<{ readonly executionId: string }>({ + key: REQUEST_STATE_KEY, + ttlSeconds, + bind: () => binding, + }); + const encoded: unknown = await Reflect.apply(codec.mint, codec, [{ executionId }, {}]); + return typeof encoded === "string" ? encoded : ""; +}; + +describe("modern Cloudflare MCP worker routing", () => { + it("echoes dynamic preflight headers with a static modern fallback", () => { + const requested = "content-type, authorization, mcp-param-search"; + expect(mcpCorsPreflightResponse(requested).headers.get("access-control-allow-headers")).toBe( + requested, + ); + expect(mcpCorsPreflightResponse().headers.get("access-control-allow-headers")).toContain( + "mcp-method", + ); + }); + + it("fails clearly when the shared modern signing secret is missing or short", () => { + expect(() => requireMcpRequestStateKey(undefined)).toThrow("MCP_REQUEST_STATE_KEY"); + expect(() => requireMcpRequestStateKey("too-short")).toThrow("at least 32 bytes"); + expect(requireMcpRequestStateKey(REQUEST_STATE_KEY)).toBe(REQUEST_STATE_KEY); + }); + + it("keeps the canonical legacy classification on the existing transport branch", async () => { + const legacyBody = { jsonrpc: "2.0", id: 1, method: "tools/list", params: {} }; + const legacy = new Request("https://executor.test/mcp", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(legacyBody), + }); + const modern = modernRequest(modernBody({ method: "tools/list" })); + + await expect(classifyMcpProtocolEra(legacy, legacyBody)).resolves.toBe("legacy"); + await expect( + classifyMcpProtocolEra(modern, modernBody({ method: "tools/list" })), + ).resolves.toBe("modern"); + }); + + it("serves modern non-tools/call methods worker-side without touching a DO", async () => { + const sessions = new MemorySessions(); + const builds = { count: 0 }; + const response = await dispatch({ + body: modernBody({ method: "tools/list" }), + sessions, + directory: new MemoryDirectory(), + builder: makeBuilder(builds), + }); + + expect(response.status).toBe(200); + expect(builds.count).toBe(1); + expect(sessions.uniqueIds).toBe(0); + expect(sessions.forwarded).toEqual([]); + }); + + it("forwards a fresh modern execute call to a new unique DO", async () => { + const sessions = new MemorySessions(); + const builds = { count: 0 }; + const response = await dispatch({ + body: modernBody({ + method: "tools/call", + name: "execute", + arguments: { code: "1 + 1" }, + }), + sessions, + directory: new MemoryDirectory(), + builder: makeBuilder(builds), + }); + + expect(await response.json()).toEqual({ id: "unique-1" }); + expect(builds.count).toBe(0); + expect(sessions.forwarded.map(({ id }) => id)).toEqual(["unique-1"]); + }); + + it("forwards malformed modern tools/call requests to a new unique DO", async () => { + const sessions = new MemorySessions(); + const builds = { count: 0 }; + + await dispatch({ + body: modernBody({ method: "tools/call" }), + sessions, + directory: new MemoryDirectory(), + builder: makeBuilder(builds), + }); + + expect(builds.count).toBe(0); + expect(sessions.forwarded.map(({ id }) => id)).toEqual(["unique-1"]); + }); + + it("verifies continuation state and forwards to its modern owner DO", async () => { + const executionId = "exec-owned"; + const state = await mintRequestState(executionId); + const directory = new MemoryDirectory(); + directory.records.set(executionId, { + executionId, + owner: modernMcpExecutionOwnerRoute("owner-do-id"), + accountId: principal.accountId, + organizationId: principal.organizationId, + expiresAt: new Date(Date.now() + 60_000).toISOString(), + ttlMs: 60_000, + }); + const sessions = new MemorySessions(); + const builds = { count: 0 }; + + await dispatch({ + body: modernBody({ + method: "tools/call", + name: "execute", + arguments: { code: "1 + 1" }, + requestState: state, + }), + sessions, + directory, + builder: makeBuilder(builds), + }); + + expect(builds.count).toBe(0); + expect(sessions.uniqueIds).toBe(0); + expect(sessions.forwarded.map(({ id }) => id)).toEqual(["id:owner-do-id"]); + }); + + it("uses a fresh worker server for an unknown continuation owner", async () => { + const state = await mintRequestState("exec-missing"); + const sessions = new MemorySessions(); + const response = await dispatch({ + body: modernBody({ + method: "tools/call", + name: "execute", + arguments: { code: "1 + 1" }, + requestState: state, + }), + sessions, + directory: new MemoryDirectory(), + builder: makeBuilder({ count: 0 }), + }); + const body = await response.json(); + + expect(body).toMatchObject({ + result: { structuredContent: { status: "execution_not_found" } }, + }); + expect(sessions.uniqueIds).toBe(0); + expect(sessions.forwarded).toEqual([]); + }); + + it("routes modern resume calls to an existing legacy owner when recorded", async () => { + const executionId = "exec-legacy"; + const directory = new MemoryDirectory(); + directory.records.set(executionId, { + executionId, + owner: { sessionId: "legacy-session" }, + accountId: principal.accountId, + organizationId: principal.organizationId, + expiresAt: new Date(Date.now() + 60_000).toISOString(), + ttlMs: 60_000, + }); + const sessions = new MemorySessions(); + + await dispatch({ + body: modernBody({ + method: "tools/call", + name: "resume", + arguments: { executionId, action: "accept" }, + }), + sessions, + directory, + builder: makeBuilder({ count: 0 }), + }); + + expect(sessions.forwarded.map(({ id }) => id)).toEqual(["name:streamable-http:legacy-session"]); + }); + + it("rejects tampered and expired continuation state without touching a DO", async () => { + const valid = await mintRequestState("exec-invalid"); + const middle = Math.floor(valid.length / 2); + const tampered = `${valid.slice(0, middle)}${valid[middle] === "A" ? "B" : "A"}${valid.slice(middle + 1)}`; + const expired = await mintRequestState("exec-expired", -1); + + for (const requestState of [tampered, expired]) { + const sessions = new MemorySessions(); + const response = await dispatch({ + body: modernBody({ + method: "tools/call", + name: "execute", + arguments: { code: "1 + 1" }, + requestState, + }), + sessions, + directory: new MemoryDirectory(), + builder: makeBuilder({ count: 0 }), + }); + + const body = await response.json(); + expect(body).toMatchObject({ error: { code: -32602 } }); + expect(sessions.uniqueIds).toBe(0); + expect(sessions.forwarded).toEqual([]); + } + }); +}); diff --git a/packages/hosts/cloudflare/src/mcp/modern-request-router.ts b/packages/hosts/cloudflare/src/mcp/modern-request-router.ts new file mode 100644 index 000000000..e86fe76b7 --- /dev/null +++ b/packages/hosts/cloudflare/src/mcp/modern-request-router.ts @@ -0,0 +1,250 @@ +import { Effect, Exit, Option, Schema } from "effect"; +import { + createMcpHandler, + isLegacyRequest, + type McpHttpHandler, + type McpRequestContext, +} from "@modelcontextprotocol/server"; + +import { + jsonRpcErrorBody, + mcpResourceKey, + type McpModernServerBuilder, + type McpResource, + type Principal, +} from "@executor-js/host-mcp"; +import { + appsEnabledForClientCapabilities, + clientCapabilitiesFromRequestBody, + mcpRequestStatePrincipal, + verifyNativeRequestState, +} from "@executor-js/host-mcp/tool-server-v2"; + +import type { McpSessionProps } from "./agent-session-durable-object"; +import type { McpExecutionOwnerDirectory } from "./execution-owner-directory"; +import { mcpSessionStubForOwner } from "./session-stub"; + +const MCP_CORS_EXPOSED_HEADERS = "mcp-session-id, mcp-protocol-version, WWW-Authenticate"; +const MCP_CORS_ALLOWED_HEADERS = + "content-type, authorization, mcp-session-id, accept, mcp-protocol-version, mcp-method, mcp-name"; + +const UnknownRecord = Schema.Record(Schema.String, Schema.Unknown); +const ModernToolsCallMethod = Schema.Struct({ method: Schema.Literal("tools/call") }); +const ModernToolCall = Schema.Struct({ + method: Schema.Literal("tools/call"), + params: Schema.Struct({ + name: Schema.String, + arguments: Schema.optional(UnknownRecord), + requestState: Schema.optional(Schema.String), + }), +}); +type ModernToolCall = typeof ModernToolCall.Type; +const decodeModernToolsCallMethod = Schema.decodeUnknownOption(ModernToolsCallMethod); +const decodeModernToolCall = Schema.decodeUnknownOption(ModernToolCall); + +interface ModernRequestInputs { + readonly builder: McpModernServerBuilder["Service"]; + readonly parsedBody: unknown; + readonly principal: Principal; + readonly requestStateSigningKey: string; +} + +/** Durable Object namespace surface required by modern execution routing. */ +export interface McpModernSessionNamespace { + readonly newUniqueId: () => Id; + readonly idFromName: (name: string) => Id; + readonly idFromString: (id: string) => Id; + readonly get: (id: Id) => unknown; +} + +/** Worker-callable modern RPC exposed by the MCP session Durable Object. */ +export interface McpModernSessionStub { + readonly serveModernMcp: ( + request: Request, + props: McpSessionProps, + parsedBody: unknown, + ) => Promise; +} + +/** Inputs needed to dispatch one authenticated modern MCP request. */ +export interface McpModernRequestDispatch { + readonly request: Request; + readonly parsedBody: unknown; + readonly principal: Principal; + readonly resource: McpResource; + readonly props: McpSessionProps; + readonly requestStateSigningKey: string; + readonly builder: McpModernServerBuilder["Service"]; + readonly sessions: McpModernSessionNamespace; + readonly executionOwners: McpExecutionOwnerDirectory | null; +} + +/** Resource-cached worker router for authenticated 2026-07-28 requests. */ +export interface McpModernRequestRouter { + readonly fetch: (input: McpModernRequestDispatch) => Promise; + readonly close: () => Promise; +} + +/** Validate the shared request-state secret at the first modern request boundary. */ +export const requireMcpRequestStateKey = (value: string | undefined): string => { + if (value !== undefined && new TextEncoder().encode(value).byteLength >= 32) return value; + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- composition boundary: modern MCP cannot safely serve or route continuation state without a deployment-provided HMAC key + throw new Error( + "MCP_REQUEST_STATE_KEY must be set to a secret of at least 32 bytes before serving MCP 2026-07-28 requests", + ); +}; + +/** Build the MCP preflight response, echoing dynamic modern header names. */ +export const mcpCorsPreflightResponse = (requestedHeaders?: string | null): Response => + new Response(null, { + status: 204, + headers: { + "access-control-allow-origin": "*", + "access-control-allow-methods": "GET, POST, DELETE, OPTIONS", + "access-control-allow-headers": + requestedHeaders && requestedHeaders.trim() !== "" + ? requestedHeaders + : MCP_CORS_ALLOWED_HEADERS, + "access-control-expose-headers": MCP_CORS_EXPOSED_HEADERS, + }, + }); + +/** Classify an already-parsed request with the SDK's canonical era predicate. */ +export const classifyMcpProtocolEra = ( + request: Request, + parsedBody: unknown, +): Promise<"legacy" | "modern"> => + isLegacyRequest(request, parsedBody).then((legacy) => (legacy ? "legacy" : "modern")); + +const withModernMcpCors = (response: Response): Response => { + const headers = new Headers(response.headers); + headers.set("access-control-allow-origin", "*"); + headers.set("access-control-expose-headers", MCP_CORS_EXPOSED_HEADERS); + return new Response(response.body, { + status: response.status, + statusText: response.statusText, + headers, + }); +}; + +const toModernSessionStub = (stub: unknown): McpModernSessionStub => + // oxlint-disable-next-line executor/no-double-cast -- boundary: Workers generates the RPC surface from the bound Durable Object class, while the portable namespace type exposes unknown. + stub as unknown as McpModernSessionStub; + +const stubForOwner = ( + sessions: McpModernSessionNamespace, + owner: { readonly sessionId: string }, +): McpModernSessionStub => toModernSessionStub(mcpSessionStubForOwner(sessions, owner)); + +const freshStub = (sessions: McpModernSessionNamespace): McpModernSessionStub => + toModernSessionStub(sessions.get(sessions.newUniqueId())); + +const resumeExecutionId = (call: ModernToolCall): string | null => { + if (call.params.name !== "resume") return null; + const executionId = call.params.arguments?.executionId; + return typeof executionId === "string" && executionId.length > 0 ? executionId : null; +}; + +/** Build the shared worker-side modern handler and DO-affinity router. */ +export const makeMcpModernRequestRouter = (): McpModernRequestRouter => { + const handlers = new Map(); + const requestInputs = new WeakMap(); + + const handlerFor = (resource: McpResource): McpHttpHandler => { + const resourceKey = mcpResourceKey(resource); + const cached = handlers.get(resourceKey); + if (cached) return cached; + + const handler = createMcpHandler( + (context: McpRequestContext) => { + const request = context.requestInfo; + const inputs = request ? requestInputs.get(request) : undefined; + if (!request || !inputs) { + // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: the third-party factory Promise has no typed failure channel; absent request context is an SDK defect + return Effect.runPromise(Effect.die("Modern MCP request has no authenticated context")); + } + const capabilities = clientCapabilitiesFromRequestBody(inputs.parsedBody); + return Effect.runPromise( + inputs.builder.build(inputs.principal, { + resource, + appsEnabled: appsEnabledForClientCapabilities(capabilities), + requestStateSigningKey: inputs.requestStateSigningKey, + requestStatePrincipal: mcpRequestStatePrincipal(inputs.principal), + }), + ); + }, + { legacy: "reject" }, + ); + handlers.set(resourceKey, handler); + return handler; + }; + + const serveWorker = async (input: McpModernRequestDispatch): Promise => { + requestInputs.set(input.request, { + builder: input.builder, + parsedBody: input.parsedBody, + principal: input.principal, + requestStateSigningKey: input.requestStateSigningKey, + }); + return handlerFor(input.resource).fetch(input.request, { parsedBody: input.parsedBody }); + }; + + const serveDo = ( + stub: McpModernSessionStub, + input: McpModernRequestDispatch, + ): Promise => stub.serveModernMcp(input.request, input.props, input.parsedBody); + + return { + fetch: async (input) => { + if (Option.isNone(decodeModernToolsCallMethod(input.parsedBody))) { + return withModernMcpCors(await serveWorker(input)); + } + + const decoded = decodeModernToolCall(input.parsedBody); + if (Option.isNone(decoded)) { + return withModernMcpCors(await serveDo(freshStub(input.sessions), input)); + } + + const call = decoded.value; + let executionId = resumeExecutionId(call); + if (call.params.name === "execute" && call.params.requestState !== undefined) { + const verified = await Effect.runPromiseExit( + verifyNativeRequestState({ + state: call.params.requestState, + method: call.method, + requestStateSigningKey: input.requestStateSigningKey, + requestStatePrincipal: mcpRequestStatePrincipal(input.principal), + }), + ); + if (Exit.isFailure(verified)) { + return withModernMcpCors(await serveWorker(input)); + } + executionId = verified.value.executionId; + } + + if (executionId === null) { + return withModernMcpCors(await serveDo(freshStub(input.sessions), input)); + } + + const owner = input.executionOwners + ? await Effect.runPromise(input.executionOwners.get(executionId)) + : null; + if (!owner) { + return withModernMcpCors(await serveWorker(input)); + } + if ( + owner.accountId !== input.principal.accountId || + owner.organizationId !== input.principal.organizationId + ) { + return withModernMcpCors( + jsonRpcErrorBody(403, -32003, "MCP execution does not belong to the current bearer"), + ); + } + return withModernMcpCors(await serveDo(stubForOwner(input.sessions, owner.owner), input)); + }, + close: () => + Promise.all(Array.from(handlers.values(), (handler) => handler.close())).then( + () => undefined, + ), + }; +}; diff --git a/packages/hosts/cloudflare/src/mcp/session-stub.ts b/packages/hosts/cloudflare/src/mcp/session-stub.ts index 3a003ff0c..753420e11 100644 --- a/packages/hosts/cloudflare/src/mcp/session-stub.ts +++ b/packages/hosts/cloudflare/src/mcp/session-stub.ts @@ -7,13 +7,22 @@ import type { McpSessionModelResumeResult, McpSessionResumeApprovalResult, } from "./agent-session-durable-object"; -import { mcpSessionDurableObjectName } from "./execution-owner-directory"; +import { + modernMcpDurableObjectId, + mcpSessionDurableObjectName, + type McpExecutionOwnerRoute, +} from "./execution-owner-directory"; export interface McpSessionNamespace { readonly idFromName: (name: string) => Id; readonly get: (id: Id) => unknown; } +/** Session namespace surface that can address both named legacy and unique modern DOs. */ +export interface McpOwnerSessionNamespace extends McpSessionNamespace { + readonly idFromString: (id: string) => Id; +} + export interface McpSessionStub { readonly validateMcpSessionOwner: ( identity: McpApprovalOwner, @@ -46,3 +55,16 @@ export const mcpSessionStub = ( namespace.get( namespace.idFromName(mcpSessionDurableObjectName(sessionId)), ) as unknown as McpSessionStub; + +/** Resolve an execution owner route to its legacy named or modern unique DO. */ +export const mcpSessionStubForOwner = ( + namespace: McpOwnerSessionNamespace, + owner: McpExecutionOwnerRoute, +): McpSessionStub => { + const modernId = modernMcpDurableObjectId(owner); + const id = modernId + ? namespace.idFromString(modernId) + : namespace.idFromName(mcpSessionDurableObjectName(owner.sessionId)); + // oxlint-disable-next-line executor/no-double-cast -- boundary: Workers generates this RPC surface from the bound DO class. + return namespace.get(id) as unknown as McpSessionStub; +}; diff --git a/packages/hosts/mcp/src/envelope.ts b/packages/hosts/mcp/src/envelope.ts index 3b04823fb..3018a3c8f 100644 --- a/packages/hosts/mcp/src/envelope.ts +++ b/packages/hosts/mcp/src/envelope.ts @@ -22,6 +22,7 @@ import { import { appsEnabledForClientCapabilities, clientCapabilitiesFromRequest, + mcpRequestStatePrincipal, requestBodyFromRequest, } from "./tool-server-v2"; @@ -263,9 +264,6 @@ interface ModernMcpRouter { ) => Promise; } -const requestStatePrincipal = (principal: Principal): string => - `${principal.accountId}\u0000${principal.organizationId}`; - /** Build the resource-keyed, process-lifetime modern handler cache. */ const makeModernMcpRouter = (): ModernMcpRouter => { const handlers = new Map(); @@ -295,7 +293,7 @@ const makeModernMcpRouter = (): ModernMcpRouter => { resource, appsEnabled: appsEnabledForClientCapabilities(clientCapabilities), requestStateSigningKey: getSigningKey(), - requestStatePrincipal: requestStatePrincipal(inputs.principal), + requestStatePrincipal: mcpRequestStatePrincipal(inputs.principal), }); }), ); diff --git a/packages/hosts/mcp/src/tool-server-v2.ts b/packages/hosts/mcp/src/tool-server-v2.ts index ef8984f39..0f3919a54 100644 --- a/packages/hosts/mcp/src/tool-server-v2.ts +++ b/packages/hosts/mcp/src/tool-server-v2.ts @@ -8,7 +8,7 @@ * {@link appsEnabledForClientCapabilities} decision. Legacy routing remains a * separate host path. */ -import { Effect, Match, Option, Schema } from "effect"; +import { Data, Effect, Match, Option, Schema } from "effect"; import * as Cause from "effect/Cause"; import { acceptedContent, @@ -48,9 +48,10 @@ import { const NATIVE_ELICITATION_RESPONSE_KEY = "elicitation"; -const NativeRequestState = Schema.Struct({ executionId: Schema.String }); -type NativeRequestState = typeof NativeRequestState.Type; -const decodeNativeRequestState = Schema.decodeUnknownOption(NativeRequestState); +const NativeRequestStateSchema = Schema.Struct({ executionId: Schema.String }); +/** Verified payload carried by a modern native-elicitation continuation. */ +export type NativeRequestState = typeof NativeRequestStateSchema.Type; +const decodeNativeRequestState = Schema.decodeUnknownOption(NativeRequestStateSchema); type V2RequestContext = McpRequestJoinKeys & { readonly serverContext: ServerContext; @@ -80,10 +81,53 @@ export const appsEnabledForClientCapabilities = ( clientCapabilities: McpAppsClientCapabilities | null | undefined, ): boolean => Boolean(getUiCapability(clientCapabilities)?.mimeTypes?.includes(RESOURCE_MIME_TYPE)); +/** Bind modern continuation state to the ownership identity used by MCP hosts. */ +export const mcpRequestStatePrincipal = (principal: { + readonly accountId: string; + readonly organizationId: string; +}): string => `${principal.accountId}\u0000${principal.organizationId}`; + +const requestStateBinding = (method: string, principal: string): string => + `${method}\u0000${principal}`; + +/** Route-level failure verifying untrusted modern continuation state. */ +export class McpRequestStateVerificationError extends Data.TaggedError( + "McpRequestStateVerificationError", +)<{ readonly cause: unknown }> {} + +/** + * Verify and parse a modern continuation before a stateless worker uses its + * execution id for Durable Object routing. + */ +export const verifyNativeRequestState = (input: { + readonly state: string; + readonly method: string; + readonly requestStateSigningKey: Uint8Array | string; + readonly requestStatePrincipal: string; +}): Effect.Effect => { + const codec = createRequestStateCodec({ + key: input.requestStateSigningKey, + bind: () => requestStateBinding(input.method, input.requestStatePrincipal), + }); + return Effect.tryPromise({ + // The route-level verifier has no handler context. Its codec binding is a + // closed value derived from the already-parsed method and principal, so the + // SDK callback never observes this inert placeholder. + try: async () => { + const decoded: unknown = await Reflect.apply(codec.verify, codec, [input.state, null]); + return Effect.runPromise(Schema.decodeUnknownEffect(NativeRequestStateSchema)(decoded)); + }, + catch: (cause) => new McpRequestStateVerificationError({ cause }), + }); +}; + const isRecord = (value: unknown): value is Record => typeof value === "object" && value !== null && !Array.isArray(value); -const clientCapabilitiesFromUnknown = (body: unknown): McpAppsClientCapabilities | null => { +/** Parse the MCP Apps capability subset from an already-decoded modern body. */ +export const clientCapabilitiesFromRequestBody = ( + body: unknown, +): McpAppsClientCapabilities | null => { if (!isRecord(body)) return null; const params = body.params; if (!isRecord(params)) return null; @@ -122,7 +166,7 @@ export const requestBodyFromRequest = (request: Request): Effect.Effect export const clientCapabilitiesFromRequest = ( request: Request, ): Effect.Effect => - requestBodyFromRequest(request).pipe(Effect.map(clientCapabilitiesFromUnknown)); + requestBodyFromRequest(request).pipe(Effect.map(clientCapabilitiesFromRequestBody)); const requestJoinKeys = (context: ServerContext): V2RequestContext => ({ requestId: context.mcpReq.id, @@ -208,11 +252,11 @@ const createV2Assembly = ( ...(config.requestStateTtlSeconds === undefined ? {} : { ttlSeconds: config.requestStateTtlSeconds }), - bind: (context) => `${context.mcpReq.method}\u0000${config.requestStatePrincipal}`, + bind: (context) => requestStateBinding(context.mcpReq.method, config.requestStatePrincipal), }); const verifyRequestState = async (state: string, context: ServerContext) => { const decoded = await requestStateCodec.verify(state, context); - return Effect.runPromise(Schema.decodeUnknownEffect(NativeRequestState)(decoded)); + return Effect.runPromise(Schema.decodeUnknownEffect(NativeRequestStateSchema)(decoded)); }; const server = new McpServer( { name: "executor", version: "1.0.0" },