> For the complete documentation index, see [llms.txt](/llms.txt).
> A full single-fetch corpus is available at [llms-full.txt](/llms-full.txt).
---
title: Workflows
description: Define durable orchestrators that sequence functions, agents, and human approvals into reliable pipelines.
last_verified: 2026-07-24
---



A **workflow** is a durable orchestrator that sequences functions, agents, and human approvals into a reliable pipeline. If a workflow crashes mid-run, it restarts and picks up from the last completed step rather than starting over.

---

## Defining a workflow



**Python:**

Decorate any async function with `@workflow`. The function is registered automatically at import time.

```python
from agnt5 import workflow, WorkflowContext

@workflow
async def onboarding_workflow(ctx: WorkflowContext, user_email: str) -> dict:
    account = await ctx.step(create_account, user_email)
    await ctx.step(send_welcome_email, account["id"])
    return {"status": "done", "account_id": account["id"]}
```

The first parameter must be `ctx: WorkflowContext`. Everything after that is your workflow's input (whatever the caller passes) when triggering the run.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `name` | `str` | `function.__name__` | Name registered with the platform |
| `triggers` | `list[TriggerSpec]` | `None` | Event or webhook triggers that start this workflow automatically |





**TypeScript:**

Pass a name and an async handler to `workflow(...)`. The workflow is registered automatically at import time.

```typescript
export const onboardingWorkflow = workflow(
  'onboarding_workflow',
  async (ctx: Context, input: { userEmail: string }) => {
    const account = await createAccount(ctx, { userEmail: input.userEmail });
    await sendWelcomeEmail(ctx, { accountId: account.id });
    return { status: 'done', accountId: account.id };
  },
);
```

The handler's first parameter is always `ctx: Context`. The second is your workflow's input (whatever the caller passes) when triggering the run.

`workflow(name, handler, options?)` accepts:

| Option | Type | Default | Description |
|---|---|---|---|
| `triggers` | `TriggerSpec[]` | `undefined` | Event or webhook triggers that start this workflow automatically |
| `cron` | `string` | `undefined` | Cron expression for scheduled execution |





**Go:**

Pass a handler to `RegisterWorkflow`, with the same generic signature as `RegisterFunction`. It's registered on the `Worker` you pass in, not automatically at import time.

```go
import "github.com/agnt5dev/sdk-go/agnt5"

err := agnt5.RegisterWorkflow(worker, "onboarding_workflow",
    func(ctx *agnt5.Context, in OnboardingInput) (OnboardingOutput, error) {
        account, err := agnt5.Step(ctx, "create_account", func(context.Context) (Account, error) {
            return createAccount(ctx, in.UserEmail)
        })
        if err != nil {
            return OnboardingOutput{}, err
        }
        if _, err := agnt5.Step(ctx, "send_welcome_email", func(context.Context) (string, error) {
            return sendWelcomeEmail(ctx, account.ID)
        }); err != nil {
            return OnboardingOutput{}, err
        }
        return OnboardingOutput{Status: "done", AccountID: account.ID}, nil
    },
)
```

The handler's first parameter is always `*agnt5.Context`. The second is your workflow's typed input.

`RegisterWorkflow[In, Out any](w *Worker, name string, handler func(*Context, In) (Out, error), opts ...ComponentOption) error` accepts the same `ComponentOption`s as functions, plus:

| Option | Description |
|---|---|
| `WithTriggers(...TriggerSpec)` | Event or webhook triggers that start this workflow automatically |
| `WithCron(expression string)` | Cron expression for scheduled execution |



---

## Steps: the unit of durable work



**Python:**

Use `ctx.step()` to call a `@function` inside a workflow. The result is checkpointed after the first successful run. On replay (after a restart), the cached result is returned immediately. The function is not called again.

```python
from agnt5 import workflow, WorkflowContext, function, FunctionContext

@function
async def fetch_user(ctx: FunctionContext, user_id: str) -> dict:
    # call your database here
    return {"id": user_id, "name": "Ada"}

@function
async def send_email(ctx: FunctionContext, user: dict, subject: str) -> str:
    # call your email provider here
    return f"Sent to {user['name']}"

@workflow
async def notify_workflow(ctx: WorkflowContext, user_id: str) -> str:
    user   = await ctx.step(fetch_user, user_id)
    result = await ctx.step(send_email, user, "Welcome!")
    return result
```

