> 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-10-06
---



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 |
| `cron` | `str` | `None` | Cron expression for scheduled execution, for example `@workflow(cron="0 9 * * *")` |





**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 ctx.step('create_account', () => createAccount(ctx, { userEmail: input.userEmail }));
    await ctx.step('send_welcome_email', () => 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
```

`ctx.step` also checkpoints async work that isn't a `@function`. Give the step a name and pass the awaitable:

```python
data = await ctx.step("load_data", fetch_expensive_data())
```





**TypeScript:**

Use `ctx.step(name, fn)` to run a function inside a workflow. After the runtime accepts the step completion, replay returns the committed result instead of calling the function again.

```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 ctx.step('fetch_user', () => fetchUser(ctx, { userId: input.userId }));
    const result = await ctx.step('send_email', () => sendEmail(ctx, { user, subject: 'Welcome!' }));
    return result;
  },
);
```

The step name identifies the step in the run's trace. When a workflow runs the same step in a loop or in parallel, pass `{ key }` as the third argument to give each one a stable identity, for example `{ key: 'user-42' }`.

### `ctx.step()`: checkpoint your own code

For work that isn't a `fn(...)` function, such as an SDK call, a query, or a value you compute, wrap it in `ctx.step(name, body)`. Its result is checkpointed the same way.

Give the step the values it works on as `input`, and the body receives them. Replay then also checks the input. A step that recomputes a different input fails the run with `NON_DETERMINISTIC_REPLAY` instead of returning a result recorded for other values:

```typescript
export const checkoutWorkflow = workflow(
  'checkout_workflow',
  async (ctx: Context, input: { orderId: string; amount: number }) => {
    const receipt = await ctx.step(
      'charge',
      ({ orderId, amount }) => chargeCard(orderId, amount),
      { input: { orderId: input.orderId, amount: input.amount } },
    );
    return receipt;
  },
);
```

The body's parameter is typed from `input`, and TypeScript requires `input` once the body declares a parameter. A step without `input` isn't checked this way: if a value its body captures changes, replay still returns the recorded result.

- **Derive the input from the workflow's input or earlier step results.** `Date.now()`, a random ID, or anything else that differs between executions fails the run at its first replay.
- **Pass plain data.** The step copies the input when you call it and runs the body on that copy. `undefined` properties are dropped, so pass `value ?? null` for something that may be missing. Pass a date as a string (`toISOString()`); a `Date`, `Map`, `Set`, function, or class instance is rejected.
- **Keep secrets out.** The input is recorded with the step and shown in the run's timeline. Pass a secret's name, not its value.

Managed workers check the input on replay. [Serverless endpoints](/docs/run/serverless-endpoints.md) pass it to the body without checking it. Step input needs `@agnt5/sdk` 0.10.13 or later.





**Go:**

Use `agnt5.Step` to wrap work inside a workflow. `Step` wraps a 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!")
    })
})
```





**Python:**

> **Make steps with external effects idempotent.** AGNT5 records one accepted completion for a durable step. If a worker fails before that completion is committed, the step body runs again. For external effects such as charging a card or sending a message, read `ctx.activation.idempotency_key` inside the step function and pass it to the downstream system, or use a transactional outbox.





**TypeScript:**

> **Make steps with external effects idempotent.** AGNT5 records one accepted completion for a durable step. If a worker fails before that completion is committed, the step body runs again. For external effects such as charging a card or sending a message, read `ctx.activation?.idempotencyKey` inside the step and pass it to the downstream system, or use a transactional outbox.





**Go:**

> **Make steps with external effects idempotent.** AGNT5 records one accepted completion for a durable step. If a worker fails before that completion is committed, the step body runs again. For external effects such as charging a card or sending a message, call `agnt5.ActivationFromContext(stepCtx)` with the `context.Context` the step closure receives and pass its `IdempotencyKey` 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=ctx.step(fetch_revenue),
        users=ctx.step(fetch_active_users),
        errors=ctx.step(fetch_error_rate),
    )
    # data["revenue"], data["users"], data["errors"]
    return data
