# Inngest Steps

Inngest steps are checkpointed, retriable units of work inside [Inngest functions](/docs-markdown/learn/inngest-functions) and [Durable Endpoints](/docs-markdown/learn/durable-endpoints). Put side effects, API calls, sleeps, waits, and function invocations in steps so Inngest can memoize results, retry failures independently, and resume long-running work from the last successful checkpoint.

Steps are the building blocks of durable execution. They let a function pause without holding compute, recover individual units of work after an error, and compose multi-step workflows from normal TypeScript, Python, or Go code.

On this page, you will learn about the benefits of using steps, and get an overview of the available step methods.

## Benefits of Using Steps

- **Improved reliability**: structured steps enable precise control and handling of each task within a function.
- **Error handling**: capturing and managing errors at the step level means better error recovery.
- **Retry mechanism**: failing steps can be retried and recovered independently, without re-executing other successful steps.
- **Independent testing**: each step can be tested and debugged independently from others.
- **Improved code readability**: modular approach makes code easier to navigate and refactor.

If you'd like to learn more about how Inngest steps are executed, check the ["How Inngest functions are executed"](/docs-markdown/learn/how-functions-are-executed) page.

## Anatomy of an Inngest Step

The first argument of every Inngest step method is an `id`. Each step is treated as a discrete task which can be individually retried, debugged, or recovered. Inngest uses the ID to memoize step state across function versions.

```typescript
export default inngest.createFunction(
  { id: "import-product-images", triggers: { event: "shop/product.imported" } },
  async ({ event, step }) => {
    const uploadedImageURLs = await step.run(
      // step ID
      "copy-images-to-s3",
      // other arguments, in this case: a handler
      async () => {
        return copyAllImagesToS3(event.data.imageURLs);
    });
  }
);
```

```go
import (
  "github.com/inngest/inngestgo"
  "github.com/inngest/inngestgo/step"
)

inngestgo.CreateFunction(
	client,
	// config
	inngestgo.FunctionOpts{
		ID: "import-product-images",
	},
	// trigger (event or cron)
	inngestgo.EventTrigger("shop/product.imported", nil),
	// handler function
	func(ctx context.Context, input inngestgo.Input[map[string]any]) (any, error) {
		// Here goes the business logic
		// By wrapping code in steps, it will be retried automatically on failure
		s3Urls, err := step.Run("copy-images-to-s3", func() ([]string, error) {
			return copyAllImagesToS3(input.Event.Data["imageURLs"].([]string))
		})
		if err != nil {
			return nil, err
		}

		return nil, nil
	},
)
```

```python
import inngest
from src.inngest.client import inngest_client

@inngest_client.create_function(
    fn_id="import-product-images",
    event="shop/product.imported"
)
async def import_product_images(ctx: inngest.Context):
    uploaded_image_urls = await ctx.step.run(
        # step ID
        "copy-images-to-s3",
        # other arguments, in this case: a handler
        lambda: copy_all_images_to_s3(ctx.event.data["image_urls"])
    )
```

The ID is also used to identify the function in the Inngest system.

Inngest's SDK also records a counter for each unique step ID. The counter increases every time the same step is called. This allows you to run the same step in a loop, without changing the ID.

> **Callout:** Place non-deterministic side effects, such as database writes or API calls, inside step.run() so Inngest can checkpoint the result and avoid re-running completed work during retries.

## Available Step Methods

### step.run()

This method executes a defined piece of code. Code within `step.run()` is automatically retried if it throws an error. When `step.run()` finishes successfully, the response is saved in the function run state and the step will not re-run.

Use it to run synchronous or asynchronous code as a retriable step in your function.

```typescript
export default inngest.createFunction(
  { id: "import-product-images", triggers: { event: "shop/product.imported" } },
  async ({ event, step }) => {
    // Here goes the business logic
    // By wrapping code in steps, it will be retried automatically on failure
    const uploadedImageURLs = await step.run("copy-images-to-s3", async () => {
      return copyAllImagesToS3(event.data.imageURLs);
    });
  }
);
```