Calling `await fetch_user(ctx, user_id)` directly also works, but that result is **not** checkpointed. The function re-runs every time the workflow replays.

| Call style | Checkpointed | Use when |
|---|---|---|
| `await ctx.step(fn, *args)` | Yes | Always prefer this inside a workflow |
| `await fn(ctx, *args)` | No | One-off calls where replay is fine |





**TypeScript:**

Functions built with `fn(name).run(...)` are already checkpoint-aware — call them directly with `ctx` as the first argument, no separate step wrapper needed.

```typescript
const fetchUser = fn('fetch_user').run(async (ctx: Context, input: { userId: string }) => {
  // call your database here
  return { id: input.userId, name: 'Ada' };
});

const sendEmail = fn('send_email').run(
  async (ctx: Context, input: { user: { id: string; name: string }; subject: string }) => {
    // call your email provider here
    return `Sent to ${input.user.name}`;
  },
);

export const notifyWorkflow = workflow(
  'notify_workflow',
  async (ctx: Context, input: { userId: string }) => {
    const user = await fetchUser(ctx, { userId: input.userId });
    const result = await sendEmail(ctx, { user, subject: 'Welcome!' });
    return result;
  },
);
```

Every call to a `fn(...).run(...)`-built function is checkpointed automatically when `ctx` is passed through — there is no separate `ctx.step()` to remember, unlike the Python SDK.





**Go:**

Use `agnt5.Step` to wrap work inside a workflow. Unlike Python, `Step` doesn't take a reference to a separately-registered function — it wraps any closure, and the closure's returned value is what gets checkpointed. On replay, a completed step's saved result is returned immediately and the closure is not run again.

```go
func fetchUser(ctx *agnt5.Context, userID string) (User, error) {
    // call your database here
    return User{ID: userID, Name: "Ada"}, nil
}

func sendEmail(ctx *agnt5.Context, user User, subject string) (string, error) {
    // call your email provider here
    return "Sent to " + user.Name, nil
}

err := agnt5.RegisterWorkflow(worker, "notify_workflow", func(ctx *agnt5.Context, in NotifyInput) (string, error) {
    user, err := agnt5.Step(ctx, "fetch_user", func(context.Context) (User, error) {
        return fetchUser(ctx, in.UserID)
    })
    if err != nil {
        return "", err
    }
    return agnt5.Step(ctx, "send_email", func(context.Context) (string, error) {
        return sendEmail(ctx, user, "Welcome!")
    })
})
```

Calling `fetchUser(ctx, in.UserID)` directly, outside a `Step`, also works, but that result is **not** checkpointed — it re-runs every time the workflow replays.

| Call style | Checkpointed | Use when |
|---|---|---|
| `agnt5.Step(ctx, "name", func(context.Context) (T, error) { ... })` | Yes | Always prefer this inside a workflow |
| Calling the function directly | No | One-off calls where replay is fine |



---

## Running in parallel



**Python:**

### `ctx.parallel()`: run tasks concurrently, collect in order

```python
@workflow
async def report_workflow(ctx: WorkflowContext, report_id: str) -> dict:
    sales, inventory, customers = await ctx.parallel(
        ctx.step(fetch_sales, report_id),
        ctx.step(fetch_inventory, report_id),
        ctx.step(fetch_customers, report_id),
    )
    return await ctx.step(compile_report, sales, inventory, customers)
```

Results come back in the same order as the tasks.

### `ctx.gather()`: run tasks concurrently, collect by name

```python
@workflow
async def dashboard_workflow(ctx: WorkflowContext) -> dict:
    data = await ctx.gather(
        revenue=fetch_revenue(),
        users=fetch_active_users(),
        errors=fetch_error_rate(),
    )
    # data["revenue"], data["users"], data["errors"]
    return data
```

### `ctx.batch()`: run a function across many inputs