```





**TypeScript:**

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

Each `ctx.step` call checkpoints independently, so plain `Promise.all` over steps is durable: a restart resumes only the steps that hadn't completed.

```typescript
export const reportWorkflow = workflow(
  'report_workflow',
  async (ctx: Context, input: { reportId: string }) => {
    const [sales, inventory, customers] = await Promise.all([
      ctx.step('fetch_sales', () => fetchSales(ctx, { reportId: input.reportId })),
      ctx.step('fetch_inventory', () => fetchInventory(ctx, { reportId: input.reportId })),
      ctx.step('fetch_customers', () => fetchCustomers(ctx, { reportId: input.reportId })),
    ]);
    return ctx.step('compile_report', () => 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: ctx.step('fetch_revenue', () => fetchRevenue(ctx, {})),
    users: ctx.step('fetch_active_users', () => fetchActiveUsers(ctx, {})),
    errors: ctx.step('fetch_error_rate', () => fetchErrorRate(ctx, {})),
  });
  // data.revenue, data.users, data.errors
  return data;
});
```

### Fan-out over a list of inputs

Map over the list and give each step a stable key, so every item is checkpointed on its own.

```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) =>
        ctx.step('process_document', () => processDocument(ctx, { docId }), { key: `doc-${docId}` }),
      ),
    );
    return {
      processed: results.filter((r) => r.status === 'processed').length,
    };
  },
);
```

### Child workflows

Run another workflow from inside a workflow with `executeChildWorkflow`, or run the same workflow over a list of inputs with `fanOut`.

```typescript
import { workflow, executeChildWorkflow, fanOut } from '@agnt5/sdk';

export const doubleWorkflow = workflow('double', async (ctx: Context, input: { n: number }) => ({
  doubled: input.n * 2,
}));

export const parentWorkflow = workflow('parent', async (ctx: Context, input: { ns: number[] }) => {
  const one = await executeChildWorkflow(ctx, 'double', { n: 21 });
  const many = await fanOut(ctx, 'double', input.ns.map((n) => ({ n })));
  return { one, many }; // { one: { doubled: 42 }, many: [{ doubled: 2 }, ...] }
});
```





**Go:**

Run concurrent work in 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
    })
})
```

The three fetches succeed or are replayed together as one step. To checkpoint items individually, call `agnt5.StepWithKey` once per item with a stable key.




> 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 ctx.step('send_confirmation', () => sendConfirmation(ctx, { userId: input.userId }));

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

    await ctx.step('send_follow_up', () => 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)
    })
})
```



---

## 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
    ctx.state.set("phase", "started")
    phase = ctx.state.get("phase")

    # 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 {"phase": phase, "visits": count + 1}
```

| Scope | Access | Persists |
|---|---|---|
| Run | `ctx.state.get(key, default)` / `ctx.state.set(key, value)` | Current run only. Cleared when the run finishes |
| Session | `await ctx.session.state.get(key, default)` / `await ctx.session.state.set(key, value)` | Across multiple runs with the same `session_id` |

Run-scoped `ctx.state` reads and writes are synchronous. Use `await ctx.state.set_async(key, value)` when you want to await the write.





**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');
    const fallback = await ctx.get<string>('missing', 'default value');

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

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





**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
    }
    phase, err := ctx.State().GetString(ctx, "phase")
    if 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{Phase: phase, 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, event: dict, **_) -> str:
    user_id = event["data"]["user_id"]  # the published event's payload
    await ctx.step(send_welcome_email, user_id)
    return "Welcome email sent."
```

### Webhook trigger

Start a workflow from an incoming webhook.

```python
import json
from agnt5.types import webhook

@workflow(triggers=[webhook("stripe", event="payment_intent.succeeded")])
async def payment_workflow(ctx: WorkflowContext, event: dict, **_) -> str:
    payload = json.loads(event["data"]["body"])  # raw request body, as a string
    intent = payload["data"]["object"]
    await ctx.step(record_payment, intent["amount"], intent["currency"])
    return "Payment recorded."
