Agents and workflows
that survive.

import { ironflow } from "@ironflow/node";

export const shipOrder = ironflow.createFunction(
  { id: "ship-order", triggers: [{ event: "order.paid" }] },
  async ({ event, step }) => {
    // Waits for the child run. The target function has no event triggers.
    const label = await step.invoke<{ url: string }>(
      "create-label",
      event.data,
      { timeout: "2m" } // longest wait for the child; default 30s
    );

    // Fire and forget: returns the child run ID at once.
    // It is already durable, so do not wrap it in step.run.
    const { runId } = await step.invokeAsync("notify-customer", event.data);

    return { label: label.url, notifyRun: runId };
  }
);

import "github.com/sahina/ironflow-go/ironflow"

var ShipOrder = ironflow.CreateFunction(ironflow.FunctionConfig{
	ID:       "ship-order",
	Triggers: []ironflow.Trigger{{Event: "order.paid"}},
}, func(ctx ironflow.Context) (any, error) {
	// Wait for the child run; the target function has no event triggers.
	label, err := ironflow.Invoke[map[string]any](
		ctx, "create-label", map[string]any{"orderId": "ord_1"},
	)
	if err != nil {
		return nil, err
	}

	// Fire and forget: returns the child run ID at once.
	child, err := ironflow.InvokeAsync(
		ctx, "notify-customer", map[string]any{"orderId": "ord_1"},
	)
	if err != nil {
		return nil, err
	}

	return map[string]any{"label": label, "notifyRun": child.RunID}, nil
})

import { createWorker } from "@ironflow/node/worker";
import { fulfilOrder } from "./functions.js";

const worker = createWorker({
  serverUrl: "http://localhost:9123",
  functions: [fulfilOrder],
  // Most jobs this worker runs at once. The default is 10.
  maxConcurrentJobs: 4,
});

await worker.start(); // blocks until the worker stops

import (
	"context"
	"os/signal"
	"syscall"
)
import "github.com/sahina/ironflow-go/ironflow"

func runWorker() error {
	// Ctrl-C or SIGTERM cancels ctx.
	ctx, stop := signal.NotifyContext(
		context.Background(), syscall.SIGINT, syscall.SIGTERM,
	)
	defer stop()

	worker := ironflow.NewWorker(ironflow.WorkerConfig{
		ServerURL:         "http://localhost:9123",
		Functions:         []ironflow.Function{FulfilOrder},
		// Most jobs this worker runs at once. The default is 10.
		MaxConcurrentJobs: 4,
	})
	return worker.Run(ctx) // blocks; cancelling ctx drains active jobs
}

from functions import fulfil_order
from ironflow.worker import Worker

worker = Worker(
    functions=[fulfil_order],
    server_url="http://localhost:9123",
    # Most jobs this worker runs at once. The default is 10.
    max_concurrent_jobs=4,
)
worker.run()  # blocks; SIGINT or SIGTERM drains, a second signal stops

import { createStreamingWorker } from "@ironflow/node/worker-streaming";
import { fulfilOrder } from "./functions.js";

// Opt-in ConnectRPC streaming worker: same config, same start().
const worker = createStreamingWorker({ functions: [fulfilOrder] });
await worker.start();

import "github.com/sahina/ironflow-go/ironflow"

// Opt-in ConnectRPC streaming worker: same WorkerConfig, same Run(ctx).
func streaming(cfg ironflow.WorkerConfig) *ironflow.StreamingWorker {
	return ironflow.NewStreamingWorker(cfg)
}

from functions import fulfil_order
from ironflow.worker import StreamingWorker  # opt-in ConnectRPC stream

StreamingWorker(functions=[fulfil_order]).run()  # same arguments as Worker

import { serve } from "@ironflow/node/serve";
import { fulfilOrder } from "./functions.js";

// Next.js App Router route handler.
// Express: app.post("/api/ironflow", handler).
export const POST = serve({
  functions: [fulfilOrder],
  // Your code reads the key. With no key, no request is verified.
  signingKey: process.env.IRONFLOW_SIGNING_KEY,
});