Use `batch()` when you have a list of items and want to process them in parallel with controlled concurrency.

```python
@function
async def process_document(ctx: FunctionContext, doc_id: str) -> dict:
    # process one document
    return {"doc_id": doc_id, "status": "processed"}

@workflow
async def bulk_processor(ctx: WorkflowContext, doc_ids: list[str]) -> dict:
    result = await ctx.batch(
        process_document,
        [{"doc_id": d} for d in doc_ids],
        max_concurrency=20,
    )
    return {
        "processed": result.stats.completed_items,
        "failed":    result.stats.failed_items,
    }
```

`ctx.map()` is a simpler wrapper around `batch()`. It returns just the outputs and raises if any item fails:

```python
outputs = await ctx.map(process_document, [{"doc_id": d} for d in doc_ids])
```





**TypeScript:**

### `Promise.all`: run steps concurrently, collect in order

Because each `fn(...).run(...)` call checkpoints independently, plain `Promise.all` over function calls is already durable — a restart resumes only the calls that hadn't completed.

```typescript
export const reportWorkflow = workflow(
  'report_workflow',
  async (ctx: Context, input: { reportId: string }) => {
    const [sales, inventory, customers] = await Promise.all([
      fetchSales(ctx, { reportId: input.reportId }),
      fetchInventory(ctx, { reportId: input.reportId }),
      fetchCustomers(ctx, { reportId: input.reportId }),
    ]);
    return compileReport(ctx, { sales, inventory, customers });
  },
);
```

### `gather()`: run tasks concurrently, collect by name

`gather` (from `@agnt5/sdk`) is a thin wrapper over `Promise.all` that returns a named object instead of a positional array.

```typescript
export const dashboardWorkflow = workflow('dashboard_workflow', async (ctx: Context) => {
  const data = await gather({
    revenue: fetchRevenue(ctx, {}),
    users: fetchActiveUsers(ctx, {}),
    errors: fetchErrorRate(ctx, {}),
  });
  // data.revenue, data.users, data.errors
  return data;
});
```

### Fan-out over a list of inputs

There's no dedicated `batch()`/`map()` helper for functions in the TypeScript SDK yet — use `.map()` with `Promise.all`, the same fan-out pattern used in the quickstart template, and batch manually if you need to cap concurrency.

```typescript
const processDocument = fn('process_document').run(
  async (ctx: Context, input: { docId: string }) => {
    // process one document
    return { docId: input.docId, status: 'processed' };
  },
);

export const bulkProcessor = workflow(
  'bulk_processor',
  async (ctx: Context, input: { docIds: string[] }) => {
    const results = await Promise.all(
      input.docIds.map((docId) => processDocument(ctx, { docId })),
    );
    return {
      processed: results.filter((r) => r.status === 'processed').length,
    };
  },
);
```





**Go:**

There's no `ctx.parallel()`/`ctx.gather()`/`ctx.batch()`/`ctx.map()` equivalent in the Go SDK yet — no durable fan-out helper exists. For concurrent, non-durable work, use plain goroutines and wrap the whole group in a single `agnt5.Step` so it's checkpointed as one unit:

```go
err := agnt5.RegisterWorkflow(worker, "report_workflow", func(ctx *agnt5.Context, in ReportInput) (Report, error) {
    return agnt5.Step(ctx, "fetch_all", func(context.Context) (Report, error) {
        var wg sync.WaitGroup
        var sales, inventory, customers Result
        var salesErr, inventoryErr, customersErr error

        wg.Add(3)
        go func() { defer wg.Done(); sales, salesErr = fetchSales(ctx, in.ReportID) }()
        go func() { defer wg.Done(); inventory, inventoryErr = fetchInventory(ctx, in.ReportID) }()
        go func() { defer wg.Done(); customers, customersErr = fetchCustomers(ctx, in.ReportID) }()
        wg.Wait()

        if err := errors.Join(salesErr, inventoryErr, customersErr); err != nil {
            return Report{}, err
        }
        return compileReport(sales, inventory, customers), nil
    })
})
```

