Defining Functions

A function is a workflow definition that responds to events. Functions contain one or more steps that execute in sequence, with automatic memoization for durability.

Basic Function Definition

TypeScript

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

export const processOrder = ironflow.createFunction(
  {
    id: "process-order",
    triggers: [{ event: "order.placed" }],
  },
  async ({ event, step }) => {
    // Event data is typed via generics or casting
    const { orderId } = event.data as { orderId: string };

    const result = await step.run("validate", async () => {
      return await validateOrder(orderId);
    });

    return { success: true, result };
  },
);

Go

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

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

    result, err := ironflow.Run(ctx, "validate", func() (bool, error) {
        return validateOrder(data.OrderID)
    })

    return map[string]any{"success": result}, err
})

Python

from ironflow.worker import Context, function


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

    result = await ctx.step.run("validate", lambda: validate_order(order_id))

    return {"success": True, "result": result}

---

Configuration Options

Functions accept a configuration object to tune retry logic, timeouts, and concurrency. The concurrency.key field is a dot-notation path into the triggering event’s data payload (e.g. orgId, or org.id for a nested field). A key that does not resolve to a scalar fails the run with CONCURRENCY_KEY_UNRESOLVED.

These are the common ones. For every field — including stepTimeout, recording, recordingProfile, actorKey, name, description and metadata — see FunctionConfig. Two fields have their own pages: debounce and cancelOn.

TypeScript

ironflow.createFunction(
  {
    id: "my-function",
    triggers: [{ event: "my.event" }],
    mode: "push", // "push" (default) or "pull"
    // Push execution budget in ms. It can only *tighten* the server's
    // `engine.pushTimeout` ceiling (default 10s), never raise it, and it is not
    // applied at all in pull mode (which is unlimited).
    timeout: 5_000,
    retry: {
      maxAttempts: 3,
      initialDelayMs: 1000,
      backoffFactor: 2.0,
    },
    concurrency: {
      limit: 10,
      key: "orgId", // Group limits by this field of event.data
    },
    secrets: ["STRIPE_KEY"], // Inject secrets at runtime
  },
  handler,
);

Go

import "time"

ironflow.CreateFunction(ironflow.FunctionConfig{
    ID:       "my-function",
    Triggers: []ironflow.Trigger{{Event: "my.event"}},
    Mode:     "pull",
    Retry: &ironflow.RetryConfig{
        MaxAttempts:   5,
        InitialDelay:  time.Second,
        BackoffFactor: 2.0,
    },
    Concurrency: &ironflow.ConcurrencyConfig{
        Limit: 10,
        Key:   "orgId",
    },
    Secrets: []string{"STRIPE_KEY"},
}, handler)

Python

from ironflow.worker import function


@function(
    id="my-function",
    triggers=[{"event": "my.event"}],
    # A bare number is seconds, not milliseconds.
    retry={"max_attempts": 5, "initial_delay": 1, "backoff_factor": 2.0},
    concurrency={"limit": 10, "key": "orgId"},
    secrets=["STRIPE_KEY"],
)
async def handler(ctx): ...

Python has no mode field. Pass the function to Worker(functions=[...]) for pull mode, or to ironflow.serve.serve([...]) for push mode.

---

Event Data Access

The event object in the context contains the triggering event’s payload and metadata.

TypeScript

import { z } from "zod";

// Define your event schema with Zod
const orderPlacedSchema = z.object({
  orderId: z.string(),
  amount: z.number(),
});

// Pass the schema in the config for type-safe event data
export const myFn = ironflow.createFunction(
  {
    id: "my-fn",
    triggers: [{ event: "order.placed" }],
    schema: orderPlacedSchema,
  },
  async ({ event, step }) => {
    // event.data is now type-safe: { orderId: string, amount: number }
    console.log(event.data.orderId);

    // Access user-defined metadata attached when the event was emitted
    const traceId = event.metadata?.traceId as string | undefined;
    const source = event.metadata?.source as string | undefined;
  },
);

The schema is enforced at runtime on every execution path (push, pull, streaming pull, and the test harness). Before the handler runs, the SDK parses event.data with the schema. On success, the handler receives the parsed value, so Zod defaults and transforms apply and z.object() strips keys the schema does not declare (use .loose() to keep them). On failure, the run fails with a non-retryable SchemaValidationError (code VALIDATION_ERROR) that names the function and the first ten Zod issues; issue text can include received values. The event record is kept, so you can inspect the payload in the dashboard. Cron ticks are not validated: their payload is engine-generated, so a function with both an event and a cron trigger keeps running on schedule. Functions without a schema receive event.data unchanged.

Go

func(ctx ironflow.Context) (any, error) {
    var data MyData
    // Unmarshal JSON data into your struct
    if err := ctx.Event.Data(&data); err != nil {
        return nil, err
    }

    // Access user-defined metadata attached when the event was emitted
    traceId, _ := ctx.Event.Metadata["traceId"].(string)
    source, _ := ctx.Event.Metadata["source"].(string)

    return data.ID, nil
}

Python

from pydantic import BaseModel
from ironflow.worker import function


class OrderPlaced(BaseModel):
    orderId: str
    amount: float


@function(id="my-fn", triggers=[{"event": "order.placed"}], schema=OrderPlaced.model_validate)
async def my_fn(ctx):
    # ctx.event.data is the value the schema returned: an OrderPlaced instance.
    print(ctx.event.data.orderId)

schema is any callable that takes the event data and returns the value the handler sees. If it raises, the run fails with a non-retryable SchemaValidationError. Cron events skip it.

ctx.event.metadata holds the metadata attached when the event was emitted. It is None when the event carries none.

from ironflow.worker import function


@function(id="trace-fn", triggers=[{"event": "order.placed"}])
async def trace_fn(ctx):
    trace_id = (ctx.event.metadata or {}).get("traceId")

---

Secrets Injection

If your function requires sensitive API keys, declare them in the secrets array. Ironflow will resolve these from its encrypted vault and inject them into the handler context at runtime.

TypeScript

async ({ event, step, secrets }) => {
  const apiKey = secrets.get("STRIPE_KEY");
};

Go

func(ctx ironflow.Context) (any, error) {
    var apiKey string
    if err := ctx.Secrets.Get("STRIPE_KEY", &apiKey); err != nil {
        return nil, err
    }
    return apiKey, nil
}

Python

async def handler(ctx):
    api_key = ctx.secrets["STRIPE_KEY"]

---

What’s Next?

• Step Primitives — Learn about run, sleep, and waitForEvent
• Execution Modes — Push mode vs. Pull mode