> 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, and checkpoint inside workflows.
last_verified: 2026-10-06
---



A **function** is a stateless unit of work: it receives inputs, runs your logic, and returns a result. The platform handles registration and retries. When a workflow calls a function, the result is checkpointed, so a replay returns the saved result instead of calling the function again.

## 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` | Maximum attempts, including the first. `retries=3` runs the function up to three times. |
| `backoff` | `str \| BackoffPolicy` | `None` | How to space out retries: `"constant"`, `"linear"`, or `"exponential"`. |
| `timeout_ms` | `int` | `None` | Cancel the function with a `TimeoutError` 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. Define the function with `async def` when you set `timeout_ms`, so the timeout can cancel the running work.





**Go:**

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

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

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

err := agnt5.RegisterFunction(worker, "send_email", sendEmail)
```

`RegisterFunction[In, Out any](w *Worker, name string, handler func(*Context, In) (Out, error), opts ...ComponentOption) error` takes `ComponentOption`s for configuration:

| Option | Description |
|---|---|
| `WithRetry(maxAttempts, initialIntervalMS, maxIntervalMS int)` | Maximum attempts, including the first, and the backoff interval bounds in milliseconds |
| `WithBackoff(backoffType string, multiplier float64)` | `"constant"`, `"linear"`, or `"exponential"` |

**Example with retries and backoff:**

```go
err := agnt5.RegisterFunction(worker, "send_email", sendEmail,
    agnt5.WithRetry(3, 250, 5000),
    agnt5.WithBackoff("exponential", 2),
)
```

To set a deadline, derive a context from `ctx`, which embeds `context.Context`: `tctx, cancel := context.WithTimeout(ctx, 10*time.Second)`. Pass `tctx` to the call that should stop at the deadline, such as an HTTP request to your email provider.





**TypeScript:**

One thing is required: an async function passed to `fn(name).run(...)`.

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

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)` | Retry policy. `maxAttempts` counts the first attempt. |
| `.backoff(policy: BackoffPolicy)` | How to space out retries. |

**Example with retries and backoff:**

```typescript
export const sendEmail = fn('send_email')
  .retry({ maxAttempts: 3 })
  .backoff({ type: 'exponential' })
  .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}`;
  });
```



## Describe inputs and outputs



**TypeScript:**

The builder also accepts metadata for Studio and the function manifest:

| Method | Description |
|---|---|
| `.description(text)` | What the function does. Used as the tool description when the function is published over MCP. |
| `.inputSchema(schema)` / `.outputSchema(schema)` | JSON Schema for the function's input and output. TypeScript types are erased at runtime, so the schema is how Studio learns the shape. |
| `.maxConcurrency(limit)` / `.priority(priority)` | Static concurrency limit and scheduling priority for generated manifests. |

```typescript
export const sendEmail = fn('send_email')
  .description('Send a transactional email')
  .inputSchema({
    type: 'object',
    properties: { to: { type: 'string' }, subject: { type: 'string' }, body: { type: 'string' } },
    required: ['to', 'subject', 'body'],
  })
  .outputSchema({ type: 'string' })
  .run(async (ctx: Context, input: { to: string; subject: string; body: string }) => {
    return `Sent to ${input.to}`;
  });
```





**Go:**

Other `ComponentOption`s attach schemas and metadata to the registration:

| Option | Description |
|---|---|
| `WithInputSchema(schema)` / `WithOutputSchema(schema)` | JSON Schema (`map[string]any`) for the input and output, registered with the component. |
| `WithComponentMetadata(map[string]string)` / `WithComponentConfig(map[string]string)` | Key/value metadata and configuration stored with the component. |

```go
err := agnt5.RegisterFunction(worker, "send_email", sendEmail,
    agnt5.WithInputSchema(map[string]any{
        "type": "object",
        "properties": map[string]any{
            "to":      map[string]any{"type": "string"},
            "subject": map[string]any{"type": "string"},
            "body":    map[string]any{"type": "string"},
        },
        "required": []string{"to", "subject", "body"},
    }),
    agnt5.WithOutputSchema(map[string]any{"type": "string"}),
)
```





**Python:**

Python builds the input and output schema from the function's type hints, so there is nothing extra to declare. Use Pydantic models for richer schemas.



## 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 without blocking the event loop. For a timer that survives a worker restart, use `ctx.sleep` in a workflow. |





**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
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)` |
| `ctx.Sleep(d, opts...)` | Pause execution for a `time.Duration`. Durable when the runtime negotiates `durable_suspension_v1`: the timer survives worker restarts and resumes after the remaining delay |

`ctx.Sleep` is the same durable timer workflows use (see [Durable sleep](/docs/build/workflows.md#durable-sleep)). Pass `agnt5.WithSleepKey("name")` when control flow can change the sleep's source-order position, and prefer it over `time.Sleep`, which isn't checkpointed and re-runs its full duration on replay.

```go
if err := ctx.Sleep(30*time.Second, agnt5.WithSleepKey("throttle")); err != nil {
    return "", err
}
```



---

## Retries and backoff

When a function fails, AGNT5 runs it again until it succeeds or reaches the attempt limit. Each attempt runs your function body once, and `ctx.attempt` is `0` on the first attempt, `1` on the first retry, and so on.



**Python:**

### Shorthand

Pass an integer for the maximum number of attempts and a string for the backoff strategy. The first attempt counts, so `retries=3` runs the function at most three times.

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

Use `ctx.attempt` if you need to vary behaviour on retries.





**TypeScript:**

Pass a `RetryPolicy` and a `BackoffPolicy` object to the builder.

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

Use `ctx.attempt` if you need to vary behaviour on retries.





**Go:**

Pass plain arguments to `WithRetry` and `WithBackoff`.

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

Use `ctx.Attempt()` if you need to vary behavior on retries.



---

## Calling from a workflow



**Python:**

Inside a workflow, 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.





**TypeScript:**

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

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.





**Go:**

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



A checkpointed result is recorded once the platform accepts the step's completion. Keep calls to external services idempotent so a repeated call has no extra effect.

## Next steps

- [Workflows](/docs/build/workflows.md): call functions as checkpointed steps and compose them into durable pipelines.
- [Agents](/docs/build/agents.md): let a model decide when to call your code.
- [Local development](/docs/build/local-development.md): run your worker with `agnt5 dev` and trigger functions with `agnt5 run`.