Add generic Cloudflare Workers AI model node

Lets users run any model from the Cloudflare catalog via a searchable
dialog; the node rebuilds its inputs, outputs, and docs from the model's
published schema. Proxy routes cache the catalog and per-model schemas
via the Workers Cache API, and a per-model pricing catalog maps real CF
costs to Dafthunk usage credits.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bchapuis

apps/api/src/index.ts

@@ -11,6 +11,7 @@ import adminRoutes from "./routes/admin";
 import apiKeyRoutes from "./routes/api-keys";
 import billingRoutes from "./routes/billing";
 import botRoutes from "./routes/bots";
+import cloudflareAiRoutes from "./routes/cloudflare-ai";
 import dashboardRoutes from "./routes/dashboard";
 import databaseRoutes from "./routes/databases";
 import datasetRoutes from "./routes/datasets";
@@ -107,6 +108,9 @@ app.route("/queues", queuePublishRoutes);
 // Replicate schema proxy (JWT-authenticated, not org-scoped)
 app.route("/replicate", replicateRoutes);
 
+// Cloudflare Workers AI schema proxy (JWT-authenticated, not org-scoped)
+app.route("/cloudflare-ai", cloudflareAiRoutes);
+
 // Public routes
 app.route("/forms", formRoutes);
 app.route("/feedback-forms", feedbackFormRoutes);

apps/api/src/routes/cloudflare-ai.ts