# app.py - run with: uvicorn app:app
from functions import fulfil_order
from ironflow.serve import serve

app = serve([fulfil_order])  # signing key comes from IRONFLOW_SIGNING_KEY

import (
	"net/http"
	"os"
)
import "github.com/sahina/ironflow-go/ironflow"

func main() {
	// The SDK does not read the env var, so pass the key here. With an
	// empty key, the handler accepts unsigned requests.
	handler := ironflow.Serve(ironflow.ServeConfig{
		Functions:  []ironflow.Function{FulfilOrder},
		SigningKey: os.Getenv("IRONFLOW_SIGNING_KEY"),
	})

	// Mount the handler at the path of the endpoint URL.
	http.Handle("/api/ironflow", handler)
	http.ListenAndServe(":3000", nil)
}

import { createClient } from "@ironflow/node";

const ironflow = createClient({ apiKey: process.env.IRONFLOW_API_KEY });

// A repeat emit with the same key returns the original event instead of
// emitting a second one.
const { eventId, runIds } = await ironflow.emit(
  "order.placed",
  { orderId: "ord_123", total: 99.5 },
  { idempotencyKey: "order.placed:ord_123" },
);

// emit returns once the event is stored. It does not wait for the runs.
console.log(eventId, runIds);

ironflow emit order.placed \
  --data '{"orderId":"ord_123","total":99.5}' \
  --idempotency-key order.placed:ord_123

import { createClient } from "@ironflow/node";

const ironflow = createClient({ apiKey: process.env.IRONFLOW_API_KEY });

// Registering a schema does not enforce it. The server must turn on
// IRONFLOW_EVENT_SCHEMA_ENFORCEMENT.
await ironflow.schemas.register({
  name: "order.placed",
  version: 1, // an emit with no version uses 1
  schema: {
    type: "object",
    required: ["orderId", "total"],
    properties: { orderId: { type: "string" }, total: { type: "number" } },
  },
});

# register the schema; this alone does not enforce it
ironflow event schema register order.placed --version 1 \
  --schema '{"type":"object","required":["orderId","total"]}'

# off (default) | warn | reject
IRONFLOW_EVENT_SCHEMA_ENFORCEMENT=reject ironflow serve

# check that enforcement really validates events
ironflow event schema check order.placed

// server
import { createClient } from "@ironflow/node";

const ironflow = createClient({ apiKey: process.env.IRONFLOW_API_KEY });

// publish() never triggers functions; emit() does.
await ironflow.publish(
  "orders.eu.shipped",
  { orderId: "ord_123", total: 250 },
  // same key while the server still holds it: no second message
  { idempotencyKey: "shipped:ord_123" },
);

// browser
import { ironflow } from "@ironflow/browser";

ironflow.configure({
  serverUrl: "https://ironflow.example.com",
  auth: { token: "<user-jwt>" },
});

// "*" matches one segment. The CEL filter runs on the server,
// so non-matching events never leave it.
const sub = await ironflow.subscribe<{ orderId: string; total: number }>(
  "topic:orders.*.shipped",
  {
    filter: "data.total > 100",
    onEvent: (e) => console.log("big shipment", e.data.orderId),
  },
);

// later: sub.unsubscribe()

import { createClient } from "@ironflow/node";
import { ironflow } from "@ironflow/browser";

// Create the group once (server side).
// ackMode "manual" redelivers anything not acked.
const admin = createClient({ apiKey: process.env.IRONFLOW_API_KEY });
await admin.consumerGroups.create({
  name: "order-workers",
  pattern: "events:order.>",
  ackMode: "manual",
});

// Each instance joins; every event reaches exactly one member.
const sub = await ironflow.joinConsumerGroup(
  "order-workers",
  "events:order.>",
  {
    onEvent: async (e) => {
      console.log("handling", e.data);
      // ack after the work is done, not before
      if (e.eventId) await sub.ack(e.eventId);
    },
  },
);

import { createClient } from "@ironflow/node";

const ironflow = createClient({ apiKey: process.env.IRONFLOW_API_KEY });

