Code nodes

A code node runs your own Node.js or Python function as a workflow step, inside a fresh, isolated Sandbox. Reach for one when the node catalog doesn't already provide the operation: reshaping JSON, computing statistics, filtering records, or processing FILE, IMAGE, AUDIO, or VIDEO data.

Write the function

For Node, the code string is an ES module with one default-exported function, sync or async:

export default async function (inputs, context) {
  return { result: inputs.value };
}

• inputs — a plain object keyed by your declared input ports.
• Return a plain object keyed by your declared output ports: every required output present, optional ones omissible, extra keys fail the step.
• context.outputDir — the only directory binary outputs may be written to. Scalar-only nodes can use the one-argument form and skip context.
• console.log() and console.error() are diagnostics, never outputs.

The source must be plain JavaScript ESM — no TypeScript syntax, CommonJS require(), or module.exports.

For Python, export a function named main:

def main(inputs, context):
    return {"result": inputs["value"]}

Python accepts main(inputs) and main(inputs, context), sync or async. Binary Artifact producers use context.output_dir. Return a plain dict containing strict JSON values. NaN, infinity, bytes, sets, non-string keys, cycles, and arbitrary objects are rejected.

Choose the immutable environment matching the work:

The runtime reference lists every installed distribution, exact version, import name, and Sandbox restriction. Do not infer that a dependency from your application is available inside a code node.

Define and run a node with the SDK

Write the port contract before the source. defineCodeNode() validates the definition, and WorkflowDraft.toWorkflow() validates the code node together with its call-site inputs and references:

import {
  BlitClient,
  CODE_ENVIRONMENT_VERSION,
  CODE_NODE_API_VERSION,
  CODE_RUNTIME_ENVIRONMENT_VERSIONS,
  createWorkflow,
  defineCodeNode,
} from "@blitflow/sdk";

const summarize = defineCodeNode({
  apiVersion: CODE_NODE_API_VERSION,
  runtime: "node",
  environmentVersion: CODE_ENVIRONMENT_VERSION,
  code: `
    export default function ({ values, title }) {
      const total = values.reduce((sum, value) => sum + value, 0);
      return {
        summary: { title, count: values.length, total },
        label: title + ": " + total,
      };
    }
  `,
  inputs: [
    { name: "values", type: "STRUCTURED" },
    { name: "title", type: "TEXT" },
  ],
  outputs: [
    { name: "summary", type: "STRUCTURED" },
    { name: "label", type: "TEXT" },
  ],
  compute: { timeoutSeconds: 30, vcpus: 1 },
});

const draft = createWorkflow();
const code = draft.addCodeNode(summarize, {
  values: [4, 8, 15, 16, 23, 42],
  title: "Sequence",
});
draft.addOutput("summary", `\${${code.id}.summary}`);
draft.addOutput("label", `\${${code.id}.label}`);

const client = new BlitClient({ apiKey: process.env.BLITFLOW_TOKEN });
const outputs = await client.run(draft.toWorkflow(), {});

WorkflowDraft.toWorkflow() emits workflow version 2. Code nodes are an additive node kind in that contract and can be mixed with catalog nodes. For Python, set runtime: "python" and environmentVersion: CODE_RUNTIME_ENVIRONMENT_VERSIONS.python; the rest of the definition and workflow APIs are unchanged.

New Node definitions use CODE_ENVIRONMENT_VERSION, the promoted Node v2 catalog. Set environmentVersion: 1 only when intentionally preserving the legacy Node v1 environment. Existing definitions retain their recorded environment version.

const pythonSummary = defineCodeNode({
  apiVersion: CODE_NODE_API_VERSION,
  runtime: "python",
  environmentVersion: CODE_RUNTIME_ENVIRONMENT_VERSIONS.python,
  code: `
def main(inputs):
    values = inputs["values"]
    return {"total": float(sum(values))}
  `,
  inputs: [{ name: "values", type: "STRUCTURED" }],
  outputs: [{ name: "total", type: "FLOAT" }],
});

Supported port types