@@ -0,0 +1,241 @@
+import {
+  type CloudflareModelOpenApiSchema,
+  mapCloudflareSchema,
+} from "@dafthunk/runtime/utils/cloudflare-schema";
+import type {
+  CloudflareModelInfo,
+  CloudflareModelSchema,
+} from "@dafthunk/types";
+import { Hono } from "hono";
+
+import { jwtMiddleware } from "../auth";
+import type { ApiContext } from "../context";
+import { cachedJson } from "../utils/edge-cache";
+
+const cloudflareAiRoutes = new Hono<ApiContext>();
+
+cloudflareAiRoutes.use("*", jwtMiddleware);
+
+/** Cache TTLs (seconds). Upper bounds — the Workers cache may evict earlier. */
+const MODELS_LIST_TTL = 60 * 60; // 1h — catalog additions are rare
+const MODEL_SCHEMA_TTL = 24 * 60 * 60; // 24h — model schemas almost never change
+
+/**
+ * Per-page size requested from the Cloudflare models/search endpoint. If the
+ * response hits this count, the catalog may be larger than we fetched — we
+ * log a warning so operators know to either raise this value or paginate.
+ */
+const MODELS_LIST_PAGE_SIZE = 200;
+
+/** Synthetic host used as the Cache API key namespace for this route. */
+const CACHE_HOST = "https://cache.dafthunk.internal";
+
+/**
+ * Error thrown when the upstream Cloudflare API returns a non-success status.
+ * Kept as a typed error so we don't cache failed responses and can map back
+ * to the right HTTP status.
+ */
+class CloudflareApiError extends Error {
+  constructor(
+    message: string,
+    public readonly status: number
+  ) {
+    super(message);
+  }
+}
+
+/**
+ * Normalise a Cloudflare model identifier. Accepts the canonical form
+ * `@cf/provider/name` as well as values with the leading `@` url-encoded
+ * (e.g. `%40cf/provider/name`).
+ */
+function normaliseModelId(raw: string): string {
+  const decoded = decodeURIComponent(raw).trim();
+  if (!decoded) return "";
+  return decoded.startsWith("@") ? decoded : `@${decoded}`;
+}
+
+/**
+ * Fetch a Cloudflare API endpoint and throw on failure. Throwing is important
+ * so `cachedJson` doesn't persist error responses.
+ */
+async function callCloudflare<T>(url: string, apiToken: string): Promise<T> {
+  const response = await fetch(url, {
+    headers: {
+      Authorization: `Bearer ${apiToken}`,
+      "Content-Type": "application/json",
+    },
+  });
+
+  if (!response.ok) {
+    throw new CloudflareApiError(
+      `Cloudflare API error: ${response.status} ${response.statusText}`,
+      response.status
+    );
+  }
+
+  const json = (await response.json()) as {
+    success: boolean;
+    errors?: Array<{ message?: string }>;
+    result?: T;
+  };
+
+  if (!json.success || !json.result) {
+    const message = json.errors?.[0]?.message ?? "Unknown error";
+    throw new CloudflareApiError(`Cloudflare API error: ${message}`, 502);
+  }
+
+  return json.result;
+}
+
+/**
+ * Keep only the fields the UI actually consumes. Dropping CF's `properties`,
+ * `tags`, `source`, and other passthrough fields shrinks the payload and
+ * avoids leaking anything we don't deliberately expose.
+ */
+function trimModelInfo(raw: unknown): CloudflareModelInfo | null {
+  if (!raw || typeof raw !== "object") return null;
+  const r = raw as Record<string, unknown>;
+  if (typeof r.id !== "string" || typeof r.name !== "string") return null;
+
+  const task =
+    r.task && typeof r.task === "object"
+      ? (r.task as Record<string, unknown>)
+      : null;
+
+  return {
+    id: r.id,
+    name: r.name,
+    ...(typeof r.description === "string"
+      ? { description: r.description }
+      : {}),
+    ...(task && typeof task.id === "string" && typeof task.name === "string"
+      ? { task: { id: task.id, name: task.name } }
+      : {}),
+  };
+}
+
+/**
+ * Consistent error response mapping Cloudflare failures to HTTP statuses.
+ */
+function handleCloudflareError(
+  c: {
+    json: (body: object, status: 400 | 404 | 500 | 502) => Response;
+  },
+  err: unknown
+) {
+  if (err instanceof CloudflareApiError) {
+    const status: 404 | 502 = err.status === 404 ? 404 : 502;
+    return c.json({ error: err.message }, status);
+  }
+  return c.json(
+    { error: err instanceof Error ? err.message : "Unknown error" },
+    502
+  );
+}
+
+/**
+ * GET /cloudflare-ai/models/schema?model=@cf/...
+ *
+ * Fetches a Cloudflare Workers AI model's JSON schema, maps it to Dafthunk
+ * Parameters, and caches the mapped result in the Workers runtime cache.
+ */
+cloudflareAiRoutes.get("/models/schema", async (c) => {
+  const accountId = c.env.CLOUDFLARE_ACCOUNT_ID;
+  const apiToken = c.env.CLOUDFLARE_API_TOKEN;
+
+  if (!accountId || !apiToken) {
+    return c.json(
+      {
+        error:
+          "CLOUDFLARE_ACCOUNT_ID and CLOUDFLARE_API_TOKEN must be configured",
+      },
+      500
+    );
+  }
+
+  const modelParam = c.req.query("model");
+  if (!modelParam) {
+    return c.json({ error: "model query parameter is required" }, 400);
+  }
+
+  const model = normaliseModelId(modelParam);
+  const cacheKey = `${CACHE_HOST}/cf-ai/models/schema/${encodeURIComponent(model)}`;
+
+  try {
+    const payload = await cachedJson<CloudflareModelSchema>(
+      cacheKey,
+      MODEL_SCHEMA_TTL,
+      c.executionCtx,
+      async () => {
+        const url = new URL(
+          `https://api.cloudflare.com/client/v4/accounts/${accountId}/ai/models/schema`
+        );
+        url.searchParams.set("model", model);
+        const raw = await callCloudflare<CloudflareModelOpenApiSchema>(
+          url.toString(),
+          apiToken
+        );
+        const { inputs, outputs } = mapCloudflareSchema(raw);
+        return { model, inputs, outputs };
+      }
+    );
+    return c.json(payload);
+  } catch (err) {
+    return handleCloudflareError(c, err);
+  }
+});
+
+/**
+ * GET /cloudflare-ai/models
+ *
+ * Returns the trimmed Cloudflare Workers AI catalog. The full catalog is
+ * fetched once per TTL window and filtered client-side by the widget; the
+ * response carries only the fields the frontend actually renders.
+ */
+cloudflareAiRoutes.get("/models", async (c) => {
+  const accountId = c.env.CLOUDFLARE_ACCOUNT_ID;
+  const apiToken = c.env.CLOUDFLARE_API_TOKEN;
+
+  if (!accountId || !apiToken) {
+    return c.json(
+      {
+        error:
+          "CLOUDFLARE_ACCOUNT_ID and CLOUDFLARE_API_TOKEN must be configured",
+      },
+      500
+    );
+  }
+
+  const cacheKey = `${CACHE_HOST}/cf-ai/models/list`;
+
+  try {
+    const payload = await cachedJson<{ models: CloudflareModelInfo[] }>(
+      cacheKey,
+      MODELS_LIST_TTL,
+      c.executionCtx,
+      async () => {
+        const url = new URL(
+          `https://api.cloudflare.com/client/v4/accounts/${accountId}/ai/models/search`
+        );
+        url.searchParams.set("per_page", String(MODELS_LIST_PAGE_SIZE));
+        url.searchParams.set("hide_experimental", "true");
+        const raw = await callCloudflare<unknown[]>(url.toString(), apiToken);
+        if (raw.length >= MODELS_LIST_PAGE_SIZE) {
+          console.warn(
+            `[cloudflare-ai] Catalog returned ${raw.length} entries (page size ${MODELS_LIST_PAGE_SIZE}); the list may be truncated — consider adding pagination.`
+          );
+        }
+        const models = raw
+          .map(trimModelInfo)
+          .filter((m): m is CloudflareModelInfo => m !== null);
+        return { models };
+      }
+    );
+    return c.json(payload);
+  } catch (err) {
+    return handleCloudflareError(c, err);
+  }
+});
+
+export default cloudflareAiRoutes;