```

Both triggers deliver the gateway's trigger envelope as keyword arguments: `event` (with `id`, `name`, `data`, `source`, and `timestamp_ns`) plus `deployment_id`, `target_kind`, and `target_ref`, so declare `event` and accept the rest with `**_`. For `event()` triggers, `event["data"]` is the published payload; for `webhook()` triggers, it is the webhook envelope whose `body` is the raw request body. See [Webhooks → What your workflow receives](/docs/build/webhooks.md#what-your-workflow-receives).





**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: { event: { data: { userId: string } } }) => {
    const { userId } = input.event.data; // the published event's payload
    await ctx.step('send_welcome_email', () => sendWelcomeEmail(ctx, { 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: { event: { data: { body: string } } }) => {
    const payload = JSON.parse(input.event.data.body); // raw request body, as a string
    const { amount, currency } = payload.data.object;
    await ctx.step('record_payment', () => recordPayment(ctx, { amount, currency }));
    return 'Payment recorded.';
  },
  { triggers: [webhook('stripe', { event: 'payment_intent.succeeded' })] },
);
```

Both triggers deliver the gateway's trigger envelope as the handler input: `{ event: { id, name, data, source, timestamp_ns }, deployment_id, target_kind, target_ref }`. For `event()` triggers, `event.data` is the published payload; for `webhook()` triggers, it is the webhook envelope whose `body` is the raw request body. See [Webhooks → What your workflow receives](/docs/build/webhooks.md#what-your-workflow-receives).





**Go:**

### Event trigger

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

```go
type WelcomeInput struct {
    Event struct {
        Data struct {
            UserID string `json:"user_id"` // the published event's payload
        } `json:"data"`
    } `json:"event"`
}

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

### Webhook trigger

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

```go
type WebhookInput struct {
    Event struct {
        Data struct {
            Body string `json:"body"` // raw request body, as a string
        } `json:"data"`
    } `json:"event"`
}

err := agnt5.RegisterWorkflow(worker, "payment_workflow",
    func(ctx *agnt5.Context, in WebhookInput) (string, error) {
        var payload struct {
            Data struct {
                Object struct {
                    Amount   int64  `json:"amount"`
                    Currency string `json:"currency"`
                } `json:"object"`
            } `json:"data"`
        }
        if err := json.Unmarshal([]byte(in.Event.Data.Body), &payload); err != nil {
            return "", err
        }
        return recordPayment(ctx, payload.Data.Object.Amount, payload.Data.Object.Currency)
    },
    agnt5.WithTriggers(agnt5.WebhookTrigger("stripe", "payment_intent.succeeded")),
)
```

Both triggers deliver the gateway's trigger envelope, `{ "event": { "id", "name", "data", "source", "timestamp_ns" }, "deployment_id", "target_kind", "target_ref" }`, not the payload directly. Declare a struct that mirrors the parts you need, as above. For `EventTrigger`, `event.data` is the published payload; for `WebhookTrigger`, it is the webhook envelope whose `body` is the raw request body. See [Webhooks → What your workflow receives](/docs/build/webhooks.md#what-your-workflow-receives).



---

## 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 ctx.step('prepare_order_summary', () => 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 ctx.step('fulfil_order', () => 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, sending ctx.activation.idempotency_key as its idempotency key
    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
    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. Send the step's activation idempotency key to 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, sending ctx.activation?.idempotencyKey as its idempotency key
    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 ctx.step('validate_order', () => validateOrder(ctx, { orderId: input.orderId }));

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

    // Step 2: charge
    const charge = await ctx.step('charge_customer', () =>
      chargeCustomer(ctx, { orderId: input.orderId, amount: order.total }),
    );

    // Step 3: confirm
    await ctx.step('send_confirmation', () =>
      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. Send the step's activation idempotency key to 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, idempotencyKey string) (Charge, error) {
    // call your payments API, sending idempotencyKey as its idempotency key
    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, using the step's activation idempotency key
    charge, err := agnt5.Step(ctx, "charge_customer", func(stepCtx context.Context) (Charge, error) {
        act, _ := agnt5.ActivationFromContext(stepCtx)
        return chargeCustomer(ctx, in.OrderID, order.Total, act.IdempotencyKey)
    })
    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. Send the step's activation idempotency key to the provider, or write the intent through a transactional outbox.



## Next steps

- [Human-in-the-loop](/docs/build/human-in-the-loop.md): pause a workflow for approvals and input.
- [Webhooks](/docs/build/webhooks.md): start workflows from external events.
- [Agents](/docs/build/agents.md): call an agent from a workflow step.
- [Deploying](/docs/run/deploying.md): move your workflows into an AGNT5 environment.