Inngest

forze[inngest] implements the durable-function contracts on Inngest — emit events that trigger functions, and run memoized steps inside them, behind the durable ports.

Install¶

uv add 'forze[inngest]'

Needs Inngest, and a served function endpoint. The FastAPI serve helper needs forze[inngest,fastapi].

The client¶

from forze_inngest import InngestClient, InngestConfig

inngest = InngestClient(app_id="orders", config=InngestConfig())

RoutedInngestClient resolves per-tenant credentials.

Wire it¶

Register the events you emit and bind your functions to operations:

from forze.application.execution import DepsRegistry
from forze_inngest import InngestDepsModule, InngestEventConfig, InngestFunctionBinding

bindings = [InngestFunctionBinding.for_registry_operation(fulfil_spec, registry)]

deps = DepsRegistry.from_modules(
    InngestDepsModule(client=inngest, events={"orders": InngestEventConfig()}, function_bindings=bindings),
)

Serve the functions from FastAPI (registers them with Inngest):

from forze_inngest.fastapi import serve

serve(app, inngest, bindings, ctx_factory=lambda req: runtime.get_context(), registry=registry)

What it provides¶

Notes¶

• A function binding maps a DurableFunctionSpec to either an operation (for_registry_operation) or a handler factory — set exactly one.
• The execution-context metadata travels in a _forze envelope and is restored in the worker, so functions run with the right identity/tenant.
• Steps resolve only inside a running function.