Python SDK

The Python SDK defines durable functions, sends events, and serves your functions from FastAPI, Flask, or Django. It also runs Connect workers. Some features are TypeScript-only for now; see feature support.

Install

The SDK needs Python 3.10 or newer.

pip install inngest

Create a client and serve functions

Create one client for your app, define functions with it, and serve them from your web framework. This example uses FastAPI:

import logging

from fastapi import FastAPI
import inngest
import inngest.fast_api

inngest_client = inngest.Inngest(
    app_id="my-app",
    logger=logging.getLogger("uvicorn"),
)

@inngest_client.create_function(
    fn_id="process-task",
    trigger=inngest.TriggerEvent(event="app/task.created"),
)
async def process_task(ctx: inngest.Context) -> dict[str, object]:
    return await ctx.step.run("handle-task", lambda: {"processed": True})

app = FastAPI()
inngest.fast_api.serve(app, inngest_client, [process_task])

The Python quick start walks through a full example with the Dev Server.

Define steps

Steps are the units of work Inngest retries and memoizes. The Python SDK gives you two ways to write them.

Imperative: call ctx.step.run() with a step ID and a handler inside your function.

@inngest_client.create_function(
    fn_id="greet-user",
    trigger=inngest.TriggerEvent(event="app/greet-user"),
)
async def greet_user(ctx: inngest.Context) -> str:
    async def prepare_greeting() -> str:
        return f"Welcome, {ctx.event.data['name']}!"

    return await ctx.step.run("prepare-greeting", prepare_greeting)

Decorator (experimental): decorate a function with step from inngest.experimental. Calls inside an Inngest function run as steps; calls anywhere else run the function directly.

from inngest.experimental import step

@step("prepare-greeting")
async def prepare_greeting(name: str) -> str:
    return f"Welcome, {name}!"

@inngest_client.create_function(
    fn_id="greet-user",
    trigger=inngest.TriggerEvent(event="app/greet-user"),
)
async def greet_user(ctx: inngest.Context) -> str:
    return await prepare_greeting(str(ctx.event.data["name"]))

Both produce the same step. Use ctx.step.run() for logic specific to one function; use the decorator for steps you share across functions or also call from regular code. See choosing step.run or the decorator.

Reference

Client and events

• Inngest client
• Send events

Functions

• Create a function: triggers, retries, flow control, and cancellation options

Steps

• run, sleep, and sleep_until
• wait_for_event
• invoke, invoke_by_id, and send_event
• parallel
• @step decorator (experimental): turn a Python function into a reusable step

Middleware

• Overview and lifecycle

Configuration

• Environment variables
• Production mode

Guides

• Testing, Pydantic, and Modal
• Realtime in Python (beta)

Migrations

• v0.4 to v0.5 and v0.3 to v0.4

Source code

The Python SDK is open source on GitHub: inngest/inngest-py. Releases are published to PyPI.