apps/api/src/runtime/cloudflare-node-registry.ts

@@ -60,6 +60,7 @@ import { CloudflareBrowserPdfNode } from "@dafthunk/runtime/nodes/browser/cloudf
 import { CloudflareBrowserScrapeNode } from "@dafthunk/runtime/nodes/browser/cloudflare-browser-scrape-node";
 import { CloudflareBrowserScreenshotNode } from "@dafthunk/runtime/nodes/browser/cloudflare-browser-screenshot-node";
 import { CloudflareBrowserSnapshotNode } from "@dafthunk/runtime/nodes/browser/cloudflare-browser-snapshot-node";
+import { CloudflareModelNode } from "@dafthunk/runtime/nodes/cloudflare/cloudflare-model-node";
 import { CsvExtractColumnNode } from "@dafthunk/runtime/nodes/csv/csv-extract-column-node";
 import { CsvFilterRowsNode } from "@dafthunk/runtime/nodes/csv/csv-filter-rows-node";
 import { CsvParseNode } from "@dafthunk/runtime/nodes/csv/csv-parse-node";
@@ -706,6 +707,9 @@ export class CloudflareNodeRegistry extends BaseNodeRegistry<Bindings> {
     // Generic Replicate model node
     this.registerImplementation(ReplicateModelNode);
 
+    // Generic Cloudflare Workers AI model node
+    this.registerImplementation(CloudflareModelNode);
+
     // Video processing nodes (Cloudflare Containers)
     if (this.env.FFMPEG_CONTAINER) {
       this.registerImplementation(AppendVideosNode);

apps/api/src/utils/edge-cache.ts

@@ -0,0 +1,40 @@
+/**
+ * Thin wrapper around the Workers Cache API for proxying upstream read-only
+ * JSON endpoints. On cache miss the fetcher runs, its result is serialised
+ * and stored with an explicit `Cache-Control: max-age=…`, and returned to
+ * the caller. Cache writes happen via `ctx.waitUntil` so responses never
+ * block on cache population.
+ *
+ * Cache keys should be stable synthetic URLs that don't collide with real
+ * routes — conventionally `https://cache.dafthunk.internal/<feature>/<key>`.
+ *
+ * Notes:
+ * - Workers Cache API is per-POP and best-effort; TTLs are upper bounds.
+ *   Acceptable for static catalogs we're willing to serve slightly stale.
+ * - Only JSON-serialisable values are supported. Non-JSON responses belong
+ *   in the raw Response-based Cache API idiom.
+ */
+export async function cachedJson<T>(
+  cacheKey: string,
+  ttlSeconds: number,
+  ctx: ExecutionContext,
+  fetcher: () => Promise<T>
+): Promise<T> {
+  const request = new Request(cacheKey);
+  const cache = caches.default;
+
+  const hit = await cache.match(request);
+  if (hit) {
+    return (await hit.json()) as T;
+  }
+
+  const fresh = await fetcher();
+  const response = new Response(JSON.stringify(fresh), {
+    headers: {
+      "Content-Type": "application/json",
+      "Cache-Control": `public, max-age=${ttlSeconds}`,
+    },
+  });
+  ctx.waitUntil(cache.put(request, response.clone()));
+  return fresh;
+}

apps/app/src/components/workflow/widgets/index.ts

@@ -11,6 +11,7 @@ import { audioRecorderInputWidget } from "./input/audio-recorder-input";
 import { blobInputWidget } from "./input/blob-input";
 import { booleanInputWidget } from "./input/boolean-input";
 import { canvasInputWidget } from "./input/canvas-input";
+import { cloudflareModelInputWidget } from "./input/cloudflare-model-input";
 import { cronInputWidget } from "./input/cron-input";
 import { dateInputWidget } from "./input/date-input";
 import { discordTriggerInputWidget } from "./input/discord-trigger-input";
@@ -87,6 +88,7 @@ const widgets = [
   audioRecorderInputWidget,
   canvasInputWidget,
   replicateModelInputWidget,
+  cloudflareModelInputWidget,
   schemaComposeInputWidget,
   schemaExtractInputWidget,
   createDynamicInputsWidget("string-concat", {