Skip to content
ironflow
Ironflow is pre-1.0: APIs can change between releases. Pin a version. What is ready →

Agent Template

Copy this template into your project as agent.md, CLAUDE.md, .cursorrules, or whichever file your AI tool reads for project-wide context.

Full Template

Section titled “Full Template”
# Ironflow Agent Context


This project uses Ironflow, a durable runtime for AI agents and event-driven
systems. Completed durable steps are memoized. Domain events are appended to
entity streams. Execution history is available when recording is enabled.


## What is Ironflow?


**Agents and workflows that survive.** Ironflow resumes completed work after
crashes, waits safely for external input, and derives state from events, all in
one binary.


- **Events (Emit):** Append-only, immutable facts. The source of truth.
- **Functions (React):** Durable workflows with memoized steps. Crash-safe — resumes from the last completed step.
- **Projections (Derive):** Pure reducers that build read models from events. State is always up to date.
- **Continuous History:** Events and recorded execution steps form an append-only history. Enable function execution recording with `recording: true`.


---


## Integration Workflow


When adding Ironflow to a project, follow this sequence. **Ask before creating files** — propose the structure and let the user confirm or adjust.


### Step 1: Confirm Project Structure


Before creating files, ask the user:


> "I'll set up Ironflow in this project. I see you're using [Next.js/Node.js/Go]. I suggest this structure:
>
> ```
> [proposed structure based on framework]
> ```
>
> Does this work, or would you prefer a different layout?"


### Step 2: Install SDK and Create Client


**TypeScript:**
```bash
pnpm add @ironflow/node
```


Create `src/lib/ironflow.ts` (or `lib/ironflow.ts` for Next.js):
```typescript
import { createClient } from "@ironflow/node";


export const ironflow = createClient({
  serverUrl: process.env.IRONFLOW_SERVER_URL || "http://localhost:9123",
  apiKey: process.env.IRONFLOW_API_KEY,
});
```


**Go:**
```bash
go get github.com/sahina/ironflow-go/ironflow
```


Create `internal/ironflow/client.go`:
```go
package ironflow


import (
    "os"


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


// An empty ServerURL falls back to http://localhost:9123.
var Client = sdk.NewClient(sdk.ClientConfig{
    ServerURL: os.Getenv("IRONFLOW_SERVER_URL"),
    APIKey:    os.Getenv("IRONFLOW_API_KEY"),
})
```


**Python:**


```bash
pip install ironflow-py
```


Install `ironflow-py`, not `ironflow` — the bare name on PyPI is an unrelated
third-party package. The import name stays `ironflow`. Functions and the worker come
from `ironflow.worker`; see the Python SDK Reference below.