```go
import (
  "github.com/inngest/inngestgo"
  "github.com/inngest/inngestgo/step"
)

inngestgo.CreateFunction(
	client,
	inngestgo.FunctionOpts{
		ID: "import-product-images",
	},
	inngestgo.EventTrigger("shop/product.imported", nil),
	func(ctx context.Context, input inngestgo.Input[map[string]any]) (any, error) {
		// Here goes the business logic
		// By wrapping code in steps, it will be retried automatically on failure
		s3Urls, err := step.Run("copy-images-to-s3", func() ([]string, error) {
			return copyAllImagesToS3(input.Event.Data["imageURLs"].([]string))
		})
		if err != nil {
			return nil, err
		}

		return nil, nil
	},
)
```

```python
import inngest
from src.inngest.client import inngest_client

@inngest_client.create_function(
    fn_id="import-product-images",
    event="shop/product.imported"
)
async def import_product_images(ctx: inngest.Context):
    # Here goes the business logic
    # By wrapping code in steps, it will be retried automatically on failure
    uploaded_image_urls = await ctx.step.run(
        # step ID
        "copy-images-to-s3",
        # other arguments, in this case: a handler
        lambda: copy_all_images_to_s3(ctx.event.data["image_urls"])
    )
```

> **Callout:** step.run() acts as a code-level transaction. The entire step must succeed to complete.

### step.sleep()