Because the whole fan-out lives inside one `Step`, a restart re-runs all three fetches together rather than resuming only the ones that hadn't finished — coarser-grained than Python's per-task checkpointing. `agnt5.Client.Batch` exists, but it's a remote gateway HTTP call for invoking a deployed component from *outside* a workflow, not an in-workflow fan-out primitive — don't use it as a substitute for `ctx.batch()`.



---

## Durable sleep



**Python:**

`ctx.sleep()` pauses the workflow for a set duration. Unlike `asyncio.sleep()`, it survives restarts. If the worker crashes during the wait, the workflow resumes and only sleeps for the time that remains.

```python
@workflow
async def follow_up_workflow(ctx: WorkflowContext, user_id: str) -> str:
    await ctx.step(send_confirmation, user_id)

    await ctx.sleep(24 * 60 * 60, name="wait_24h")   # wait 24 hours

    await ctx.step(send_follow_up, user_id)
    return "Follow-up sent."
```





**TypeScript:**

`ctx.sleep()` pauses the workflow for a set duration, in **milliseconds**. It is implemented as an in-process timer, so it is **not** restart-safe: if the worker process restarts during the wait, the timer is lost and the sleep does not resume. Reserve it for short waits, and avoid relying on it across long durations in workers that may restart.

```typescript
export const followUpWorkflow = workflow(
  'follow_up_workflow',
  async (ctx: Context, input: { userId: string }) => {
    await sendConfirmation(ctx, { userId: input.userId });

    await ctx.sleep(24 * 60 * 60 * 1000, 'wait_24h'); // wait 24 hours — see caveat above

    await sendFollowUp(ctx, { userId: input.userId });
    return 'Follow-up sent.';
  },
);
```





**Go:**

There's no `ctx.Sleep()` in the Go SDK yet — no durable sleep primitive exists. A plain `time.Sleep` works but is **not** restart-safe: if the worker process restarts during the wait, the sleep is lost and does not resume from where it left off. Reserve it for short waits, and don't rely on it across long durations.

```go
err := agnt5.RegisterWorkflow(worker, "follow_up_workflow", func(ctx *agnt5.Context, in FollowUpInput) (string, error) {
    if _, err := agnt5.Step(ctx, "send_confirmation", func(context.Context) (string, error) {
        return sendConfirmation(ctx, in.UserID)
    }); err != nil {
        return "", err
    }

    time.Sleep(24 * time.Hour) // not restart-safe — see caveat above

    return agnt5.Step(ctx, "send_follow_up", func(context.Context) (string, error) {
        return sendFollowUp(ctx, in.UserID)
    })
})
```



---

## State



**Python:**

Workflows have access to three state scopes through `ctx`.

```python
@workflow
async def stateful_workflow(ctx: WorkflowContext, user_id: str) -> dict:
    # Run-scoped: lives for this run only
    await ctx.state.set("phase", "started")

    # Session-scoped: persists across turns for the same session
    count = await ctx.session.state.get("visit_count", 0)
    await ctx.session.state.set("visit_count", count + 1)

    return {"visits": count + 1}
```

| Scope | Access | Persists |
|---|---|---|
| Run | `ctx.state` | Current run only. Cleared when the run finishes |
| Session | `ctx.session.state` | Across multiple runs with the same `session_id` |





**TypeScript:**

Workflows have access to run-scoped state directly on `ctx`.

```typescript
export const statefulWorkflow = workflow(
  'stateful_workflow',
  async (ctx: Context, input: { userId: string }) => {
    // Run-scoped: lives for this run only
    await ctx.set('phase', 'started');
    const phase = await ctx.get<string>('phase');

    return { phase };
  },
);
```

| Scope | Access | Persists |
|---|---|---|
| Run | `ctx.get(key)` / `ctx.set(key, value)` | Current run only. Cleared when the run finishes |

Session-scoped state (`ctx.session`, persisting across multiple runs with the same session ID) is not yet exposed on `Context` in the TypeScript SDK — coming soon.





**Go:**

`ctx.State()` returns a run-scoped `*StateManager` by default. Call `.Scope(...)` to switch scopes.

