Inngest Python SDK

Warnings

• gotcha In `Connect` mode, user code that blocked the main thread could cause the Inngest server to perceive the worker as unresponsive or dead, leading to unexpected disconnections or re-runs.
  Fix: Upgrade to `inngest>=0.5.17`. This version moves Connect internals into a dedicated thread, improving resilience. However, for best practices, still ensure long-running synchronous operations are offloaded or made asynchronous where possible.
• gotcha Event payloads larger than 1MB sent via Inngest `Connect` could be silently ignored or truncated without error feedback.
  Fix: Upgrade to `inngest>=0.5.15`. Always monitor your Inngest dashboard for successful event ingestion and verify that large payloads are processed correctly. Consider storing large data externally (e.g., S3) and passing references instead of the full payload.
• gotcha Parallel steps in workflows might not have executed correctly, potentially leading to serial execution or unexpected behavior for functions designed for concurrency.
  Fix: Upgrade to `inngest>=0.5.15` to resolve issues with parallel step execution. Thoroughly test any workflows utilizing `step.parallel()` after upgrading to confirm desired behavior.

Install

• pip install inngest Install core library

Imports

• Inngest

  from inngest import Inngest

• Event

  from inngest import Event

• Function

  from inngest import Function

• Step

  from inngest import Step

• InngestFastAPI

  from inngest.framework.fastapi import InngestFastAPI

  Required for integrating with FastAPI applications.
• InngestFlask

  from inngest.framework.flask import InngestFlask

  Required for integrating with Flask applications.

Quickstart

This example sets up a simple Inngest function using FastAPI. The function `hello_world` is triggered by an `app/hello.world` event, sleeps for a second, and returns a greeting. It includes a FastAPI endpoint `/trigger` to manually send this event. Remember to install `fastapi` and `uvicorn`, and set `INGEST_EVENT_KEY` and `INGEST_SIGNING_KEY` environment variables (or disable production mode) to run this.

import os
from fastapi import FastAPI
from inngest import Inngest, Event, Function, Step
from inngest.framework.fastapi import InngestFastAPI

# Initialize Inngest client
inngest_client = Inngest(
    app_id="my-python-app",
    event_key=os.environ.get('INGEST_EVENT_KEY', ''),
    signing_key=os.environ.get('INGEST_SIGNING_KEY', ''),
    # Uncomment for local dev without an Inngest deployment
    # is_production=False,
)

# Define an Inngest function
@inngest_client.function(
    Function(
        id="hello-world",
        name="Hello World Function",
        trigger=Event(name="app/hello.world"),
    )
)
async def hello_world(step: Step, event: Event):
    name = event.data.get("name", "world")
    await step.sleep("1s")
    message = await step.run(
        "say-hello", lambda: f"Hello, {name}! This is Inngest from Python."
    )
    return {"message": message}

# Initialize FastAPI app
app = FastAPI()

# Integrate Inngest with FastAPI
inngest_fastapi = InngestFastAPI(inngest_client, [hello_world])
app.include_router(inngest_fastapi.create_router())

# Example endpoint to trigger the Inngest function
@app.post("/trigger")
async def trigger_hello_world(name: str = "friend"):
    await inngest_client.send(Event(name="app/hello.world", data={"name": name}))
    return {"status": "Event sent", "name": name}

# To run this example:
# 1. pip install fastapi uvicorn inngest
# 2. Set environment variables: INGEST_EVENT_KEY and INGEST_SIGNING_KEY
#    (or comment them out and set is_production=False for local-only testing)
# 3. Run: uvicorn main:app --reload
# 4. Access Inngest dashboard or send a POST request to /trigger