This method pauses execution for a specified duration. Even though it seems like a `setInterval`, your function does not run for that time (you don't use any compute). Inngest handles the scheduling for you. Use it to add delays or to wait for a specific amount of time before proceeding. At maximum, functions can sleep for a year (seven days for the [free tier plans](/pricing)).

```typescript
export default inngest.createFunction(
  { id: "send-delayed-email", triggers: { event: "app/user.signup" } },
  async ({ event, step }) => {
    await step.sleep("wait-a-couple-of-days", "2d");
    // Do something else
  }
);
```

```go
import (
  "github.com/inngest/inngestgo"
  "github.com/inngest/inngestgo/step"
)

inngestgo.CreateFunction(
	client,
	inngestgo.FunctionOpts{
		ID: "send-delayed-email",
	},
	inngestgo.EventTrigger("app/user.signup", nil),
	// handler function
	func(ctx context.Context, input inngestgo.Input[map[string]any]) (any, error) {
		step.Sleep("wait-a-couple-of-days", 2*time.Day)
		return nil, nil
	},
)
```

```python
import inngest
from src.inngest.client import inngest_client

@inngest_client.create_function(
    fn_id="send-delayed-email",
    trigger=inngest.TriggerEvent(event="app/user.signup")
)
async def send_delayed_email(ctx: inngest.Context):
    await ctx.step.sleep("wait-a-couple-of-days", datetime.timedelta(days=2))
    # Do something else
```

### step.sleepUntil() / step.sleep\_until()

This method pauses execution until a specific date time. Any date time string in the format accepted by the Date object, for example `YYYY-MM-DD` or `YYYY-MM-DDHH:mm:ss`. At maximum, functions can sleep for a year (seven days for the [free tier plans](/pricing)).

```typescript
export default inngest.createFunction(
  { id: "send-scheduled-reminder", triggers: { event: "app/reminder.scheduled" } },
  async ({ event, step }) => {
    const date = new Date(event.data.remind_at);
    await step.sleepUntil("wait-for-the-date", date);
    // Do something else
  }
);
```

Go SDK does not have a `sleepUntil` method. Use `step.Sleep()` with a calculated duration instead.

```python
import inngest
from src.inngest.client import inngest_client
from datetime import datetime

@inngest_client.create_function(
    fn_id="send-scheduled-reminder",
    trigger=inngest.TriggerEvent(event="app/reminder.scheduled")
)
async def send_scheduled_reminder(ctx: inngest.Context):
    date = datetime.fromisoformat(ctx.event.data["remind_at"])
    await ctx.step.sleep_until("wait-for-the-date", date)
    # Do something else
```

### step.waitForEvent() / step.wait\_for\_event()

This method pauses a run's execution until a specific event is received.

```typescript
export default inngest.createFunction(
  { id: "send-onboarding-nudge-email", triggers: { event: "app/account.created" } },
  async ({ event, step }) => {
    const onboardingCompleted = await step.waitForEvent(
      "wait-for-onboarding-completion",
      { event: "app/onboarding.completed", timeout: "3d", if: "event.data.userId == async.data.userId" }
    );
    // Do something else
  }
);
```

```go
import (
  "github.com/inngest/inngestgo"
  "github.com/inngest/inngestgo/errors"
  "github.com/inngest/inngestgo/step"
)

inngestgo.CreateFunction(
	client,
	inngestgo.FunctionOpts{
		ID: "send-delayed-email",
	},
	inngestgo.EventTrigger("app/user.signup", nil),
	// handler function
	func(ctx context.Context, input inngestgo.Input[map[string]any]) (any, error) {
		// Sample from the event stream for new events.  The function will stop
		// running and automatically resume when a matching event is found, or if
		// the timeout is reached.
		fn, err := step.WaitForEvent[FunctionCreatedEvent](
			ctx,
			"wait-for-activity",
			step.WaitForEventOpts{
				Name:    "Wait for a function to be created",
				Event:   "api/function.created",
				Timeout: time.Hour * 72,
				// Match events where the user_id is the same in the async sampled event.
				If: inngestgo.StrPtr("event.data.user_id == async.data.user_id"),
			},
		)
		if err == step.ErrEventNotReceived {
			// A function wasn't created within 3 days.  Send a follow-up email.
			_, _ = step.Run(ctx, "follow-up-email", func(ctx context.Context) (any, error) {
				// ...
				return true, nil
			})
			return nil, nil
		}
		return nil, nil
	},
)
```

```python
import inngest
from src.inngest.client import inngest_client

@inngest_client.create_function(
    fn_id="send-onboarding-nudge-email",
    trigger=inngest.TriggerEvent(event="app/account.created")
)
async def send_onboarding_nudge_email(ctx: inngest.Context):
    onboarding_completed = await ctx.step.wait_for_event(
      "wait-for-onboarding-completion",
      event="app/wait_for_event.fulfill",
      if_exp="event.data.user_id == async.data.user_id",
      timeout=datetime.timedelta(days=1),
    );
    # Do something else
```

### step.waitForSignal() / step.wait\_for\_signal()

This method pauses a run's execution until a specific signal is received via our SDK or API.  Signals must
be unique across runs, and offer an API to resume specific runs given a unique signal vs events.

```typescript
export default inngest.createFunction(
  { id: "send-onboarding-nudge-email", triggers: { event: "app/account.created" } },
  async ({ event, step }) => {
    const onboardingCompleted = await step.waitForSignal(
      "wait-for-specific-signal",
      { signal: "task/0e7d16d1-8335-4b93-9122-76e7cc6d9eb6", timeout: "3d" }
    );
    // Do something with signal data
  }
);
```

```go
import (
  "github.com/inngest/inngestgo"
  "github.com/inngest/inngestgo/errors"
  "github.com/inngest/inngestgo/step"
)

inngestgo.CreateFunction(
	client,
	inngestgo.FunctionOpts{
		ID: "send-delayed-email",
	},
	inngestgo.EventTrigger("app/user.signup", nil),
	// handler function
	func(ctx context.Context, input inngestgo.Input[map[string]any]) (any, error) {
		// Sample from the event stream for new events.  The function will stop
		// running and automatically resume when a matching event is found, or if
		// the timeout is reached.
		signal, err := step.WaitForSignal[string](
			ctx,
			"wait-for-signal",
			step.WaitForSignalOpts{
				Name:    "Wait for a signal to be POSTed",
				Signal:   "task/0e7d16d1-8335-4b93-9122-76e7cc6d9eb6",
				Timeout: time.Hour * 72,
			},
		)
		if err == step.ErrSignalNotReceived {
			// A signal wasn't received within 3 days.  Send a follow-up email.
		}
		return nil, nil
	},
)
```

```python
import inngest
from src.inngest.client import inngest_client

@inngest_client.create_function(
    fn_id="send-onboarding-nudge-email",
    trigger=inngest.TriggerEvent(event="app/account.created")
)
async def send_onboarding_nudge_email(ctx: inngest.Context):
    onboarding_completed = await ctx.step.wait_for_signal(
      "wait-for-onboarding-completion",
      signal="task/0e7d16d1-8335-4b93-9122-76e7cc6d9eb6",
      if_exp="event.data.user_id == async.data.user_id",
      timeout=datetime.timedelta(days=1),
    );
    # Do other steps
```

### step.invoke()

This method is used to asynchronously call another Inngest function ([written in any language SDK](/blog/cross-language-support-with-new-sdks)) and handle the result. Invoking other functions allows you to easily re-use functionality and compose them to create more complex workflows or map-reduce type jobs.

This method comes with its own configuration, which enables defining specific settings like concurrency limits.

```typescript
// A function we will call in another place in our app
const computeSquare = inngest.createFunction(
  { id: "compute-square", triggers: { event: "calculate/square" } },
  async ({ event }) => {
    return { result: event.data.number * event.data.number }; // Result typed as { result: number }
  }
);

// In this function, we'll call `computeSquare`
const mainFunction = inngest.createFunction(
  { id: "main-function", triggers: { event: "main/event" } },
  async ({ step }) => {
    const square = await step.invoke("compute-square-value", {
      function: computeSquare,
      data: { number: 4 }, // input data is typed, requiring input if it's needed
    });

    return `Square of 4 is ${square.result}.`; // square.result is typed as number
  }
);
```

```go
import (
  "github.com/inngest/inngestgo"
  "github.com/inngest/inngestgo/errors"
  "github.com/inngest/inngestgo/step"
)

inngestgo.CreateFunction(
	client,
	inngestgo.FunctionOpts{
		ID: "send-delayed-email",
	},
	inngestgo.EventTrigger("app/user.signup", nil),
	// handler function
	func(ctx context.Context, input inngestgo.Input[map[string]any]) (any, error) {
		// Invoke another function and wait for its result
		result, err := step.Invoke[any](
			ctx,
			"invoke-email-function",
			step.InvokeOpts{
				FunctionID: "send-welcome-email",
				// Pass data to the invoked function
				Data: map[string]any{
					"user_id": input.Event.Data["user_id"],
					"email":   input.Event.Data["email"],
				},
				// Optional: Set a concurrency limit
				Concurrency: step.ConcurrencyOpts{
					Limit: 5,
					Key:   "user-{{event.data.user_id}}",
				},
			},
		)
		if err != nil {
			return nil, err
		}
		return result, nil
	},
)
```

```python
import inngest
from src.inngest.client import inngest_client

@inngest_client.create_function(
    fn_id="fn-1",
    trigger=inngest.TriggerEvent(event="app/fn-1"),
)
async def fn_1(ctx: inngest.Context) -> None:
    return "Hello!"

@inngest_client.create_function(
    fn_id="fn-2",
    trigger=inngest.TriggerEvent(event="app/fn-2"),
)
async def fn_2(ctx: inngest.Context) -> None:
    output = await ctx.step.invoke(
        "invoke",
        function=fn_1,
    )

    # Prints "Hello!"
    print(output)
```

### step.sendEvent() / step.send\_event()

This method sends events to Inngest to invoke functions with a matching event. Use `sendEvent()` when you want to trigger other functions, but you do not need to return the result. It is useful for example in [fan-out functions](/docs-markdown/guides/fan-out-jobs).

```typescript
export default inngest.createFunction(
  { id: "user-onboarding", triggers: { event: "app/user.signup" } },
  async ({ event, step }) => {
    // Do something
    await step.sendEvent("send-activation-event", {
      name: "app/user.activated",
      data: { userId: event.data.userId },
    });
    // Do something else
  }
);
```

Go SDK does not have a dedicated `step.sendEvent()` method. Use the Inngest client's `Send()` method within a `step.Run()` instead.

```python
import inngest
from src.inngest.client import inngest_client

@inngest_client.create_function(
    fn_id="my_function",
    trigger=inngest.TriggerEvent(event="app/my_function"),
)
async def fn(ctx: inngest.Context) -> list[str]:
    return await ctx.step.send_event("send", inngest.Event(name="foo"))
```

## Further reading

- [Quick Start](/docs-markdown/getting-started/nextjs-quick-start?ref=docs-inngest-steps): learn how to build complex workflows.
- ["How Inngest functions are executed"](/docs-markdown/learn/how-functions-are-executed): Learn more about Inngest's execution model, including how steps are handled.
- Docs guide: ["Multi-step functions"](/docs-markdown/learn/inngest-steps).