diff --git a/apps/host-cloudflare/README.md b/apps/host-cloudflare/README.md index 1ac16f32d6..f7af790611 100644 --- a/apps/host-cloudflare/README.md +++ b/apps/host-cloudflare/README.md @@ -53,6 +53,15 @@ followed. A host naming a missing binding fails config load: the Worker answers 503 with a message naming the host and the binding. Both the Worker and `McpSessionDO` read the same config, so both paths route. +`INTERNAL_MCP_REGISTRIES` declares what each internal host serves, as a JSON +array of `{ "host", "servers": [{ "slug", "name", "description", "route" }] }` +with `route` equal to `/mcp/` and every `host` present in +`INTERNAL_MCP_HOSTS` (otherwise config load fails). Once per isolate, each +declared server whose slug does not exist yet is created as an org-level MCP +integration at `https://` (no auth) with an org connection named +`main`, and its tools are loaded. An existing integration with the same slug is +never changed, and nothing is ever deleted. + **Trust boundary.** The boundary is the Cloudflare account, not Executor: any Worker in the same account could bind the target Worker, so the bound Worker must be private (no public route, `workers_dev` and preview URLs off) and diff --git a/apps/host-cloudflare/src/account/account-provider.test.ts b/apps/host-cloudflare/src/account/account-provider.test.ts index 973344127f..7f98c131a9 100644 --- a/apps/host-cloudflare/src/account/account-provider.test.ts +++ b/apps/host-cloudflare/src/account/account-provider.test.ts @@ -16,6 +16,7 @@ const config: CloudflareConfig = { secretKey: "x".repeat(32), allowLocalNetwork: false, internalHosts: {}, + internalIntegrations: [], webBaseUrl: "https://localhost", enableDevAuth: false, }; diff --git a/apps/host-cloudflare/src/app.ts b/apps/host-cloudflare/src/app.ts index b5f3b00ade..a523a408d9 100644 --- a/apps/host-cloudflare/src/app.ts +++ b/apps/host-cloudflare/src/app.ts @@ -1,10 +1,18 @@ import { Effect } from "effect"; import { HttpEffect, HttpRouter } from "effect/unstable/http"; -import { dbProviderLayer, ExecutorApp, textFailureStrategy } from "@executor-js/api/server"; +import { + dbProviderLayer, + ExecutorApp, + type ExecutorDbHandle, + makeScopedExecutor, + textFailureStrategy, +} from "@executor-js/api/server"; +import { internalPrincipal } from "./auth/cloudflare-access"; +import { reconcileInternalIntegrations } from "./internal-integrations/reconcile"; import { loadConfig, type CloudflareConfig, type CloudflareEnv } from "./config"; -import { makeCloudflarePlugins } from "./plugins"; +import { makeCloudflarePlugins, type CloudflarePlugins } from "./plugins"; import { createD1ExecutorDb } from "./db/d1"; import { cloudflareAccessIdentityLayer } from "./auth/cloudflare-access"; import { @@ -33,6 +41,41 @@ import { preloadQuickJs } from "./quickjs"; // so the providers close over it instead of reading process.env. // =========================================================================== +let internalIntegrationsReconciled: Promise | null = null; + +// Once per isolate, by a system executor that may write org rows. Never fails the boot. +const reconcileOncePerIsolate = ( + config: CloudflareConfig, + dbHandle: ExecutorDbHandle, +): Promise => { + if (config.internalIntegrations.length === 0) return Promise.resolve(); + internalIntegrationsReconciled ??= Effect.runPromise( + Effect.gen(function* () { + const accountId = internalPrincipal(config, { subject: null, email: null })?.accountId; + if (accountId === undefined) return; + const executor = yield* makeScopedExecutor( + accountId, + config.organizationId, + config.organizationName, + { orgWrites: "allowed" }, + ); + yield* reconcileInternalIntegrations(executor, config.internalIntegrations).pipe( + Effect.ensuring(executor.close().pipe(Effect.ignore)), + ); + }).pipe( + Effect.provide(dbProviderLayer(Effect.succeed(dbHandle))), + Effect.provide(makeCloudflarePluginsProvider(config)), + Effect.provide(makeCloudflareHostConfig(config)), + Effect.catchCause((cause) => + Effect.logWarning("internal integration reconcile aborted", cause).pipe( + Effect.annotateLogs({ step: "reconcile-internal-integrations" }), + ), + ), + ), + ); + return internalIntegrationsReconciled; +}; + export const makeCloudflareApp = async ( env: CloudflareEnv, config: CloudflareConfig = loadConfig(env), @@ -46,6 +89,7 @@ export const makeCloudflareApp = async ( // Open and idempotently bring up the D1 schema once. This is the long-lived // handle the per-request scoped executor reads through the DbProvider seam. const dbHandle = await createD1ExecutorDb(env.DB, env.BLOBS); + await reconcileOncePerIsolate(config, dbHandle); const identityLayer = cloudflareAccessIdentityLayer(config); const mcpAgentHandler = makeCloudflareMcpAgentHandler(config); const approvalHandler = makeCloudflareApprovalHandler(config, env); diff --git a/apps/host-cloudflare/src/auth/cloudflare-access.test.ts b/apps/host-cloudflare/src/auth/cloudflare-access.test.ts index c1d9ad99f0..cf1952f114 100644 --- a/apps/host-cloudflare/src/auth/cloudflare-access.test.ts +++ b/apps/host-cloudflare/src/auth/cloudflare-access.test.ts @@ -24,6 +24,7 @@ const config: CloudflareConfig = { secretKey: "x".repeat(32), allowLocalNetwork: false, internalHosts: {}, + internalIntegrations: [], webBaseUrl: "https://localhost", enableDevAuth: false, }; diff --git a/apps/host-cloudflare/src/config.test.ts b/apps/host-cloudflare/src/config.test.ts index ddf00e15f7..d6a70789ef 100644 --- a/apps/host-cloudflare/src/config.test.ts +++ b/apps/host-cloudflare/src/config.test.ts @@ -156,3 +156,57 @@ describe("INTERNAL_MCP_HOSTS", () => { ); }); }); + +describe("INTERNAL_MCP_REGISTRIES", () => { + const TOOLS = { fetch: async () => new Response("ok") }; + const REGISTRIES = + '[{"host":"tools.internal","servers":[{"slug":"websearch","name":"Web Search","description":"Search the web","route":"/mcp/websearch"}]}]'; + const devEnv = (overrides: Record): ConfigEnv => + Object.assign(makeEnv({ ENABLE_DEV_AUTH: "true" }), overrides); + const withHosts = (registries: string) => + devEnv({ + INTERNAL_MCP_HOSTS: "tools.internal=TOOLS", + INTERNAL_MCP_REGISTRIES: registries, + TOOLS, + }); + + it("derives one integration per declared server", () => { + expect(loadConfig(withHosts(REGISTRIES)).internalIntegrations).toEqual([ + { + slug: "websearch", + name: "Web Search", + description: "Search the web", + endpoint: "https://tools.internal/mcp/websearch", + }, + ]); + }); + + it("declares nothing when the variable is unset", () => { + expect(loadConfig(devEnv({})).internalIntegrations).toEqual([]); + }); + + it("fails loudly when a registry host has no INTERNAL_MCP_HOSTS entry", () => { + expect(() => loadConfig(devEnv({ INTERNAL_MCP_REGISTRIES: REGISTRIES }))).toThrowError( + "INTERNAL_MCP_REGISTRIES names host tools.internal, but INTERNAL_MCP_HOSTS has no entry for it", + ); + }); + + it("refuses a route that is not /mcp/", () => { + expect(() => + loadConfig(withHosts(REGISTRIES.replace("/mcp/websearch", "/other"))), + ).toThrowError('INTERNAL_MCP_REGISTRIES route "/other" must be /mcp/websearch'); + }); + + it("refuses a duplicate slug", () => { + const server = '{"slug":"a","name":"A","description":"d","route":"/mcp/a"}'; + expect(() => + loadConfig(withHosts(`[{"host":"tools.internal","servers":[${server},${server}]}]`)), + ).toThrowError("INTERNAL_MCP_REGISTRIES slug a is declared more than once"); + }); + + it("refuses a value that is not the registry shape", () => { + expect(() => loadConfig(withHosts('{"host":"tools.internal"}'))).toThrowError( + "INTERNAL_MCP_REGISTRIES must be a JSON array of { host, servers: [{ slug, name, description, route }] } objects", + ); + }); +}); diff --git a/apps/host-cloudflare/src/config.ts b/apps/host-cloudflare/src/config.ts index 1e07874d0f..6cfc752769 100644 --- a/apps/host-cloudflare/src/config.ts +++ b/apps/host-cloudflare/src/config.ts @@ -1,3 +1,4 @@ +import { Option, Schema } from "effect"; import type { D1Database, DurableObjectNamespace, R2Bucket } from "@cloudflare/workers-types"; import type { JevGatewayConfig } from "@executor-js/execution"; @@ -73,6 +74,14 @@ export interface CloudflareEnv { * this Worker (wrangler `services`). */ readonly INTERNAL_MCP_HOSTS?: string; + /** + * JSON array of registries, one per internal host: + * `[{"host":"tools.internal","servers":[{"slug","name","description","route"}]}]`. + * Each server becomes an org-level MCP integration at `https://`, + * created at boot when its slug does not exist yet. Every `host` must be a key + * of `INTERNAL_MCP_HOSTS`. + */ + readonly INTERNAL_MCP_REGISTRIES?: string; /** * Cloudflare AI Gateway, for the Jev classifier (`custom-typesafe`). Every * model call goes through a gateway rather than a provider directly, so the @@ -116,6 +125,8 @@ export interface CloudflareConfig { readonly allowLocalNetwork: boolean; /** Internal MCP hosts mapped to their service bindings (see `INTERNAL_MCP_HOSTS`). */ readonly internalHosts: HostedInternalHosts; + /** Integrations declared by `INTERNAL_MCP_REGISTRIES`, ready to reconcile. */ + readonly internalIntegrations: readonly InternalIntegration[]; /** Explicit web base URL (`VITE_PUBLIC_SITE_URL`). Unset on a Worker with no * static URL — the per-request origin is used instead (see RequestWebOrigin). */ readonly webBaseUrl?: string; @@ -235,6 +246,79 @@ const resolveInternalHosts = (env: CloudflareConfigEnv): HostedInternalHosts => return hosts; }; +export interface InternalIntegration { + readonly slug: string; + readonly name: string; + readonly description: string; + readonly endpoint: string; +} + +const InternalRegistries = Schema.Array( + Schema.Struct({ + host: Schema.String, + servers: Schema.Array( + Schema.Struct({ + slug: Schema.String, + name: Schema.String, + description: Schema.String, + route: Schema.String, + }), + ), + }), +); + +const decodeInternalRegistries = Schema.decodeUnknownOption( + Schema.fromJsonString(InternalRegistries), +); + +const INTERNAL_SLUG = /^[a-z][a-z0-9_]*$/; + +const resolveInternalIntegrations = ( + env: CloudflareConfigEnv, + internalHosts: HostedInternalHosts, +): readonly InternalIntegration[] => { + const raw = (env.INTERNAL_MCP_REGISTRIES ?? "").trim(); + if (raw.length === 0) return []; + const refuse = (message: string): never => { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: a malformed internal registry must fail the boot, not drop integrations silently + throw new Error(`INTERNAL_MCP_REGISTRIES ${message}`); + }; + const registries = decodeInternalRegistries(raw); + if (Option.isNone(registries)) { + return refuse( + "must be a JSON array of { host, servers: [{ slug, name, description, route }] } objects", + ); + } + const seen = new Set(); + const integrations: InternalIntegration[] = []; + for (const registry of registries.value) { + const host = normalizeHostname(registry.host); + if (!Object.hasOwn(internalHosts, host)) { + refuse(`names host ${host}, but INTERNAL_MCP_HOSTS has no entry for it`); + } + for (const server of registry.servers) { + if (!INTERNAL_SLUG.test(server.slug)) { + refuse(`slug ${JSON.stringify(server.slug)} must match ^[a-z][a-z0-9_]*$`); + } + if (server.route !== `/mcp/${server.slug}`) { + refuse(`route ${JSON.stringify(server.route)} must be /mcp/${server.slug}`); + } + if (server.name.trim().length === 0 || server.description.trim().length === 0) { + refuse(`slug ${server.slug} needs a non-empty name and description`); + } + if (seen.has(server.slug)) refuse(`slug ${server.slug} is declared more than once`); + seen.add(server.slug); + integrations.push({ + slug: server.slug, + name: server.name, + description: server.description, + endpoint: `https://${host}${server.route}`, + }); + } + } + return integrations; +}; + export interface LoadConfigOptions { /** Build the trusted config for the service-binding entrypoint (see * {@link CloudflareConfig.trustedInternal}). */ @@ -270,6 +354,7 @@ export const loadConfig = ( }), ); } + const internalHosts = resolveInternalHosts(env); return { accessTeamDomain, accessAud, @@ -282,7 +367,8 @@ export const loadConfig = ( organizationSlug: resolveOrgSlug(env.SELF_HOSTED_ORG_SLUG), secretKey, allowLocalNetwork: env.ALLOW_LOCAL_NETWORK === "true", - internalHosts: resolveInternalHosts(env), + internalHosts, + internalIntegrations: resolveInternalIntegrations(env, internalHosts), // Pinned origin via the shared resolver. A Worker receives no PaaS platform // vars (env: {} — there is nothing to detect), so only the explicit // VITE_PUBLIC_SITE_URL applies; when it's unset we leave webBaseUrl undefined diff --git a/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts b/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts new file mode 100644 index 0000000000..d1f5856d19 --- /dev/null +++ b/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts @@ -0,0 +1,171 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Option, Schema } from "effect"; +import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; + +import { createExecutor, IntegrationSlug } from "@executor-js/sdk"; +import { makeHostedFetch, makeHostedHttpClientLayer } from "@executor-js/sdk/host-internal"; +import { makeTestConfig } from "@executor-js/sdk/testing"; +import { makeAnnotationsMcpServer } from "@executor-js/plugin-mcp/testing"; + +import type { InternalIntegration } from "../config"; +import { makeCloudflarePlugins } from "../plugins"; +import { reconcileInternalIntegrations } from "./reconcile"; + +const decodeJson = Schema.decodeUnknownOption(Schema.fromJsonString(Schema.Unknown)); + +const FIXTURE: InternalIntegration = { + slug: "fixture", + name: "Fixture", + description: "A fixture tool server", + endpoint: "https://tools.internal/mcp/fixture", +}; + +const makeBinding = () => { + const methods: string[] = []; + return { + methods, + fetch: async (request: Request) => { + const body = decodeJson(await request.clone().text()); + if (Option.isSome(body) && typeof body.value === "object" && body.value !== null) { + if ("method" in body.value) { + methods.push(`${new URL(request.url).pathname} ${String(body.value.method)}`); + } + } + const server = makeAnnotationsMcpServer(); + const transport = new WebStandardStreamableHTTPServerTransport({ enableJsonResponse: true }); + await server.connect(transport); + return transport.handleRequest(request); + }, + }; +}; + +const makeExecutor = (binding: ReturnType) => + Effect.acquireRelease( + Effect.gen(function* () { + const internalHosts = { "tools.internal": binding }; + const config = makeTestConfig({ + plugins: makeCloudflarePlugins("test-secret-key-0123456789abcdef"), + }); + const executor = yield* createExecutor({ + ...config, + httpClientLayer: makeHostedHttpClientLayer({ internalHosts }), + fetch: makeHostedFetch({ internalHosts }), + }); + return { executor, testDb: config.testDb }; + }), + ({ executor, testDb }) => + executor + .close() + .pipe( + Effect.ignore, + Effect.andThen(Effect.promise(() => testDb.close()).pipe(Effect.ignore)), + ), + ); + +const toolIds = (tools: ReadonlyArray<{ readonly address: unknown }>) => + tools.map((tool) => String(tool.address)).sort(); + +describe("reconcileInternalIntegrations", () => { + it.live("creates the integration and its org connection main with the tools loaded", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { executor } = yield* makeExecutor(binding); + + const outcome = yield* reconcileInternalIntegrations(executor, [FIXTURE]); + + expect(outcome).toEqual([["fixture", "created"]]); + const connections = yield* executor.connections.list({ + integration: IntegrationSlug.make("fixture"), + }); + expect(connections.map((c) => [c.owner, String(c.name)])).toEqual([["org", "main"]]); + const tools = yield* executor.tools.list({ integration: IntegrationSlug.make("fixture") }); + expect(toolIds(tools)).toEqual([ + "tools.fixture.org.main.delete", + "tools.fixture.org.main.delete_titled", + "tools.fixture.org.main.list", + "tools.fixture.org.main.meta_stamped", + "tools.fixture.org.main.ping", + ]); + }), + ), + ); + + it.live("a second run writes nothing and does not touch the binding", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { executor } = yield* makeExecutor(binding); + yield* reconcileInternalIntegrations(executor, [FIXTURE]); + const callsAfterFirst = binding.methods.length; + + const outcome = yield* reconcileInternalIntegrations(executor, [FIXTURE]); + + expect(outcome).toEqual([["fixture", "skipped"]]); + expect(binding.methods.length).toBe(callsAfterFirst); + const connections = yield* executor.connections.list({ + integration: IntegrationSlug.make("fixture"), + }); + expect(connections).toHaveLength(1); + }), + ), + ); + + it.live("leaves an existing integration with the same slug untouched", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { executor } = yield* makeExecutor(binding); + yield* executor.mcp.addServer({ + transport: "remote", + name: "Admin made", + slug: "fixture", + description: "Admin description", + endpoint: "https://other.example.com/mcp", + }); + + const outcome = yield* reconcileInternalIntegrations(executor, [FIXTURE]); + + expect(outcome).toEqual([["fixture", "skipped"]]); + const integration = yield* executor.integrations.get(IntegrationSlug.make("fixture")); + expect(integration).toMatchObject({ name: "Admin made", description: "Admin description" }); + const connections = yield* executor.connections.list({ + integration: IntegrationSlug.make("fixture"), + }); + expect(connections).toEqual([]); + expect(binding.methods).toEqual([]); + }), + ), + ); + + it.live("refreshes only what it created", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { executor } = yield* makeExecutor(binding); + yield* executor.mcp.addServer({ + transport: "remote", + name: "Existing", + slug: "existing", + endpoint: "https://tools.internal/mcp/existing", + }); + + const outcome = yield* reconcileInternalIntegrations(executor, [ + { ...FIXTURE, slug: "existing", endpoint: "https://tools.internal/mcp/existing" }, + FIXTURE, + ]); + + expect(outcome).toEqual([ + ["existing", "skipped"], + ["fixture", "created"], + ]); + const existingTools = yield* executor.tools.list({ + integration: IntegrationSlug.make("existing"), + }); + expect(existingTools).toEqual([]); + expect(binding.methods.some((call) => call.startsWith("/mcp/existing"))).toBe(false); + expect(binding.methods.some((call) => call === "/mcp/fixture tools/list")).toBe(true); + }), + ), + ); +}); diff --git a/apps/host-cloudflare/src/internal-integrations/reconcile.ts b/apps/host-cloudflare/src/internal-integrations/reconcile.ts new file mode 100644 index 0000000000..64c5178936 --- /dev/null +++ b/apps/host-cloudflare/src/internal-integrations/reconcile.ts @@ -0,0 +1,90 @@ +import { Effect } from "effect"; + +import { + AuthTemplateSlug, + ConnectionName, + IntegrationSlug, + type ConnectionRef, +} from "@executor-js/sdk"; + +import type { InternalIntegration } from "../config"; + +export const INTERNAL_CONNECTION_NAME = "main"; + +export type ReconcileOutcome = "created" | "skipped" | "failed"; + +export interface InternalIntegrationExecutor { + readonly integrations: { + readonly get: (slug: IntegrationSlug) => Effect.Effect; + }; + readonly mcp: { + readonly addServer: (input: { + readonly transport: "remote"; + readonly name: string; + readonly slug: string; + readonly description: string; + readonly endpoint: string; + readonly remoteTransport: "streamable-http"; + readonly auth: { readonly kind: "none" }; + }) => Effect.Effect; + }; + readonly connections: { + readonly create: (input: { + readonly owner: "org"; + readonly name: ConnectionName; + readonly integration: IntegrationSlug; + readonly template: AuthTemplateSlug; + readonly values: Record; + }) => Effect.Effect; + readonly refresh: (ref: ConnectionRef) => Effect.Effect; + }; +} + +const reconcileOne = (executor: InternalIntegrationExecutor, entry: InternalIntegration) => + Effect.gen(function* () { + const slug = IntegrationSlug.make(entry.slug); + const existing = yield* executor.integrations.get(slug); + if (existing !== null && existing !== undefined) { + yield* Effect.logInfo(`internal integration ${entry.slug} already exists, left untouched`); + return "skipped" as const; + } + yield* executor.mcp.addServer({ + transport: "remote", + name: entry.name, + slug: entry.slug, + description: entry.description, + endpoint: entry.endpoint, + remoteTransport: "streamable-http", + auth: { kind: "none" }, + }); + const name = ConnectionName.make(INTERNAL_CONNECTION_NAME); + yield* executor.connections.create({ + owner: "org", + name, + integration: slug, + template: AuthTemplateSlug.make("none"), + values: {}, + }); + yield* executor.connections.refresh({ owner: "org", integration: slug, name }); + return "created" as const; + }).pipe( + Effect.catch((cause) => + Effect.logWarning(`internal integration ${entry.slug} could not be reconciled`, cause).pipe( + Effect.annotateLogs({ step: "reconcile-internal-integration", slug: entry.slug }), + Effect.as("failed" as const), + ), + ), + ); + +/** + * Creates each declared integration that does not exist yet, with its org + * connection `main`, and loads its tools. Never updates or deletes anything, + * and never fails the caller. + */ +export const reconcileInternalIntegrations = ( + executor: InternalIntegrationExecutor, + entries: readonly InternalIntegration[], +): Effect.Effect> => + Effect.forEach(entries, (entry) => + reconcileOne(executor, entry).pipe(Effect.map((outcome) => [entry.slug, outcome] as const)), + );