Do not return undefined, bigint, functions, symbols, NaN, infinities, class instances, cyclic objects, getters, sparse arrays, or undeclared properties. Port names must be JavaScript identifiers and cannot be __proto__, prototype, or constructor.

Cookbook

The snippets below are module bodies for the code field. Match their reads and returned keys with the definition's inputs and outputs exactly.

Return scalar and structured outputs

Declare label as TEXT, ratio as FLOAT, and details as STRUCTURED:

export default function ({ name, completed, total }) {
  const ratio = total === 0 ? 0 : completed / total;
  return {
    label: `${name}: ${completed}/${total}`,
    ratio,
    details: { completed, total, done: completed === total },
  };
}

Use simple-statistics

simple-statistics@7.9.3 is available in Node v1 and v2:

import { mean, median, standardDeviation } from "simple-statistics";

export default function ({ values }) {
  return {
    mean: mean(values),
    report: {
      median: median(values),
      standardDeviation: standardDeviation(values),
    },
  };
}

Declare values as STRUCTURED, mean as FLOAT, and report as STRUCTURED.

Use NumPy and Pandas

Choose runtime: "python". NumPy and Pandas are preinstalled in Python v1:

import numpy as np
import pandas as pd

def main(inputs):
    frame = pd.DataFrame(inputs["rows"])
    values = frame[inputs["column"]].to_numpy(dtype=float)
    return {
        "mean": float(np.mean(values)),
        "rows": int(len(frame)),
    }

Declare rows as STRUCTURED, column as TEXT, mean as FLOAT, and the output rows as INT. Convert NumPy scalar values to Python float or int before returning them.

Use async Python

Python main may be async. This is useful for local asynchronous work, but it does not enable network access:

import asyncio

async def main(inputs):
    await asyncio.sleep(0)
    return {"result": inputs["value"]}

Declare value and result with the same compatible scalar or STRUCTURED type.

Create an IMAGE with Pillow

Python IMAGE producers write below context.output_dir and return the same descriptor shape as Node producers:

from pathlib import Path
from PIL import Image

def main(inputs, context):
    path = Path(context.output_dir) / "swatch.png"
    Image.new("RGB", (64, 64), inputs["color"]).save(path, format="PNG")
    return {"image": {"path": str(path), "mimeType": "image/png"}}

Declare color as TEXT and image as IMAGE. The collector verifies the PNG signature and persists a durable Artifact.

Read a FILE

A FILE input is a read-only local descriptor, not a Blob URL:

import { createHash } from "node:crypto";
import { createReadStream } from "node:fs";

export default async function ({ source }) {
  const hash = createHash("sha256");
  for await (const chunk of createReadStream(source.path)) {
    hash.update(chunk);
  }
  return {
    size: source.sizeBytes,
    metadata: {
      mimeType: source.mimeType,
      sha256: hash.digest("hex"),
    },
  };
}

Declare source as FILE, size as INT, and metadata as STRUCTURED. Never persist or return source.path; it is internal to this attempt.

Create a FILE

Write below the supplied output directory and return a descriptor, not bytes:

import { writeFile } from "node:fs/promises";
import { join } from "node:path";

export default async function ({ rows }, { outputDir }) {
  const path = join(outputDir, "report.json");
  await writeFile(path, JSON.stringify(rows, null, 2), "utf8");
  return { report: { path, mimeType: "application/json" } };
}

Declare rows as STRUCTURED and report as FILE. BlitFlow collects the file, uploads it, and replaces the descriptor with a durable file Artifact.

Read and copy an IMAGE

Image inputs use the same read-only descriptor. This example copies supported image bytes without assuming an unavailable image-processing package:

import { copyFile } from "node:fs/promises";
import { join } from "node:path";

export default async function ({ image }, { outputDir }) {
  const path = join(outputDir, "copy.bin");
  await copyFile(image.path, path);
  return { copy: { path, mimeType: image.mimeType } };
}

Declare both image and copy as IMAGE. The collector detects the real MIME from the bytes and requires it to match the returned MIME.

Chain code nodes

Binary inputs in a workflow definition must be references to a compatible output from an earlier node. After wrapping the FILE recipes above as createReport and inspectReport, connect them through the producer's output ref:

const created = draft.addCodeNode(createReport, {
  rows: [{ id: 1, status: "ready" }],
});
const inspected = draft.addCodeNode(inspectReport, {
  source: `\${${created.id}.report}`,
});

draft.addOutput("report", `\${${created.id}.report}`);
draft.addOutput("metadata", `\${${inspected.id}.metadata}`);

The first step's durable Artifact is materialized as a new read-only local descriptor for the second step. User code never receives the private storage capability URL.

See a deterministic invalid-output failure

This source fails when result is declared as INT because the returned value has the wrong type:

export default function () {
  return { result: "not an integer" };
}

Fix the returned value or the declared port. Retrying the same definition will not change a deterministic contract failure.

Binary artifacts

FILE, IMAGE, AUDIO, and VIDEO workflow inputs must reference a compatible output from an earlier node. Descriptor literals, external URLs, data: URIs, base64 strings, and raw buffers are rejected before a Sandbox starts. BlitFlow materializes its own Artifact as a local descriptor:

type CodeArtifactInput = {
  path: string;
  mimeType: string;
  sizeBytes: number;
};

To produce a binary Artifact, write a regular file below context.outputDir (Node) or context.output_dir (Python), then return { path, mimeType? } for the declared port. The final workflow output is a durable Artifact reference with MIME, byte size, and SHA-256 metadata. It is not bytes or base64.

IMAGE accepts PNG, JPEG, WebP, GIF, and AVIF. AUDIO accepts AAC, FLAC, MPEG, OGG, and WAV. VIDEO accepts MP4, M4V, WebM, QuickTime, and Matroska. Every typed media MIME is detected from the bytes and must match a declared MIME. SVG, HTML, XML, JavaScript, and other active content cannot be typed media. FILE defaults to application/octet-stream; a declared FILE MIME must be valid, non-active, and compatible with any detected binary signature.

Each file is limited to 200 MiB. Binary inputs and outputs together are limited to 400 MiB per attempt. Missing files, paths outside outputDir, symlinks, hardlinks, sparse files, special files, undeclared outputs, MIME mismatches, and limit violations fail the step.

Run through API, SDK, or MCP

A code node always executes as part of a workflow. The workflow can be inline or a published workflowRef; its contract is identical on all surfaces.

HTTP API

Send the workflow JSON produced by draft.toWorkflow() in the workflow field, or send a published reference:

curl -N "https://studio.blitflow.com/api/v1/runs" \
  -H "Authorization: Bearer $BLITFLOW_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"workflowRef":"acme/code-report@latest","inputs":{}}'

Keep the SSE connection open through run.completed or run.failed. Preserve the id from run.started if you need to fetch the durable run later.

TypeScript SDK

Use client.run(workflow, inputs) when only terminal outputs matter. Use startRun() to inspect every event, and client.getRun(id) to fetch a known durable record. The complete inline builder example at the top uses run().

MCP

Call runs_create with exactly one of workflow or workflowRef, plus any run inputs. It consumes the event stream and returns terminal outputs. Use runs_get when you already have a run id and need its durable state.

Limits and isolation

Each attempt uses a fresh, non-persistent Sandbox. Outbound network access is denied, no ports are exposed, runtime package installation is unavailable, and the Sandbox stops after the attempt. Infrastructure failures are retried by the durable workflow; deterministic source, input, and output failures are not. Public errors never include captured stdout or stderr.

Troubleshooting

Final checklist

• Define ports before writing code and use their names exactly.
• Read the runtime reference before importing.
• Use exported API and environment constants; do not invent versions.
• Return only JSON-compatible values matching declared scalar port types.
• Read FILE/IMAGE/AUDIO/VIDEO descriptors and write outputs only below context.outputDir (Node) or context.output_dir (Python).
• Return binary descriptors, never bytes, base64, or internal paths.
• Treat stdout and stderr as diagnostics, never results.
• Assume no network, runtime package installation, or persistent disk.
• Run defineCodeNode() and draft.toWorkflow() locally before starting a run.