```go
// toInt handles both a fresh in-process int and a value that round-tripped
// through the state store's JSON encoding, which decodes numbers as float64.
func toInt(v any) int {
    switch n := v.(type) {
    case int:
        return n
    case float64:
        return int(n)
    default:
        return 0
    }
}

err := agnt5.RegisterWorkflow(worker, "stateful_workflow", func(ctx *agnt5.Context, in StatefulInput) (StatefulOutput, error) {
    // Run-scoped: lives for this run only
    if err := ctx.State().Set(ctx, "phase", "started"); err != nil {
        return StatefulOutput{}, err
    }

    // Session-scoped: persists across runs sharing the same namespace
    sessionState := ctx.State().Scope(agnt5.StateScopeSession, "visits")
    raw, err := sessionState.Get(ctx, "visit_count")
    if err != nil && !errors.Is(err, agnt5.ErrStateNotFound) {
        return StatefulOutput{}, err
    }
    count := toInt(raw) + 1

    if err := sessionState.Set(ctx, "visit_count", count); err != nil {
        return StatefulOutput{}, err
    }

    return StatefulOutput{Visits: count}, nil
})
```

| Scope | Access | Persists |
|---|---|---|
| Run | `ctx.State()` | Current run only. Cleared when the run finishes |
| Session / User / Global | `ctx.State().Scope(agnt5.StateScopeSession\|StateScopeUser\|StateScopeGlobal, namespace)` | Across runs sharing that scope and namespace |

`agnt5.ErrStateNotFound` is the sentinel returned by `Get` on a missing key — check for it with `errors.Is`, and propagate any other error instead of ignoring it. Numeric values read back from the state store decode as `float64` (they round-trip through JSON), not the `int` you originally stored, so convert before doing arithmetic — `toInt` above handles both the first-time (`nil`/zero-value) case and the persisted case.



---

## Triggering workflows



**Python:**

### Event trigger

Start a workflow automatically when a named event fires.

```python
from agnt5 import workflow, WorkflowContext
from agnt5.types import event

@workflow(triggers=[event("user.signed_up")])
async def welcome_workflow(ctx: WorkflowContext, user_id: str) -> str:
    await ctx.step(send_welcome_email, user_id)
    return "Welcome email sent."
```

### Webhook trigger

Start a workflow from an incoming webhook.

```python
from agnt5.types import webhook

@workflow(triggers=[webhook("stripe", event="payment_intent.succeeded")])
async def payment_workflow(ctx: WorkflowContext, amount: int, currency: str) -> str:
    await ctx.step(record_payment, amount, currency)
    return "Payment recorded."
```





**TypeScript:**

### Event trigger

Start a workflow automatically when a named event fires.

```typescript
export const welcomeWorkflow = workflow(
  'welcome_workflow',
  async (ctx: Context, input: { userId: string }) => {
    await sendWelcomeEmail(ctx, { userId: input.userId });
    return 'Welcome email sent.';
  },
  { triggers: [event('user.signed_up')] },
);
```

### Webhook trigger

Start a workflow from an incoming webhook.

```typescript
export const paymentWorkflow = workflow(
  'payment_workflow',
  async (ctx: Context, input: { amount: number; currency: string }) => {
    await recordPayment(ctx, { amount: input.amount, currency: input.currency });
    return 'Payment recorded.';
  },
  { triggers: [webhook('stripe', { event: 'payment_intent.succeeded' })] },
);
```





**Go:**

### Event trigger

Start a workflow automatically when a named event fires, using `WithTriggers` and `EventTrigger`.

```go
err := agnt5.RegisterWorkflow(worker, "welcome_workflow",
    func(ctx *agnt5.Context, in WelcomeInput) (string, error) {
        return sendWelcomeEmail(ctx, in.UserID)
    },
    agnt5.WithTriggers(agnt5.EventTrigger("user.signed_up")),
)
```

### Webhook trigger

Start a workflow from an incoming webhook, using `WebhookTrigger(source, event)`.