Create `ironflow/client.py` (or wherever the project's helpers live):
```python
import os
from ironflow import IronflowClient


client = IronflowClient(
    server_url=os.environ.get("IRONFLOW_SERVER_URL", "http://localhost:9123"),
    api_key=os.environ.get("IRONFLOW_API_KEY"),
)
```


### Step 3: Environment Variables


Add to `.env.example` (or equivalent):
```
IRONFLOW_SERVER_URL=http://localhost:9123
IRONFLOW_API_KEY=         # Optional in dev mode
```


### Step 4: Create Worker Entry Point


Ask the user:
> "I'll create a worker entry point at `[proposed path]`. This will register your functions and projections. Should I include any initial functions, or start with an empty worker?"


### Step 5: Offer Tests


After creating functions or projections, ask:
> "Should I create tests for the Ironflow components? I can add:
> - Unit tests for projection handlers (pure functions, easy to test)
> - Integration tests that emit events and verify function execution"


---


## Testing Ironflow Components


### Testing Projections (Unit)


```typescript
import { describe, it, expect } from "vitest";
import { orderStats } from "./order-stats";


describe("orderStats projection", () => {
  it("increments totalOrders on order.placed", () => {
    const state = orderStats.config.initialState!();
    const result = orderStats.config.handler(state, {
      name: "order.placed",
      data: { orderId: "123", total: 99.99 },
      timestamp: new Date().toISOString(),
    });
    expect(result.totalOrders).toBe(1);
    expect(result.totalRevenue).toBe(99.99);
  });
});
```


### Testing Functions (Integration)


```typescript
import { describe, it, expect, beforeAll } from "vitest";
import { createClient } from "@ironflow/node";


const ironflow = createClient();


describe("process-order function", () => {
  it("processes a valid order", async () => {
    const result = await ironflow.emit("order.placed", {
      orderId: "test-123",
      total: 49.99,
    });


    // Poll until the triggered run completes
    let run = await ironflow.getRun(result.runIds[0]);
    while (run.status === "running" || run.status === "waiting_for_capacity" || run.status === "waiting" || run.status === "paused") {
      await new Promise((r) => setTimeout(r, 100));
      run = await ironflow.getRun(result.runIds[0]);
    }
    expect(run.status).toBe("completed");
  });
});
```


**Python:** unit-test a projection by calling its handler, and a function with the
in-process test kit (`ironflow.testing`), which needs no server:


```python
import asyncio


from ironflow.projection import EventInfo, ProjectionContext, ProjectionEvent, ProjectionInfo
from ironflow.testing import TestClient


from functions import order_stats, process_order




def test_order_stats_increments_on_order_placed():
    event = ProjectionEvent(id="e1", name="order.placed", data={"orderId": "123", "total": 99.99},
                            seq=1, timestamp="2026-01-01T00:00:00Z", source="", metadata={})
    ctx = ProjectionContext(event=EventInfo(id="e1", name="order.placed", seq=1, timestamp=event.timestamp),
                            projection=ProjectionInfo(name="order-stats", version=1))
    result = order_stats.handler(order_stats.initial_state(), event, ctx)
    assert result["totalOrders"] == 1
    assert result["totalRevenue"] == 99.99




def test_process_order_completes():
    run = asyncio.run(TestClient([process_order]).emit("order.placed", {"orderId": "test-123", "total": 49.99}))
    assert run.status == "completed"
```


---


## Running Ironflow


```bash
# Start the server (dev mode — no auth required)
ironflow serve --dev


# Trigger events via CLI
ironflow emit order.placed --data '{"orderId": "123", "total": 99.99}'


# Query derived state
curl -s -X POST http://localhost:9123/ironflow.v1.ProjectionService/GetProjection -H 'Content-Type: application/json' -d '{"name":"order-stats"}' | jq '.stateValue // .state'


# Time-travel replay
ironflow run list
ironflow inspect <run_id> --replay


# Access Dashboard
open http://localhost:9123
```


---


## Project Structure Conventions


### TypeScript (Next.js)


```
app/
  api/
    ironflow/
      route.ts          # Push mode handler (serve)
src/
  functions/            # Function definitions
    process-order.ts
  projections/          # Projection definitions
    order-stats.ts
worker.ts               # Pull mode worker entry point
```


### TypeScript (Standalone Node.js)


```
src/
  index.ts              # Worker entry point
  functions/            # Function definitions
    process-order.ts
  projections/          # Projection definitions
    order-stats.ts
```


### Go


```
cmd/
  worker/
    main.go             # Worker entry point
internal/
  functions/            # Function definitions
    process_order.go
  projections/          # Projection definitions
    order_stats.go
```


### TypeScript (DDD / CQRS)


```
src/
  domain/                  # Domain layer (bounded contexts)
    order/
      aggregate.ts         # State rebuilders, invariants
      events.ts            # Domain event types/schemas
      commands.ts          # Command types/schemas
    customer/
      aggregate.ts
      events.ts
  application/             # Application layer
    handlers/              # Command handlers (functions)
      validate-order.ts
      fulfill-order.ts
    projections/           # Read models (CQRS)
      order-dashboard.ts
      customer-orders.ts
    sagas/                 # Process managers
      order-fulfillment.ts
  infrastructure/
    ironflow.ts            # Client config
worker.ts                  # Entry point
```


---


## TypeScript SDK Reference


### Installation
```bash
pnpm add @ironflow/node        # Server-side: functions, workers, projections
pnpm add @ironflow/browser     # Client-side: subscriptions, triggers, KV
```


### Creating Functions


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


export const processOrder = createFunction(
  {
    id: "process-order",
    triggers: [{ event: "order.placed" }],
    recording: true, // Enable time-travel debugging
  },
  async ({ event, step }) => {
    const data = event.data as OrderData;


    // Each step is memoized — safe to retry, survives crashes
    const order = await step.run("validate-order", async () => {
      return { valid: true, orderId: data.orderId };
    });


    const payment = await step.run("charge-payment", async () => {
      return { status: "success", txId: `txn_${Date.now()}` };
    });


    return { orderId: order.orderId, payment };
  },
);
```


### Function Config Options


```typescript
{
  id: string;              // Required. Unique function identifier
  triggers: Trigger[];     // Required. Events that invoke this function
  name?: string;           // Display name
  recording?: boolean;     // Enable audit recording (time-travel debugging)
  recordingProfile?: "all" | "run_lifecycle" | "steps"; // Narrower capture; set it and `recording` is implied
  retry?: {                // Retry configuration for failed steps
    maxAttempts?: number;
    initialDelayMs?: number;
    backoffFactor?: number;
    maxDelayMs?: number;   // Max retry delay in ms (default: 300000)
  };
  timeout?: number;        // Function timeout in ms (unset: no per-function budget)
  stepTimeout?: string;    // Default timeout for all step.run() calls
  concurrency?: {          // Concurrency control
    limit: number;         // Required. Max concurrent runs
    key?: string;          // Path under event data, e.g. "customerId"
  };
  mode?: "push" | "pull";  // Execution mode
  secrets?: string[];      // Secret names resolved at execution time
  metadata?: Record<string, unknown>;
  description?: string;           // Human-readable description shown in dashboard
  schema?: TEventSchema;           // Zod schema for type-safe event validation
  debounce?: DebounceConfig;       // Collapse rapid-fire events
  cancelOn?: CancelOnConfig[];     // Auto-cancel on matching events
  actorKey?: string;               // Stored on the function; the engine does not yet use it for routing
}
```


### Step Methods


```typescript
// Memoized step execution — the core primitive
const result = await step.run("step-name", async () => {
  return { processed: true };
});


// Durable sleep — survives restarts
await step.sleep("wait-before-retry", "1h");


// Sleep until a specific time
await step.sleepUntil("wait-for-market-open", "2026-03-16T09:30:00Z");


// Wait for an external event (human-in-the-loop, callbacks)
const approval = await step.waitForEvent("wait-for-approval", {
  event: "order.approved",
  match: "data.orderId",   // Correlate by field
  timeout: "24h",
});


// Parallel branches
const [a, b] = await step.parallel("fetch-all", [
  async (s) => s.run("fetch-user", async () => fetchUser(id)),
  async (s) => s.run("fetch-order", async () => fetchOrder(id)),
]);


// Parallel map with concurrency control
const results = await step.map("process-items", items, async (item, s, i) => {
  return s.run(`process-${i}`, async () => processItem(item));
}, { concurrency: 5 });


// Saga compensation (undo on failure)
await step.run("charge-card", async () => charge(amount));
step.compensate("charge-card", async () => refund(amount));


// Invoke another function (sync)
const result = await step.invoke<InvoiceResult>("generate-invoice", { orderId });


// Invoke another function (fire-and-forget)
const { runId } = await step.invokeAsync("send-report", { userId });


// Publish to a pub/sub topic (does NOT trigger functions)
await step.publish("notifications", { type: "order.shipped", orderId });
```


### Managed Projections


```typescript
import { createProjection, type IronflowProjection } from "@ironflow/node";


const orderStats = createProjection({
  name: "order-stats",
  events: ["order.placed"],
  initialState: () => ({ totalOrders: 0, totalRevenue: 0 }),
  handler: (
    state: { totalOrders: number; totalRevenue: number },
    event: { name: string; data: unknown },
  ) => ({
    totalOrders: state.totalOrders + 1,
    totalRevenue: state.totalRevenue + ((event.data as OrderData).total ?? 0),
  }),
});
```


### Worker Setup (Pull Mode)


```typescript
import { createWorker, type IronflowProjection } from "@ironflow/node";


const worker = createWorker({
  functions: [processOrder],
  projections: [orderStats as IronflowProjection],
});


await worker.start();
```


### Serve Setup (Push Mode — Serverless)


```typescript
// app/api/ironflow/route.ts (Next.js App Router)
import { serve } from "@ironflow/node";


export const POST = serve({
  functions: [processOrder],
});
```


### Entity Streams (Event Sourcing)


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


const client = createClient();


// Append an event to an entity stream
await client.streams.append(
  "order-123",
  {
    name: "order.placed",
    data: { total: 99.99 },
    entityType: "order", // Required. Entity type for the stream
  },
  {
    expectedVersion: 0, // Optimistic concurrency (in options, not input)
  },
);


// Read the full entity history
const { events } = await client.streams.read("order-123");
```


### Error Handling


```typescript
import { NonRetryableError, StepError } from "@ironflow/node";


// Non-retryable — fails immediately, no retry
throw new NonRetryableError("Invalid order ID");


// Retryable — will be retried per retry config
throw new Error("Payment gateway timeout");
```


---


## Go SDK Reference


### Installation
```bash
go get github.com/sahina/ironflow-go/ironflow
```


### Creating Functions
```go
import "github.com/sahina/ironflow-go/ironflow"


var processOrder = ironflow.CreateFunction(ironflow.FunctionConfig{
    ID:        "process-order",
    Triggers:  []ironflow.Trigger{{Event: "order.placed"}},
    Recording: true,
}, func(ctx ironflow.Context) (any, error) {
    var data OrderData
    if err := ctx.Event.Data(&data); err != nil {
        return nil, err
    }


    order, err := ironflow.Run(ctx, "validate-order", func() (Order, error) {
        return Order{Valid: true, OrderID: data.OrderID}, nil
    })
    if err != nil {
        return nil, err
    }


    return map[string]any{"order": order}, nil
})
```


### Step Methods
```go
// Memoized step execution (generic — infers return type)
result, err := ironflow.Run(ctx, "step-name", func() (MyType, error) {
    return MyType{Done: true}, nil
})


// With timeout override
result, err := ironflow.Run(ctx, "slow-step", func() (string, error) {
    return callExternalAPI()
}, ironflow.WithTimeout(30*time.Second))


// Durable sleep
ironflow.Sleep(ctx, "wait-before-retry", 1*time.Hour)


// Sleep until a specific time
ironflow.SleepUntil(ctx, "wait-for-open", time.Date(2026, 3, 16, 9, 30, 0, 0, time.UTC))


// Wait for an external event
event, err := ironflow.WaitForEvent[any](ctx, "wait-approval", ironflow.EventFilter{
    Event:   "order.approved",
    Match:   "data.orderId",
    Timeout: 24 * time.Hour,
})


// Parallel branches
results, err := ironflow.Parallel(ctx, "fetch-all", []func(*ironflow.BranchContext) (any, error){
    func(b *ironflow.BranchContext) (any, error) { return fetchUser(id) },
    func(b *ironflow.BranchContext) (any, error) { return fetchOrder(id) },
})


// Parallel map with concurrency
results, err := ironflow.Map(ctx, "process-items", items, func(item Item, b *ironflow.BranchContext, i int) (Result, error) {
    return processItem(item)
}, ironflow.ParallelOptions{Concurrency: 5})


// Saga compensation
ironflow.Compensate(ctx, "charge-card", func() error {
    return refund(amount)
})


// Invoke another function (sync)
result, err := ironflow.Invoke[InvoiceResult](ctx, "generate-invoice", input)


// Invoke another function (fire-and-forget)
asyncResult, err := ironflow.InvokeAsync(ctx, "send-report", input)
// asyncResult.RunID contains the child run ID


// Publish to a pub/sub topic
ironflow.Publish(ctx, "notifications", data)
```


### Managed Projections
```go
var orderStats = ironflow.CreateProjection(ironflow.ProjectionConfig{
    Name:   "order-stats",
    Events: []string{"order.placed"},
    InitialState: func() map[string]any {
        // Use float64 literals so the .(float64) assertions below hold on
        // first in-memory call and on rebuild (JSON unmarshal yields float64).
        return map[string]any{"totalOrders": 0.0, "totalRevenue": 0.0}
    },
    Handler: func(state map[string]any, event ironflow.ProjectionEvent, ctx ironflow.ProjectionContext) (map[string]any, error) {
        // Return a fresh map — never mutate `state` then return it.
        // Mutation + return of the same reference is an aliasing hazard.
        next := make(map[string]any, len(state)+1)
        for k, v := range state {
            next[k] = v
        }
        next["totalOrders"] = state["totalOrders"].(float64) + 1
        next["totalRevenue"] = state["totalRevenue"].(float64) + event.Data["total"].(float64)
        return next, nil
    },
})
```


### Worker Setup (Pull Mode)
```go
worker := ironflow.NewWorker(ironflow.WorkerConfig{
    ServerURL:   "http://localhost:9123",
    Functions:   []ironflow.Function{processOrder},
    Projections: []ironflow.Projection{orderStats},
})
worker.Run(ctx)
```


### Serve Setup (Push Mode — HTTP Handler)
```go
handler := ironflow.Serve(ironflow.ServeConfig{
    Functions: []ironflow.Function{processOrder},
})
http.ListenAndServe(":3001", handler)
```


---


## Python SDK Reference


### Installation
```bash
pip install ironflow-py
```


### Creating Functions
```python
from ironflow.worker import function




@function(id="process-order", triggers=[{"event": "order.placed"}], recording=True)
async def process_order(ctx):
    data = ctx.event.data


    order = await ctx.step.run("validate-order", lambda: {"valid": True, "orderId": data["orderId"]})


    return {"order": order}
```


Handlers are `async def`. Step callbacks may be sync (run in a thread) or `async`.
A bare number duration is **seconds**, not milliseconds.


### Step Methods
```python
# Memoized step execution
result = await ctx.step.run("step-name", lambda: {"done": True})


# With timeout override
result = await ctx.step.run("slow-step", call_external_api, timeout="30s")


# Durable sleep
await ctx.step.sleep("wait-before-retry", "1h")


# Sleep until a specific time
await ctx.step.sleep_until("wait-for-open", datetime(2026, 3, 16, 9, 30, tzinfo=timezone.utc))


# Wait for an external event
event = await ctx.step.wait_for_event(
    "wait-approval", event="order.approved", match="data.orderId", timeout="24h",
)


# Parallel branches — each branch gets its own scoped step client
results = await ctx.step.parallel("fetch-all", [
    lambda b: b.run("fetch-user", lambda: fetch_user(id)),
    lambda b: b.run("fetch-order", lambda: fetch_order(id)),
])


# Parallel map with concurrency
results = await ctx.step.map(
    "process-items", items,
    lambda item, b, i: b.run(f"process-{i}", lambda: process_item(item)),
    concurrency=5,
)


# Saga compensation (registers the undo; not awaited)
ctx.step.compensate("charge-card", lambda: refund(amount))


# Invoke another function (sync)
result = await ctx.step.invoke("generate-invoice", input)


# Invoke another function (fire-and-forget)
async_result = await ctx.step.invoke_async("send-report", input)
# async_result.run_id contains the child run ID


# Publish to a pub/sub topic (durable, memoized)
await ctx.step.publish("agent.events", {"status": "done"})
```


### Managed Projections
```python
from ironflow.projection import create_projection


order_stats = create_projection(
    name="order-stats",
    events=["order.placed"],
    initial_state=lambda: {"totalOrders": 0, "totalRevenue": 0},
    # Return a new dict — never mutate `state` and return it.
    handler=lambda state, event, ctx: {
        **state,
        "totalOrders": state["totalOrders"] + 1,
        "totalRevenue": state["totalRevenue"] + event.data["total"],
    },
)
```


### Worker Setup (Pull Mode)
```python
from ironflow.worker import Worker


Worker(
    server_url="http://localhost:9123",
    functions=[process_order],
    projections=[order_stats],
).run()
```


### Serve Setup (Push Mode — ASGI)
```python
from ironflow.serve import serve


app = serve([process_order])  # run with: uvicorn app:app --port 3001
```


---


## Critical Patterns


### 1. All Side Effects Inside Steps
All I/O (API calls, DB writes, emails) MUST be inside `step.run()`. Code outside steps re-executes on every retry.


### 2. Unique Step IDs
Each step within a function MUST have a unique ID. In loops, use `step.map()` **with its scoped step client** — `step.map(name, items, (item, s, i) => s.run(\`process-\${i}\`, ...))` — or include the index in a plain `step.run(\`process-\${i}\`, ...)`. `step.map()` on its own records no steps at all; the scoped `s` is what makes each item durable.


### 3. Pure Projections
Managed projection handlers must be **pure functions** — no API calls, no database writes, no side effects. For side effects, use external projections with `mode: "external"`.


### 4. Optimistic Concurrency
When appending to entity streams, always fetch stream info first and pass `expectedVersion` to detect concurrent writes.


### 5. Enable Recording for Debugging
Set `recording: true` on any function you want to time-travel debug. This records each step for replay. Rows are subject to retention and can [drop under load](/explanation/run-outcomes/#audit-rows-can-drop-under-load).


### 6. Projection Type Casting
When passing projections to `createWorker`, cast them: `projections: [myProjection as IronflowProjection]`.


## Common Mistakes to Avoid
- Putting I/O outside of `step.run()` — it will re-execute on retries
- Reusing step IDs in loops — use `step.map()` with its scoped step client, or indexed IDs
- Ignoring the scoped step client inside `step.map()` / `step.parallel()` — a branch is not itself a recorded step, so work done in the callback body re-runs in full on every retry (and work routed through the enclosing `step` is memoized outside the branch's scope)
- Forgetting `recording: true` — without it, time-travel debugging is not available
- Using `initialState` as a plain object instead of a function: `initialState: () => ({...})`
- Expecting code after `step.waitForEvent` to run on timeout — the engine fails the run instead. Model a deadline as a second event
- Using push mode for tasks that take longer than 10 seconds — use pull mode instead
- Using `import { ironflow } from "@ironflow/node"` with `ironflow.createFunction()` — prefer the direct import: `import { createFunction } from "@ironflow/node"`