> 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: Functions
description: Define stateless units of work that AGNT5 can register, retry, time out, and checkpoint inside workflows.
last_verified: 2026-07-24
---



A **function** is a stateless unit of work: it receives inputs, runs your logic, and returns a result. The platform handles registration, retries, and timeouts. When called from a workflow, the result is checkpointed so a restart never runs it twice.

## Creating a function



**Python:**

One thing is required: a decorated async (or sync) Python function.

```python
from agnt5 import function, FunctionContext

@function
async def send_email(ctx: FunctionContext, to: str, subject: str, body: str) -> str:
    # call your email provider here
    return f"Sent to {to}"
```

| Parameter | Type | Default | Description |
|---|---|---|---|
| `name` | `str` | `function.__name__` | How the function appears in the platform. |
| `retries` | `int \| RetryPolicy` | `None` | How many times to retry on failure. |
| `backoff` | `str \| BackoffPolicy` | `None` | How to space out retries: `"constant"`, `"linear"`, or `"exponential"`. |
| `timeout_ms` | `int` | `None` | Cut off the function after this many milliseconds. |

**Example with retries and timeout:**

```python
@function(name="send_email", retries=3, backoff="exponential", timeout_ms=10000)
async def send_email(ctx: FunctionContext, to: str, subject: str, body: str) -> str:
    ctx.logger.info("Sending email", to=to)
    # call your email provider here
    return f"Sent to {to}"
```

Sync functions work too. AGNT5 automatically runs them in a thread pool.





**Go:**

One thing is required: a handler function registered with `RegisterFunction`.

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

`fn(name)` returns a builder with chainable configuration methods, applied before `.run()`:

| Method | Description |
|---|---|
| `.retry(policy: RetryPolicy)` | How many times to retry on failure. |
| `.backoff(policy: BackoffPolicy)` | How to space out retries. |
| `.timeout(ms: number)` | Cut off the function after this many milliseconds. |

**Example with retries and timeout:**

```typescript
export const sendEmail = fn('send_email')
  .retry({ maxAttempts: 3 })
  .backoff({ type: 'exponential' })
  .timeout(10000)
  .run(async (ctx: Context, input: { to: string; subject: string; body: string }) => {
    ctx.logger.info('Sending email', { to: input.to });
    // call your email provider here
    return `Sent to ${input.to}`;
  });
```



## Function context



**Python:**

Name the first parameter `ctx` and AGNT5 injects a `FunctionContext` automatically.

```python
@function
async def send_email(ctx: FunctionContext, to: str, subject: str, body: str) -> str:
    ctx.logger.info("Sending email", to=to, attempt=ctx.attempt)
    # call your email provider here
    return f"Sent to {to}"
```

| Property / method | Description |
|---|---|
| `ctx.run_id` | Unique ID for this execution (useful for logging and tracing) |
| `ctx.attempt` | Which retry this is (0 = first try) |
| `ctx.logger` | Structured logger. Pass extra fields as keyword args: `ctx.logger.info("msg", key=value)` |
| `ctx.sleep(seconds)` | Pause execution without blocking the event loop |

If you don't need context, leave `ctx` out and the function still works fine.





**TypeScript:**

The handler's first parameter is always a `Context`, injected automatically by the runtime.

```typescript
export const sendEmail = fn('send_email').run(
  async (ctx: Context, input: { to: string; subject: string; body: string }) => {
    ctx.logger.info('Sending email', { to: input.to, attempt: ctx.attempt });
    // call your email provider here
    return `Sent to ${input.to}`;
  },
);
```

| Property / method | Description |
|---|---|
| `ctx.runId` | Unique ID for this execution (useful for logging and tracing) |
| `ctx.attempt` | Which retry this is (0 = first try) |
| `ctx.logger` | Structured logger. Pass extra fields as an object: `ctx.logger.info('msg', { key: value })` |
| `ctx.sleep(ms)` | Pause execution without blocking the event loop (milliseconds, not seconds) |





**Go:**

