Creating middleware

Creating middleware means defining the lifecycles and subsequent hooks in those lifecycles to run code in. Lifecycles are actions such as a function run or sending events, and individual hooks within those are where we run code, usually with a before and after step.

Callout: Support for Middleware in the Go SDK in planned.

A middleware is created by extending the Middleware.BaseMiddleware class.

class MyMiddleware extends Middleware.BaseMiddleware

import { Middleware } from "inngest";

// Create a new middleware
class MyMiddleware extends Middleware.BaseMiddleware {
  id = "my-middleware";
  // Override hook methods here
}

// Register it on the client
const inngest = new Inngest({
  id: "my-app",
  middleware: [MyMiddleware],
});

A Middleware is created using the inngest.Middleware class.

class MyMiddleware(inngest.Middleware):

import inngest

class MyMiddleware(inngest.Middleware):
    def __init__(
        self,
        client: inngest.Inngest,
        raw_request: object,
    ) -> None:
        #  ...

    async def before_send_events( self, events: list[inngest.Event]) -> None:
        print(f"Sending {len(events)} events")

    async def after_send_events(self, result: inngest.SendEventsResult) -> None:
        print("Done sending events")

inngest_client = inngest.Inngest(
    app_id="my_app",
    middleware=[MyMiddleware],
)

Initialization

As you can see above, we start with a class that extends Middleware.BaseMiddleware. A fresh instance is created for every request, so you can safely use instance properties (this) to store per-request state.

import { Middleware } from "inngest";

class MyMiddleware extends Middleware.BaseMiddleware {
  id = "my-middleware";
  // A new instance is created for every request
  // Override hook methods here
}

As you can see above, we start with the __init__ method, which is called when the client is initialized.

import inngest

    class MyMiddleware(inngest.Middleware):
        def __init__(
            self,
            client: inngest.Inngest,
            raw_request: object,
        ) -> None:
            # This runs when the client is initialized
            # Use this to set up anything your middleware needs
            #  ...

Hook methods can be synchronous or async functions. For one-time setup like establishing a database connection, use the static onRegister hook which runs when the middleware class is registered with a client or function.

class MyMiddleware extends Middleware.BaseMiddleware {
  id = "my-middleware";
  static onRegister({ client }: Middleware.OnRegisterArgs) {
    // One-time setup, e.g. initialize a database connection
  }
}

Function registration, lifecycles, and hooks can all be with synchronous or async functions. This makes it easy for our initialization handler to do some async work, like setting up a database connection.

import inngest

    class MyMiddleware(inngest.Middleware):
        def __init__(
            self,
            client: inngest.Inngest,
            raw_request: object,
        ) -> None:
            #  ...connect to database

All lifecycle and hook functions can be synchronous or async functions - the SDK will always wait until a middleware's function has resolved before continuing to the next one.

Callout: As it's possible for an application to use multiple Inngest clients, it's recommended to always initialize dependencies within the initializer function/method, instead of in the global scope.

Specifying lifecycles and hooks

Override the hook methods you need directly on your middleware class. See the Middleware lifecycle reference for a full list of available hooks.

import { Middleware } from "inngest";

class MyMiddleware extends Middleware.BaseMiddleware {
  id = "my-middleware";

  // Observable hook: runs before the function handler on the first attempt
  onRunStart({ ctx, functionInfo }: Middleware.OnRunStartArgs) {
    console.log(`Starting ${functionInfo.id}`);
  }

  // Wrapping hook: wraps function execution
  async wrapFunctionHandler({ next }: Middleware.WrapFunctionHandlerArgs) {
    console.log("before");
    const result = await next();
    console.log("after");
    return result;
  }

  // Transform hook: modify function input (e.g. dependency injection)
  transformFunctionInput(args: Middleware.TransformFunctionInputArgs) {
    return {
      ...args,
      ctx: {
        ...args.ctx,
        // Add custom properties here
      },
    };
  }
}

Hooks you don't define have zero overhead: the SDK skips them entirely. A fresh instance is created for every request, so you can safely use instance properties (this) to store per-request state without worrying about leaks between runs.

Learn more about hooks with:

• Lifecycle - middleware ordering and see all available hooks
• Examples - real-world middleware examples

You might have notice that our custom middleware defines custom method such as before_send_events and after_send_events. Those methods, called hooks, enable your middleware to hook itself to specific steps of the Function and Steps execution lifecycle.

import inngest

class MyMiddleware(inngest.Middleware):
    def __init__(
        self,
        client: inngest.Inngest,
        raw_request: object,
    ) -> None:
        #  ...

    async def before_send_events( self, events: list[inngest.Event]) -> None:
        # called before an event is sent from within a Function or Step
        print(f"Sending {len(events)} events")

    async def after_send_events(self, result: inngest.SendEventsResult) -> None:
        # called after an event is sent from within a Function or Step
        print("Done sending events")

You can find the full list of available hooks in the Python SDK reference.

Adding configuration

It's common for middleware to require additional customization or options from developers. For this, we recommend creating a function that takes in some options and returns the middleware.

import { Middleware } from "inngest";

export function createMyMiddleware(logEventOutput: string) {
  return class MyMiddleware extends Middleware.BaseMiddleware {
    id = "my-middleware";
    onRunComplete({ ctx, functionInfo, output }: Middleware.OnRunCompleteArgs) {
      if (ctx.event.name === logEventOutput) {
        console.log(
          `${logEventOutput} output: ${JSON.stringify(output)}`
        );
      }
    }
  };
}

import { createMyMiddleware } from "./middleware/myMiddleware";

export const inngest = new Inngest({
  id: "my-client",
  middleware: [createMyMiddleware("app/user.created")],
});

Callout: Make sure to let TypeScript infer the output of the function instead of strictly typing it; this helps Inngest understand changes to input and output of arguments. See Middleware lifecycle for more information.

Adding configuration to a custom middleware can be achieved by adding a factory() class method, leveraging the Factory pattern.

For example, let's add a secret_key configuration option to our MyMiddleware middleware:

import inngest

class MyMiddleware(inngest.Middleware):
    def __init__(
        self,
        client: inngest.Inngest,
        raw_request: object,
    ) -> None:
        #  ...

    @classmethod
    def factory(
        cls,
        secret_key: typing.Union[bytes, str],
    ) -> typing.Callable[[inngest.Inngest, object], MyMiddleware]:
        def _factory(
            client: inngest.Inngest,
            raw_request: object,
        ) -> MyMiddleware:
            return cls(
                client,
                raw_request,
                secret_key,
            )

        return _factory

    async def before_send_events( self, events: list[inngest.Event]) -> None:
        # called before an event is sent from within a Function or Step
        print(f"Sending {len(events)} events")

    async def after_send_events(self, result: inngest.SendEventsResult) -> None:
        # called after an event is sent from within a Function or Step
        print("Done sending events")

Our middleware can now be registered as follow:

inngest_client = inngest.Inngest(
    app_id="my_app",
    middleware=[MyMiddleware.factory(_secret_key)],
)

Next steps

Check out our pre-built middleware and examples:

Dependency Injection: Provide shared client instances (ex, OpenAI) to your Inngest Functions.

Encryption Middleware: End-to-end encryption for events, step output, and function output.

Sentry Middleware: Quickly setup Sentry for your Inngest Functions.

Datadog middleware: Add tracing with Datadog under a few minutes.

Cloudflare Workers & Hono middleware: Access environment variables within Inngest functions.