```go
err := agnt5.RegisterWorkflow(worker, "payment_workflow",
    func(ctx *agnt5.Context, in map[string]any) (string, error) {
        amount, _ := in["amount"].(float64)
        currency, _ := in["currency"].(string)
        return recordPayment(ctx, amount, currency)
    },
    agnt5.WithTriggers(agnt5.WebhookTrigger("stripe", "payment_intent.succeeded")),
)
```

Webhook-triggered handlers receive the provider's payload as an untyped `map[string]any` — there's no typed webhook-envelope struct in the Go SDK, so extract fields manually. See [Webhooks](/docs/build/webhooks.md) for more on shaping these inputs.



---

## Human-in-the-loop



**Python:**

Pause a workflow and wait for user input using `ctx.wait_for_user()`. See [Human-in-the-loop](/docs/build/human-in-the-loop.md) for the full reference.

```python
@workflow
async def approval_workflow(ctx: WorkflowContext, order_id: str) -> dict:
    summary = await ctx.step(prepare_order_summary, order_id)

    decision = await ctx.wait_for_user(
        question=f"Approve this order?\n\n{summary}",
        input_type="approval",
        options=[
            {"id": "approve", "label": "Approve"},
            {"id": "reject",  "label": "Reject"},
        ],
    )

    if decision == "reject":
        return {"status": "rejected"}

    result = await ctx.step(fulfil_order, order_id)
    return result
```





**TypeScript:**

Pause a workflow and wait for user input using `ctx.waitForUser()`. See [Human-in-the-loop](/docs/build/human-in-the-loop.md) for the full reference.

```typescript
export const approvalWorkflow = workflow(
  'approval_workflow',
  async (ctx: Context, input: { orderId: string }) => {
    const summary = await prepareOrderSummary(ctx, { orderId: input.orderId });

    const decision = await ctx.waitForUser(`Approve this order?\n\n${summary}`, {
      inputType: 'approval',
      options: [
        { id: 'approve', label: 'Approve' },
        { id: 'reject', label: 'Reject' },
      ],
    });

    if (decision === 'reject') {
      return { status: 'rejected' };
    }

    return fulfilOrder(ctx, { orderId: input.orderId });
  },
);
```





**Go:**

Pause a workflow and wait for user input using `ctx.AskUser()` or the `ctx.RequestApproval()` convenience wrapper. See [Human-in-the-loop](/docs/build/human-in-the-loop.md) for the full reference.

```go
err := agnt5.RegisterWorkflow(worker, "approval_workflow", func(ctx *agnt5.Context, in ApprovalInput) (ApprovalOutput, error) {
    summary, err := agnt5.Step(ctx, "prepare_order_summary", func(context.Context) (string, error) {
        return prepareOrderSummary(ctx, in.OrderID)
    })
    if err != nil {
        return ApprovalOutput{}, err
    }

    approved, err := ctx.RequestApproval("Approve this order?\n\n"+summary, nil)
    if err != nil {
        return ApprovalOutput{}, err
    }
    if !approved {
        return ApprovalOutput{Status: "rejected"}, nil
    }

    return fulfilOrder(ctx, in.OrderID)
})
```



---

## Real-world example

An order processing pipeline: validate the order, charge the customer, and send a confirmation, all as durable steps. If any step fails, the workflow retries from that point without re-running earlier steps.



**Python:**

```python
from agnt5 import workflow, WorkflowContext, function, FunctionContext


@function
async def validate_order(ctx: FunctionContext, order_id: str) -> dict:
    # check stock, validate address, etc.
    return {"order_id": order_id, "total": 89.99, "valid": True}


@function
async def charge_customer(ctx: FunctionContext, order_id: str, amount: float) -> dict:
    # call your payments API
    return {"order_id": order_id, "charge_id": "ch_abc123", "status": "paid"}


@function
async def send_confirmation(ctx: FunctionContext, order_id: str, charge_id: str) -> str:
    # send confirmation email
    return f"Confirmation sent for order {order_id}"


@workflow
async def order_workflow(ctx: WorkflowContext, order_id: str) -> dict:
    # Step 1: validate
    order = await ctx.step(validate_order, order_id)

    if not order["valid"]:
        return {"status": "invalid", "order_id": order_id}

    # Step 2: charge (checkpointed, never charged twice even if workflow restarts)
    charge = await ctx.step(charge_customer, order_id, order["total"])

    # Step 3: confirm
    await ctx.step(send_confirmation, order_id, charge["charge_id"])

    return {"status": "complete", "order_id": order_id, "charge_id": charge["charge_id"]}
```