The handler's first parameter is always `*agnt5.Context` — Go has no equivalent to omitting it.

```go
func sendEmail(ctx *agnt5.Context, in EmailInput) (string, error) {
    ctx.Logger().Info("Sending email", "to", in.To, "attempt", ctx.Attempt())
    // call your email provider here
    return "Sent to " + in.To, nil
}
```

| Method | Description |
|---|---|
| `ctx.RunID()` | Unique ID for this execution (useful for logging and tracing) |
| `ctx.Attempt()` | Which retry this is (0 = first try) |
| `ctx.Logger()` | Structured logger. Pass extra fields as key/value pairs: `ctx.Logger().Info("msg", "key", value)` |

There's no `ctx.Sleep` yet — the Go SDK has no durable sleep primitive. A plain `time.Sleep` works but isn't checkpointed, so it re-runs its full duration on replay.



---

## Retries and backoff



**Python:**

### Shorthand

Pass an integer for retries and a string for the backoff strategy.

```python
@function(retries=3, backoff="exponential")
async def send_email(ctx: FunctionContext, to: str, subject: str, body: str) -> str:
    if ctx.attempt > 0:
        ctx.logger.info("Retrying email", to=to, attempt=ctx.attempt)
    # call your email provider here
    return f"Sent to {to}"
```

### Full control

Use `RetryPolicy` and `BackoffPolicy` for precise tuning.

```python
from agnt5 import function, FunctionContext
from agnt5.types import RetryPolicy, BackoffPolicy, BackoffType

@function(
    retries=RetryPolicy(max_attempts=5, initial_interval_ms=500, max_interval_ms=30000),
    backoff=BackoffPolicy(type=BackoffType.EXPONENTIAL, multiplier=2.0),
)
async def send_email(ctx: FunctionContext, to: str, subject: str, body: str) -> str:
    ctx.logger.info("Sending email", to=to, attempt=ctx.attempt)
    # call your email provider here
    return f"Sent to {to}"
```

`RetryPolicy` parameters:

| Parameter | Default | Description |
|---|---|---|
| `max_attempts` | `3` | Total tries, including the first |
| `initial_interval_ms` | `1000` | Wait before the first retry |
| `max_interval_ms` | `60000` | Maximum wait between retries |

`BackoffPolicy` parameters:

| Parameter | Default | Description |
|---|---|---|
| `type` | `EXPONENTIAL` | `CONSTANT` (fixed wait), `LINEAR` (grows steadily), `EXPONENTIAL` (doubles each time) |
| `multiplier` | `2.0` | How fast the wait grows |

AGNT5 runs your function body once per attempt. Use `ctx.attempt` if you need to vary behaviour on retries.





**TypeScript:**

The TypeScript SDK always takes a full `RetryPolicy`/`BackoffPolicy` object — there's no bare-number/string shorthand.

```typescript
export const sendEmail = fn('send_email')
  .retry({ maxAttempts: 5, initialIntervalMs: 500, maxIntervalMs: 30000 })
  .backoff({ type: 'exponential', multiplier: 2.0 })
  .run(async (ctx: Context, input: { to: string; subject: string; body: string }) => {
    ctx.logger.info('Sending email', { to: input.to, attempt: ctx.attempt });
    // call your email provider here
    return `Sent to ${input.to}`;
  });
```

`RetryPolicy` fields:

| Field | Default | Description |
|---|---|---|
| `maxAttempts` | `3` | Total tries, including the first |
| `initialIntervalMs` | `1000` | Wait before the first retry |
| `maxIntervalMs` | `60000` | Maximum wait between retries |

`BackoffPolicy` fields:

| Field | Default | Description |
|---|---|---|
| `type` | `'exponential'` | `'constant'` (fixed wait), `'linear'` (grows steadily), `'exponential'` (doubles each time) |
| `multiplier` | `2.0` | How fast the wait grows |

AGNT5 runs your function body once per attempt. Use `ctx.attempt` if you need to vary behaviour on retries.





**Go:**

