> 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-08-05
---



A **workflow** is a durable orchestrator that sequences functions, agents, and human approvals into a reliable pipeline. If a workflow crashes mid-run, it replays results whose completions the runtime already accepted and retries work that was not committed.

---

## 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
import { workflow } from '@agnt5/sdk';
import type { Context } from '@agnt5/sdk';
import { createAccount, sendWelcomeEmail } from './functions.js';

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. After the runtime accepts the step completion, replay returns the committed result instead of calling the function 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
import { workflow } from '@agnt5/sdk';
import type { Context } from '@agnt5/sdk';
import { fn } from '@agnt5/sdk';

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. After the runtime accepts the completion, replay returns the committed result. There is no separate `ctx.step()` to remember for this common case, 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. After the runtime accepts the completion, replay returns the saved result instead of running the closure 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 |




> **Durable replay is not exactly-once execution.** AGNT5 records one accepted completion for a durable step, but the step body can run again if a worker fails before that completion is committed. For external effects such as charging a card or sending a message, pass the activation idempotency key to the downstream system or use a transactional outbox.


---

## 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
import { gather } from '@agnt5/sdk';

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()`.




> In branches, loops, and fan-out, assign stable explicit step keys whenever source-order ordinals can change. Use `key=` with Python `ctx.step`, `{ key: ... }` with TypeScript `ctx.step`, or `agnt5.StepWithKey` in Go.


---

## Durable sleep



**Python:**

`ctx.sleep()` pauses the workflow for a set duration. When the runtime negotiates `durable_suspension_v1`, the timer survives worker restarts and resumes after the remaining delay.

```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**. When the runtime negotiates `durable_suspension_v1`, the timer survives worker restarts and resumes after the remaining delay.

```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

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





**Go:**

`ctx.Sleep()` pauses the workflow for a `time.Duration`. When the runtime negotiates `durable_suspension_v1`, the timer survives worker restarts and resumes after the remaining delay. Use `WithSleepKey` when control flow can change the timer's source-order position.

```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
    }

    if err := ctx.Sleep(24*time.Hour, agnt5.WithSleepKey("wait_24h")); err != nil {
        return "", err
    }

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




> Cross-crash durable sleep requires runtime capability `durable_suspension_v1`. Compatibility paths may use a process-local timer and do not carry that guarantee.


---

## 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
import { workflow, event } from '@agnt5/sdk';
import type { Context } from '@agnt5/sdk';

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
import { workflow, webhook } from '@agnt5/sdk';
import type { Context } from '@agnt5/sdk';

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 as durable steps. Committed results replay; external effects still need downstream deduplication because a failure can happen after the effect succeeds but before its completion is committed.



**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 (pass ctx.activation.idempotency_key to the payment provider)
    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"]}
```

If the runtime accepted step 2 before a restart, its saved charge result replays and execution continues at step 3. If the payment provider accepted the charge but AGNT5 did not commit the step completion, step 2 can run again. Use the activation idempotency key with the provider or write the intent through a transactional outbox.





**TypeScript:**

```typescript
import { fn, workflow } from '@agnt5/sdk';
import type { Context } from '@agnt5/sdk';

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: pass ctx.activation?.idempotencyKey to the payment provider
    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 };
  },
);
```

If the runtime accepted step 2 before a restart, its saved charge result replays and execution continues at step 3. If the payment provider accepted the charge but AGNT5 did not commit the function completion, step 2 can run again. Use the activation idempotency key with the provider or write the intent through a transactional outbox.





**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: pass the idempotency key returned by ctx.Activation() to the provider
    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
})
```

If the runtime accepted step 2 before a restart, its saved charge result replays and execution continues at step 3. If the payment provider accepted the charge but AGNT5 did not commit the `Step` completion, step 2 can run again. Use the activation idempotency key with the provider or write the intent through a transactional outbox.