// expectedVersion: 0 = create only. A later writer passes the
// version it read; a stale one gets 409.
const { entityVersion, eventId } = await ironflow.streams.append(
  "ord_123",
  { name: "order.placed", entityType: "order", data: { total: 99.5 } },
  { expectedVersion: 0, idempotencyKey: "place:ord_123" },
);

// Second write builds on the version the first one returned
await ironflow.streams.append(
  "ord_123",
  { name: "order.paid", entityType: "order", data: { method: "card" } },
  { expectedVersion: entityVersion, idempotencyKey: "pay:ord_123" },
);

// Tombstones the stream; purge also deletes every earlier event.
// Irreversible.
await ironflow.streams.delete("ord_123", { purge: true });
console.log(eventId);

# Append with a version check and an idempotency key
ironflow stream append ord_123 --type order --event order.placed \
  --data '{"total":99.5}' --expected-version 0 \
  --idempotency-key place:ord_123

# Delete the stream and its history; --yes skips the prompt
ironflow stream delete ord_123 --purge --yes

// worker process
import {
  createProjection,
  createWorker,
  ironflow,
  type IronflowProjection,
} from "@ironflow/node";

type OrderStats = { orders: number; revenue: number };

// initialState makes this a managed projection: Ironflow keeps the state
const orderStats = createProjection<OrderStats>({
  name: "order-stats",
  events: ["order.placed"],
  initialState: () => ({ orders: 0, revenue: 0 }),
  // Pure reducer: return a new state; no clocks, no I/O.
  handler: (s: OrderStats, e: { data: unknown }) => ({
    orders: s.orders + 1,
    revenue: s.revenue + (e.data as { total: number }).total,
  }),
});

// A Node worker needs at least one function to register.
const placeOrder = ironflow.createFunction(
  { id: "place-order", triggers: [{ event: "order.requested" }] },
  async ({ event, step }) => {
    // Command handler: validate, then append order.placed.
  },
);

// Register the projection with a worker
await createWorker({
  functions: [placeOrder],
  projections: [orderStats] as IronflowProjection[],
}).start();

// API process
import { createClient } from "@ironflow/node";

const ironflow = createClient({ apiKey: process.env.IRONFLOW_API_KEY });

// Write side: append returns the id of the new event
const { eventId } = await ironflow.streams.append(
  "ord_123",
  { name: "order.placed", entityType: "order", data: { total: 99.5 } },
);

// Read-your-writes: wait until order-stats has processed that event.
// On timeout the result has timedOut: true; the call does not throw.
await ironflow.projections.waitForEvent(
  eventId,
  "order-stats",
  { timeoutMs: 5000 },
);

// browser
import { ironflow } from "@ironflow/browser";

ironflow.configure({
  serverUrl: "https://ironflow.example.com",
  auth: { token: "<user-jwt>" },
});

// onUpdate gets the new projection state after each change
await ironflow.subscribeToProjection<{ orders: number }>(
  "order-stats",
  { onUpdate: (s) => console.log(s.orders) },
);

import { createClient } from "@ironflow/node";

const ironflow = createClient({ apiKey: process.env.IRONFLOW_API_KEY });

// The table plus one SQL handler per event; the server runs the handlers.
// :entity_id and :data.* are bound from each event.
await ironflow.sqlProjections.create({
  name: "active_orders",
  tableSql: `
    CREATE TABLE proj_active_orders (
      order_id TEXT PRIMARY KEY,
      customer_id TEXT NOT NULL,
      total NUMERIC(12,2)
    )`,
  events: ["order.placed", "order.cancelled"],
  eventHandlers: {
    "order.placed":
      "INSERT INTO proj_active_orders (order_id, customer_id, total) VALUES (:entity_id, :data.customerId, :data.total)",
    "order.cancelled":
      "DELETE FROM proj_active_orders WHERE order_id = :entity_id",
  },
});

// Read side: filter the table with a SQL where clause
const { rows } = await ironflow.sqlProjections.query(
  "active_orders",
  { where: "customer_id = 'acme'", limit: 50 },
);
console.log(rows);