Each step is checkpointed. If the worker restarts between step 2 and step 3, the charge is not made again. The saved result is used and execution continues at step 3.





**TypeScript:**

```typescript
const validateOrder = fn('validate_order').run(
  async (ctx: Context, input: { orderId: string }) => {
    // check stock, validate address, etc.
    return { orderId: input.orderId, total: 89.99, valid: true };
  },
);

const chargeCustomer = fn('charge_customer').run(
  async (ctx: Context, input: { orderId: string; amount: number }) => {
    // call your payments API
    return { orderId: input.orderId, chargeId: 'ch_abc123', status: 'paid' };
  },
);

const sendConfirmation = fn('send_confirmation').run(
  async (ctx: Context, input: { orderId: string; chargeId: string }) => {
    // send confirmation email
    return `Confirmation sent for order ${input.orderId}`;
  },
);

export const orderWorkflow = workflow(
  'order_workflow',
  async (ctx: Context, input: { orderId: string }) => {
    // Step 1: validate
    const order = await validateOrder(ctx, { orderId: input.orderId });

    if (!order.valid) {
      return { status: 'invalid', orderId: input.orderId };
    }

    // Step 2: charge (checkpointed, never charged twice even if workflow restarts)
    const charge = await chargeCustomer(ctx, { orderId: input.orderId, amount: order.total });

    // Step 3: confirm
    await sendConfirmation(ctx, { orderId: input.orderId, chargeId: charge.chargeId });

    return { status: 'complete', orderId: input.orderId, chargeId: charge.chargeId };
  },
);
```

Each function call is checkpointed. If the worker restarts between step 2 and step 3, the charge is not made again. The saved result is used and execution continues at step 3.





**Go:**

```go
import (
    "github.com/agnt5dev/sdk-go/agnt5"
    "context"
)

func validateOrder(ctx *agnt5.Context, orderID string) (Order, error) {
    // check stock, validate address, etc.
    return Order{ID: orderID, Total: 89.99, Valid: true}, nil
}

func chargeCustomer(ctx *agnt5.Context, orderID string, amount float64) (Charge, error) {
    // call your payments API
    return Charge{OrderID: orderID, ChargeID: "ch_abc123", Status: "paid"}, nil
}

func sendConfirmation(ctx *agnt5.Context, orderID, chargeID string) (string, error) {
    // send confirmation email
    return "Confirmation sent for order " + orderID, nil
}

err := agnt5.RegisterWorkflow(worker, "order_workflow", func(ctx *agnt5.Context, in OrderInput) (OrderResult, error) {
    // Step 1: validate
    order, err := agnt5.Step(ctx, "validate_order", func(context.Context) (Order, error) {
        return validateOrder(ctx, in.OrderID)
    })
    if err != nil {
        return OrderResult{}, err
    }
    if !order.Valid {
        return OrderResult{Status: "invalid", OrderID: in.OrderID}, nil
    }

    // Step 2: charge (checkpointed, never charged twice even if workflow restarts)
    charge, err := agnt5.Step(ctx, "charge_customer", func(context.Context) (Charge, error) {
        return chargeCustomer(ctx, in.OrderID, order.Total)
    })
    if err != nil {
        return OrderResult{}, err
    }

    // Step 3: confirm
    if _, err := agnt5.Step(ctx, "send_confirmation", func(context.Context) (string, error) {
        return sendConfirmation(ctx, in.OrderID, charge.ChargeID)
    }); err != nil {
        return OrderResult{}, err
    }

    return OrderResult{Status: "complete", OrderID: in.OrderID, ChargeID: charge.ChargeID}, nil
})
```

Each `Step` is checkpointed. If the worker restarts between step 2 and step 3, the charge is not made again. The saved result is used and execution continues at step 3.