`WithRetry` and `WithBackoff` always take plain arguments — there's no shorthand/full-control distinction like Python's `retries=3` vs `RetryPolicy(...)`.

```go
err := agnt5.RegisterFunction(worker, "send_email", sendEmail,
    agnt5.WithRetry(5, 500, 30000),      // maxAttempts, initialIntervalMS, maxIntervalMS
    agnt5.WithBackoff("exponential", 2.0),
)
```

`WithRetry(maxAttempts, initialIntervalMS, maxIntervalMS int)`:

| Argument | Description |
|---|---|
| `maxAttempts` | Total tries, including the first |
| `initialIntervalMS` | Wait before the first retry, in milliseconds |
| `maxIntervalMS` | Maximum wait between retries, in milliseconds |

`WithBackoff(backoffType string, multiplier float64)`:

| Argument | Description |
|---|---|
| `backoffType` | `"constant"` (fixed wait), `"linear"` (grows steadily), `"exponential"` (doubles each time) |
| `multiplier` | How fast the wait grows |

AGNT5 runs your function body once per attempt. Use `ctx.Attempt()` if you need to vary behavior on retries.



---

## Calling from a workflow



**Python:**

Inside a workflow, always call functions with `ctx.step()`. This tells AGNT5 to checkpoint the result. If the workflow restarts, the function is skipped and the saved result is returned directly.

```python
from agnt5 import workflow, WorkflowContext
from myapp.functions import send_email

@workflow
async def notify_workflow(ctx: WorkflowContext, user_email: str) -> str:
    result = await ctx.step(send_email, user_email, "Welcome!", "Thanks for signing up.")
    return result
```

`result` is whatever `send_email` returned. A plain `await send_email(ctx, ...)` also works but is not checkpointed. The function re-runs on every replay.

| How you call it | Checkpointed | When to use |
|---|---|---|
| `await ctx.step(send_email, ...)` | Yes | Inside a workflow (always prefer this) |
| `await send_email(ctx, ...)` | No | Outside a workflow, or in local tests |





**TypeScript:**

Inside a workflow, always call functions with `ctx.step()`. This tells AGNT5 to checkpoint the result. If the workflow restarts, the function is skipped and the saved result is returned directly.

```typescript
export const notifyWorkflow = workflow(
  'notify_workflow',
  async (ctx: Context, input: { userEmail: string }) => {
    const result = await ctx.step('send_email', () =>
      sendEmail(ctx, {
        to: input.userEmail,
        subject: 'Welcome!',
        body: 'Thanks for signing up.',
      }),
    );
    return result;
  },
);
```

`result` is whatever `sendEmail` returned. A plain `await sendEmail(ctx, ...)` also works but is not checkpointed. The function re-runs on every replay.

| How you call it | Checkpointed | When to use |
|---|---|---|
| `await ctx.step(name, () => sendEmail(...))` | Yes | Inside a workflow (always prefer this) |
| `await sendEmail(ctx, ...)` | No | Outside a workflow, or in local tests |





**Go:**

Inside a workflow, wrap the call in `agnt5.Step`. Unlike Python/TypeScript, `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.

```go
err := agnt5.RegisterWorkflow(worker, "notify_workflow", func(ctx *agnt5.Context, in NotifyInput) (string, error) {
    result, err := agnt5.Step(ctx, "send_email", func(context.Context) (string, error) {
        return sendEmail(ctx, EmailInput{
            To:      in.UserEmail,
            Subject: "Welcome!",
            Body:    "Thanks for signing up.",
        })
    })
    return result, err
})
```

`result` is whatever the closure returned. Calling `sendEmail(ctx, ...)` directly, outside a `Step`, also works but isn't checkpointed — the function re-runs on every replay.

| How you call it | Checkpointed | When to use |
|---|---|---|
| `agnt5.Step(ctx, "name", func(context.Context) (T, error) { ... })` | Yes | Inside a workflow (always prefer this) |
| `sendEmail(ctx, ...)` directly | No | Outside a workflow, or in local tests |