import { createClient, createFunction } from "@ironflow/node";

const ironflow = createClient({ apiKey: process.env.IRONFLOW_API_KEY });

// KV: TTL is a bucket setting; every key in "sessions" expires
// after an hour.
const kv = ironflow.kv();
await kv.createBucket({ name: "sessions", ttlSeconds: 3600 });
await kv.bucket("sessions").put("user.42", { cart: ["sku_1"] });

// Watch keys that match "user.*"
kv.bucket("sessions").watch(
  { onUpdate: (e) => console.log(e.key, e.operation) },
  { key: "user.*" },
);

// Config: set replaces, patch merges (and creates the document if
// missing).
const config = ironflow.config();
await config.set("app-settings", { theme: "dark", retries: 3 });
await config.patch("app-settings", { retries: 5 });
config.watch("app-settings", { onUpdate: (e) => console.log(e.data) });

// Secrets: a function sees only the secrets it declares.
export const payout = createFunction(
  {
    id: "payout",
    triggers: [{ event: "payout.requested" }],
    secrets: ["STRIPE_KEY"],
  },
  async ({ secrets }) => secrets.get("STRIPE_KEY") !== undefined,
);

ironflow secret set STRIPE_KEY sk_test_example \
  --description "Stripe test key"

import {
  createClient,
  createFunction,
} from "@ironflow/node";

const files = createClient({
  apiKey: process.env.IRONFLOW_API_KEY,
}).files();

// emitEvents: the bucket emits ironflow.file.* events.
// allowSignedUrls: the bucket accepts credential-free URLs.
await files.createBucket("inbox", {
  emitEvents: true,
  allowSignedUrls: true,
});

// Upload URL valid for 10 min; the upload is capped at 5 MB.
// Send `url` to the browser.
const { url } = await files.bucket("inbox").signUpload(
  "scans/ord_123.pdf",
  {
    ttlSeconds: 600,
    maxBytes: 5 * 1024 * 1024,
    contentType: "application/pdf",
  },
);

// Runs when a file lands in "inbox"
export const onScan = createFunction(
  {
    id: "process-scan",
    triggers: [{
      event: "ironflow.file.created",
      expression: "data.bucket == 'inbox'",
    }],
  },
  async ({ event }) => {
    const { path, etag } = event.data as {
      path: string;
      etag: string
    };
    // ifMatch pins the version that caused the event; delivery is
    // at-least-once, so key work on etag.
    return (
      await files.bucket("inbox").get(path, { ifMatch: etag })
    ).size;
  },
);

// browser
import { uploadToSignedUrl } from "@ironflow/browser";

// url comes from signUpload() on the backend
declare const url: string, file: File;

// No Ironflow credentials: the token in the URL is the credential
await uploadToSignedUrl(url, file);

apiVersion: ironflow/v1
kind: Server
spec:
  blobs:
    # blob backend; any S3-compatible store works
    url: "s3://ironflow-artifacts?endpoint=https://seaweedfs.internal:8333"
    # run and step outputs above this size go to the blob backend
    artifactThreshold: "1MB"

# or, as environment variables (bash/zsh; in fish use `set -x NAME value`)
export IRONFLOW_BLOB_URL="s3://ironflow-artifacts?endpoint=https://seaweedfs.internal:8333"
export IRONFLOW_ARTIFACT_THRESHOLD=1MB

# credentials for the S3 backend
export AWS_ACCESS_KEY_ID=<access-key>
export AWS_SECRET_ACCESS_KEY=<secret-key>

package ironflow.authz

# Policies only deny. RBAC grants; a policy cannot add permission.
# Only holders of the finance-admin role may emit payment.refund.* events.
deny contains "payment.refund.* needs the finance-admin role" if {
	# the request is an emit, and its resource holds a refund event name
	input.request.action == "events:emit"
	contains(input.request.resource, ":payment.refund.")
	# the caller lacks the role
	not "finance-admin" in input.subject.roles
}

# creates the dir and .ironflow-policy.json
ironflow policy pull ./policies

# evaluate the set against one request, before it goes live
ironflow policy dry-run ./policies --input request.json --explain

# a new set starts in monitor mode
ironflow policy push ./policies
ironflow policy push ./policies --mode enforce

# rollback saves an earlier version as a new version; the mode stays
ironflow policy versions
ironflow policy rollback 1 --dir ./policies

# ironflow.yaml
spec:
  auth:
    # off by default; while off, no request meets a policy set
    policyEnforcement: true

# Read who emitted an event. The engine sets it from the caller's
# credential; the event body cannot change it.
curl -s -H "Authorization: Bearer $IRONFLOW_API_KEY" \
  http://localhost:9123/api/v1/events/$EVENT_ID | jq -r .emitted_by
# Values: apikey:<id> | user:<id> | platform_user:<id>
#   | webhook:<source-id> | nats | platform

import { createFunction } from "@ironflow/node";

export const release = createFunction(
  { id: "release", triggers: [{ event: "deploy.requested" }] },
  async ({ event, step }) => {
    // Park the run until an approval with the same data.deployId arrives.
    // If the 24h timeout expires first, the run fails.
    const approval = await step.waitForEvent("wait-approval", {
      event: "deploy.approved",
      match: "data.deployId",
      timeout: "24h",
    });

    // Set by the engine from the caller's credential; the event body
    // cannot forge it.
    if (approval.emittedBy !== "apikey:ak_release_bot") {
      throw new Error(
        `approval from ${approval.emittedBy ?? "unknown"} is not accepted`
      );
    }
    return { deployId: event.data, approvedBy: approval.emittedBy };
  },
);

import { exposeMcp } from "@ironflow/node/agent";
import { z } from "zod";

// Register the tools with the Ironflow server.
const handle = await exposeMcp({
  name: "billing",
  version: "1.0.0",
  // Your serve() mount. Ironflow POSTs signed calls here.
  callbackUrl: "https://tools.example.com",
  tools: [
    {
      name: "get-invoice",
      description: "Fetch an invoice by ID",
      // The zod schema becomes the tool's JSON Schema.
      input: z.object({ id: z.string() }),
      // The calling key needs this in its --tool-scope list.
      scopes: ["billing-read"],
      handler: async ({ id }) => ({ id, status: "paid" }),
    },
  ],
});
console.log(handle.toolNames); // ["billing.get-invoice"]

# Remote callers send this token to the adapter.
export IRONFLOW_A2A_BEARER_TOKEN=$(openssl rand -hex 24)
# The adapter's own key for the server. It needs the developer role.
export IRONFLOW_API_KEY=ifkey_...

# --caller-input is <function-id>/<event-name>. A wait that is not
# listed stays hidden from callers. Use the same name in approve().
ironflow a2a \
  --agent code-review \
  --caller-input code-review/code-review.post-review.approved \
  --public-url https://agents.example.com
# Agent Card, served at:
# https://agents.example.com/agents/code-review/.well-known/agent-card.json

{
  "mcpServers": {
    "ironflow": {
      "command": "ironflow",
      "args": ["mcp", "--allow-writes"],
      "env": { "IRONFLOW_API_KEY": "${IRONFLOW_API_KEY}" }
    }
  }
}

{
  "ironflow": {
    "command": "ironflow",
    "args": ["mcp", "--allow-writes", "--server-url", "http://localhost:9123"]
  }
}

# Write the bundled skills to ./.agents/skills (matches the binary version).
ironflow skills sync

# Detect Claude Code, Codex and Gemini. Print the wiring commands.
# It changes nothing.
ironflow skills doctor

# Or write to $HOME/.agents/skills for every project.
ironflow skills sync --global

{
  "agent_servers": {
    "Ironflow": {
      "type": "custom",
      "command": "npx",
      "args": ["-y", "@ironflow/agent"],
      "env": { "IRONFLOW_SERVER_URL": "http://localhost:9123" }
    }
  }
}

import { test, expect } from "vitest";
import { createTestClient } from "@ironflow/node/test";
import { fulfilOrder } from "./functions.js";

test("fulfils an order once payment is confirmed", async () => {
  const t = createTestClient({ functions: [fulfilOrder] });

  // The handler gets this value in place of the real step body.
  t.mockStep("reserve-stock", () => "sku-1");
  // Queue the event that waitForEvent receives. Send it before emit().
  t.sendEvent("payment.confirmed", { orderId: "ord_1" });

  // Run the function that order.placed triggers, in memory.
  const run = await t.emit("order.placed", { orderId: "ord_1" });

  expect(run.status).toBe("completed");
  expect(run.stepOutput("reserve-stock")).toBe("sku-1");
});

import (
	"testing"

	"github.com/sahina/ironflow-go/ironflow"
	"github.com/sahina/ironflow-go/ironflow/ironflowtest"
)

func TestFulfilOrder(t *testing.T) {
	tc := ironflowtest.NewClient(t, ironflowtest.Config{
		Functions: []ironflow.Function{FulfilOrder},
	})

	// The handler gets this value in place of the real step body.
	tc.MockStep("reserve-stock", func() (any, error) {
		return "sku-1", nil
	})
	// Queue the event that WaitForEvent receives. Send it before Emit.
	tc.SendEvent("payment.confirmed", map[string]any{"orderId": "ord_1"})

	// Run the function that order.placed triggers, in memory.
	run := tc.Emit(t, "order.placed", map[string]any{"orderId": "ord_1"})

	if run.Status != "completed" {
		t.Fatalf("status %s: %v", run.Status, run.Error)
	}
	if run.StepOutput("reserve-stock") != "sku-1" {
		t.Fatal("step output missing")
	}
}

import asyncio

from functions import fulfil_order
from ironflow.testing import TestClient


def test_fulfil_order():
    t = TestClient([fulfil_order])

    # The handler gets this value in place of the real step body.
    t.mock_step("reserve-stock", lambda: "sku-1")
    # Queue the event that wait_for_event receives. Send it before emit().
    t.send_event("payment.confirmed", {"orderId": "ord_1"})

    # emit() is async, so asyncio.run waits for the run to finish.
    run = asyncio.run(t.emit("order.placed", {"orderId": "ord_1"}))

    assert run.status == "completed"
    assert run.step_output("reserve-stock") == "sku-1"

ironflow config init --prod > ironflow.yaml
# Check the file. It also lists env vars that -f makes serve ignore.
ironflow validate -f ironflow.yaml
ironflow serve -f ironflow.yaml

apiVersion: ironflow/v1
kind: Server
spec:
  port: 9123
  storage:
    driver: postgres
    # ${VAR} reads the environment; startup fails if it is not set
    url: ${IRONFLOW_DATABASE_URL}
    # Pool limits; minConns cannot be more than maxConns
    pool:
      maxConns: 25
      minConns: 5
      maxIdleTime: 30s
  nats:
    # JetStream data directory; required with PostgreSQL and embedded NATS
    storeDir: /data/nats
  auth:
    # Encrypts stored secrets. If you rotate it, old secrets are unreadable
    masterKey: ${IRONFLOW_MASTER_KEY}
  observability:
    metrics:
      # Prometheus metrics at /metrics
      enabled: true

# Get the Compose files
git clone --depth 1 https://github.com/sahina/ironflow-releases.git
cd ironflow-releases/deploy/docker-compose

# One engine container with embedded NATS and SQLite. API on :9123.
docker compose -f docker-compose.single-node.yml up -d

# Get the Compose files
git clone --depth 1 https://github.com/sahina/ironflow-releases.git
cd ironflow-releases/deploy/docker-compose/cluster

# Write .env with generated keys. Keep a copy: it holds the master key.
# The script never overwrites an existing .env.
./init-env.sh

# Starter: one node, NATS, PostgreSQL and SeaweedFS. API on :9123.
# Or use docker-compose.cluster.yml (3 nodes behind nginx) or .ha.yml.
docker compose -f docker-compose.starter.yml up -d