### Install Dependencies

Source: https://github.com/inngest/inngest-py/blob/main/CONTRIBUTING.md

Run this command to install all necessary project dependencies.

```bash
make install
```

--------------------------------

### Install Client Dependencies and Build

Source: https://github.com/inngest/inngest-py/blob/main/examples/fast_api_realtime/README.md

Install npm dependencies and build the client-side application. Ensure your .env file is configured.

```bash
cp .env.example .env
npm install
npm run build
```

--------------------------------

### Start FastAPI Server

Source: https://github.com/inngest/inngest-py/blob/main/examples/fast_api_realtime/README.md

Use this command to start the FastAPI development server.

```bash
make dev
```

--------------------------------

### Environment Variables Setup

Source: https://github.com/inngest/inngest-py/blob/main/tests/smoke_tests/model_adapters/README.md

Configure your API keys and Inngest base URL in a .env file. This is essential for the smoke test to authenticate with the AI providers and the Inngest service.

```dotenv
INNGEST_BASE_URL=http://localhost:8288
OPENAI_API_KEY=your_openai_api_key_here
ANTHROPIC_API_KEY=your_anthropic_api_key_here
GEMINI_API_KEY=your_gemini_api_key_here
GROK_API_KEY=your_grok_api_key_here
DEEPSEEK_API_KEY=your_deepseek_api_key_here
```

--------------------------------

### Configure Inngest Client for Production

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/README.md

Instantiate the Inngest client for production use. This example sets `is_production` based on the presence of the `INNGEST_DEV` environment variable, defaulting to production if it's not set.

```python
inngest_client = inngest.Inngest(
    app_id="my_app",
    is_production=os.getenv("INNGEST_DEV") is None,
)
```

--------------------------------

### Run Inngest Dev Server

Source: https://github.com/inngest/inngest-py/blob/main/examples/fast_api_realtime/README.md

Start the Inngest Dev Server, pointing it to your FastAPI application's Inngest endpoint.

```bash
npx inngest-cli@latest dev -u http://localhost:8000/api/inngest
```

--------------------------------

### Run Smoke Test

Source: https://github.com/inngest/inngest-py/blob/main/tests/smoke_tests/model_adapters/README.md

Navigate to the smoke test directory and execute the make dev command to start the FastAPI server. This command initiates the test environment, loads variables, and exposes the Inngest function.

```bash
cd tests/smoke_tests/model_adapters
make dev
```

--------------------------------

### Inngest Server to SDK Request Example

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/docs/REQUEST_LIFECYCLE.md

Illustrates the HTTP POST request format from the Inngest server to the SDK for function execution. It includes query parameters for function and step identification, and the request body contains event data, memoized outputs, and execution context.

```http
Inngest Server → POST /api/inngest?fnId=...&stepId=...

```

--------------------------------

### Initialize Inngest Client in Production Mode

Source: https://github.com/inngest/inngest-py/blob/main/README.md

Instantiate the Inngest client for production. Ensure INNGEST_DEV environment variable is not set for production mode. This setup is crucial for connecting to Inngest Cloud and enabling request signature verification.

```python
import os
import inngest

inngest_client = inngest.Inngest(
    app_id="my_app",
    is_production=os.getenv("INNGEST_DEV") is None,
)
```

--------------------------------

### Establish WebSocket Connection

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/docs/CONNECT.md

Use `inngest.connect()` to establish a WebSocket connection with the Inngest server. Pass a list of clients and their associated functions. Calling `start()` on the returned connection object will block until the connection is closed.

```python
connection = inngest.connect([(client, [func1, func2])])
await connection.start()  # Blocks until the connection closes
```

--------------------------------

### Python Function with Inngest Steps

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/docs/EXECUTION_MODEL.md

Defines an Inngest function with two sequential steps: creating a user and sending an email. This example illustrates how steps are declared and executed within the Inngest framework, highlighting the replay mechanism.

```python
import inngest

client = inngest.Inngest()

async def create_user():
    # Simulate user creation
    return {"id": "user-123", "email": "test@example.com"}

async def send_email(user):
    # Simulate sending email
    print(f"Sending email to {user['email']}")
    return "email sent"

@client.create_function(
    fn_id="my-fn",
    trigger=inngest.TriggerEvent(event="app/signup"),
)
async def my_fn(ctx: inngest.Context) -> str:
    user = await ctx.step.run("create-user", create_user)       # Step A
    await ctx.step.run("send-email", lambda: send_email(user))  # Step B
    return "done"
```

--------------------------------

### Establish WebSocket Connection

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/docs/CONNECT.md

Use `inngest.connect()` to establish a persistent WebSocket connection to the Inngest server. This function returns a `WorkerConnection` object. The `start()` method on this object will block until the connection is closed.

```APIDOC
## Establish WebSocket Connection

### Description
Establishes a persistent WebSocket connection to the Inngest server, allowing for long-lived processes without HTTP timeouts. This is an alternative to HTTP-based serving.

### Method Signature
```python
inngest.connect([(client, [func1, func2])])
```

### Parameters
- **client**: An Inngest client instance.
- **functions**: A list of Inngest functions to be served over the connection.

### Return Value
A `WorkerConnection` object.

### Usage
```python
connection = inngest.connect([(client, [func1, func2])])
await connection.start()  # Blocks until the connection closes
```
```

--------------------------------

### Asynchronous Inngest Function

Source: https://github.com/inngest/inngest-py/blob/main/README.md

An example of an asynchronous Inngest function using `httpx.AsyncClient` for making HTTP requests. This is suitable for I/O-bound operations in an async context.

```python
@inngest_client.create_function(
    fn_id="find_person",
    trigger=inngest.TriggerEvent(event="app/person.find"),
)
async def fetch_person(ctx: inngest.Context) -> dict:
    person_id = ctx.event.data["person_id"]
    async with httpx.AsyncClient(verify=False) as client:
        res = await client.get(f"https://swapi.dev/api/people/{person_id}")
        return res.json()
```

--------------------------------

### Initialize Inngest Client with Encryption

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest_encryption/README.md

Configure the Inngest client with encryption middleware by providing an encryption key and optionally fallback keys for key rotation.

```python
import inngest
from inngest_encryption import EncryptionMiddleware

inngest.Inngest(
    name="my-app",
    encryption_key="my-encryption-key",
    middleware=[EncryptionMiddleware.factory("my-secret-key")],
)
```

```python
inngest.Inngest(
    name="my-app",
    encryption_key="my-encryption-key",
    middleware=[
        EncryptionMiddleware.factory(
            "new-secret-key",
            fallback_decryption_keys=["old-secret-key"],
        ),
    ],
)
```

--------------------------------

### Run Code Quality Checks

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/CONTRIBUTING.md

Execute auto-formatting, linting, and type checking for code quality. Run these commands from the `pkg/inngest/` directory.

```bash
make format
```

```bash
make lint
```

```bash
make type-check
```

--------------------------------

### step.run() Execution Flow

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/docs/EXECUTION_MODEL.md

Illustrates the internal process when step.run() executes user-provided code. It details how the context manager, handler execution, and interrupt raising work for both success and error scenarios.

```text
report_step returns (no memo, not in parallel, is STEP_RUN)
  → enters the context manager body
  → calls the user's handler
  → on success: raises ResponseInterrupt with output + STEP_RUN opcode
  → on error: raises ResponseInterrupt with STEP_ERROR or STEP_FAILED opcode
```

--------------------------------

### Run Integration Tests for Specific Package

Source: https://github.com/inngest/inngest-py/blob/main/CONTRIBUTING.md

Change directory into a specific package and run 'make itest' to execute its integration tests.

```bash
cd pkg/inngest && make itest
```

--------------------------------

### Run Code Quality Checks

Source: https://github.com/inngest/inngest-py/blob/main/pkg/test_core/CLAUDE.md

Execute code formatting, linting, and type checking using make commands. These commands should be run from the `pkg/test_core/` directory.

```bash
make format      # Auto-format (use from monorepo root)
make lint        # Lint with ruff
make type-check  # Type check with mypy
```

--------------------------------

### Package Structure

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest_encryption/CLAUDE.md

Illustrates the directory structure of the Inngest Encryption Package.

```tree
pkg/inngest_encryption/
├── inngest_encryption/           # Main package code
│   ├── __init__.py              # Public API exports
│   ├── main.py                  # EncryptionMiddleware implementation
│   ├── _internal/               # Internal implementation
│   │   └── strategies/          # Future: Multiple encryption strategies
│   └── py.typed                 # Type information marker
├── README.md                    # Package documentation (visible on PyPI)
└── pyproject.toml              # Package configuration
```

--------------------------------

### Run Unit Tests

Source: https://github.com/inngest/inngest-py/blob/main/CONTRIBUTING.md

Run this command to execute unit tests for all packages in the monorepo.

```bash
make utest
```

--------------------------------

### Run Tests

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/CONTRIBUTING.md

Execute integration and unit tests to ensure code functionality. Integration tests are in `tests/test_inngest`, and unit tests are located next to their respective code.

```bash
make itest
```

```bash
make utest
```

--------------------------------

### Format Code

Source: https://github.com/inngest/inngest-py/blob/main/CONTRIBUTING.md

Execute this command to automatically format the code across all packages in the monorepo.

```bash
make format
```

--------------------------------

### Sending Events with Inngest Client (Sync)

Source: https://github.com/inngest/inngest-py/blob/main/README.md

Demonstrates how to send a synchronous Inngest event from a non-Inngest function using the `send_sync` method. Useful for triggering Inngest functions from existing application logic.

```python
inngest_client.send_sync(inngest.Event(name="app/test", data={"person_id": 1}))
```

--------------------------------

### Run Integration Tests

Source: https://github.com/inngest/inngest-py/blob/main/CONTRIBUTING.md

Execute this command to run integration tests for all packages within the monorepo.

```bash
make itest
```

--------------------------------

### Development Commands

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest_encryption/CLAUDE.md

Common commands for code quality and testing within the Inngest Encryption Package directory.

```bash
make format      # Auto-format (use from monorepo root)
make lint        # Lint with ruff
make type-check  # Type check with mypy
```

```bash
make itest  # Run integration tests. Tests live in `tests/test_inngest_encryption` (relative to the monorepo root)
make utest  # Run unit tests. Tests live next to the code they test
```

--------------------------------

### Sending Events with Inngest Client (Async)

Source: https://github.com/inngest/inngest-py/blob/main/README.md

Demonstrates how to send an asynchronous Inngest event from a non-Inngest function using the `send` method. This is the asynchronous equivalent of `send_sync`.

```python
await inngest_client.send(inngest.Event(name="app/test", data={"person_id": 1}))
```

--------------------------------

### Run Step with Encrypted Output

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest_encryption/README.md

Execute a step function whose entire return value will be encrypted when sent back to the Inngest server. The output is decrypted within the current function context.

```python
def _my_step() -> dict[str, object]:
    # Encrypted when sending back to the Inngest server.
    return {"msg": "hello"}

output = await step.run("my-step", _my_step)

# Decrypted within this function.
print(output)
```

--------------------------------

### Lint Specific Package

Source: https://github.com/inngest/inngest-py/blob/main/CONTRIBUTING.md

Navigate to a specific package directory and run 'make lint' to lint only that package.

```bash
cd pkg/inngest && make lint
```

--------------------------------

### Inngest Function with Step Run for Retries

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/README.md

This function demonstrates using `step.run` to wrap operations that should be retried on failure. It fetches person data and then iterates through their starship URLs, fetching each ship's details. This pattern ensures that individual steps within a function are resilient.

```python
@inngest_client.create_function(
    fn_id="find_ships",
    trigger=inngest.TriggerEvent(event="app/ships.find"),
)
def fetch_ships(ctx: inngest.ContextSync) -> dict:
    """
    Find all the ships a person has.
    """

    person_id = ctx.event.data["person_id"]

    def _fetch_person() -> dict:
        res = requests.get(f"https://swapi.dev/api/people/{person_id}", verify=False)
        return res.json()

    # Wrap the function with step.run to enable retries
    person = step.run("fetch_person", _fetch_person)

    def _fetch_ship(url: str) -> dict:
        res = requests.get(url)
        return res.json()

    ship_names = []
    for ship_url in person["starships"]:
        # step.run works in loops!
        ship = step.run("fetch_ship", _fetch_ship, ship_url)

        ship_names.append(ship["name"])

    return {
        "person_name": person["name"],
        "ship_names": ship_names,
    }
```

--------------------------------

### Parallel Execution Discovery Phase

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/docs/EXECUTION_MODEL.md

Explains the discovery phase for Group.parallel(), detailing how the 'in_parallel' context variable is set, how steps are called sequentially, and how report_step() handles new steps versus memoized ones.

```text
1. Set the `in_parallel` context variable to `True`
2. Call each callable sequentially
3. Each step calls `report_step()`, which sees `in_parallel=True` and:
   - If the step has a memo → return it (already completed)
   - If the step is new → change opcode to `PLANNED` and raise `ResponseInterrupt`
4. Collect all `ResponseInterrupt`s into a single list
5. If any new steps were discovered: raise one combined `ResponseInterrupt` with all planned steps
6. If all steps had memos: return all outputs (parallel group is fully replayed)
```

--------------------------------

### Lint Code

Source: https://github.com/inngest/inngest-py/blob/main/CONTRIBUTING.md

Run this command to lint all packages in the monorepo, checking for code style and potential errors.

```bash
make lint
```

--------------------------------

### Inngest Function with Steps for Fetching Ships

Source: https://github.com/inngest/inngest-py/blob/main/README.md

An Inngest function that uses steps to fetch a person's details and then their starships. Each step is wrapped with `step.run` for retries and transactional execution. Supports loops within steps.

```python
@inngest_client.create_function(
    fn_id="find_ships",
    trigger=inngest.TriggerEvent(event="app/ships.find"),
)
def fetch_ships(ctx: inngest.ContextSync) -> dict:
    """
    Find all the ships a person has.
    """

    person_id = ctx.event.data["person_id"]

    def _fetch_person() -> dict:
        res = requests.get(f"https://swapi.dev/api/people/{person_id}", verify=False)
        return res.json()

    # Wrap the function with step.run to enable retries
    person = step.run("fetch_person", _fetch_person)

    def _fetch_ship(url: str) -> dict:
        res = requests.get(url)
        return res.json()

    ship_names = []
    for ship_url in person["starships"]:
        # step.run works in loops!
        ship = step.run("fetch_ship", _fetch_ship, ship_url)

        ship_names.append(ship["name"])

    return {
        "person_name": person["name"],
        "ship_names": ship_names,
    }
```

```json
{
  "name": "app/ships.find",
  "data": {
    "person_id": 1
  }
}
```

--------------------------------

### Step Targeting in Parallel Execution

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/docs/EXECUTION_MODEL.md

Describes how step targeting works when the Executor needs to run a specific step from a parallel group using the 'stepId' query parameter. It explains how matching and non-matching steps are handled.

```text
When the Executor wants to run a specific step from a parallel group, it sends `stepId=<hashed_id>` in the query params. During replay:

- Steps whose hashed ID matches the target execute normally
- Steps that don't match raise `SkipInterrupt`, which the execution engine handles
- This allows the Executor to run parallel steps concurrently across multiple requests
```

--------------------------------

### Activate Virtual Environment

Source: https://github.com/inngest/inngest-py/blob/main/tests/smoke_tests/model_adapters/README.md

Ensure your Python virtual environment is activated before running the test. This command activates the environment located at .venv/bin/activate.

```bash
source .venv/bin/activate
```

--------------------------------

### Basic Inngest Function with Flask

Source: https://github.com/inngest/inngest-py/blob/main/README.md

A minimal Inngest function using Flask. It defines a function to fetch person data and registers it with the Inngest server. Requires Flask and requests libraries.

```python
import flask
import inngest.flask
import requests

inngest_client = inngest.Inngest(
    app_id="flask_example",
    is_production=False,
)

@inngest_client.create_function(
    fn_id="find_person",
    trigger=inngest.TriggerEvent(event="app/person.find"),
)
def fetch_person(ctx: inngest.ContextSync) -> dict:
    person_id = ctx.event.data["person_id"]
    res = requests.get(f"https://swapi.dev/api/people/{person_id}", verify=False)
    return res.json()

app = flask.Flask(__name__)

# Register functions with the Inngest server
inngest.flask.serve(
    app,
    inngest_client,
    [fetch_person],
)

app.run(port=8000)
```

```json
{
  "name": "app/person.find",
  "data": {
    "person_id": 1
  }
}
```

--------------------------------

### Disabling Immediate Execution for Sequential Steps

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/docs/EXECUTION_MODEL.md

Explains the 'disable_immediate_execution' flag set in the request context after parallel steps complete. This ensures subsequent sequential steps are planned rather than executed immediately to prevent duplicate runs.

```text
After parallel steps complete, the next sequential `step.run()` encountered will have `disable_immediate_execution=True` in the request context. This forces the step to be planned rather than immediately executed, ensuring that a single step isn't executed multiple times for a single run.
```

--------------------------------

### Run Specific Test

Source: https://github.com/inngest/inngest-py/blob/main/CONTRIBUTING.md

Use 'uv run pytest' with a specific test file and class/method to run a single test case.

```bash
uv run pytest tests/test_inngest/test_function/test_fast_api.py::TestFunctions::test_crazy_ids -v
```

--------------------------------

### Basic Inngest Function with Flask

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/README.md

A minimal Inngest function using Flask to handle an event. It requires the `inngest`, `inngest.flask`, and `requests` libraries. The function is registered with the Inngest server and the Flask app is run on port 8000.

```python
import flask
import inngest.flask
import requests

inngest_client = inngest.Inngest(
    app_id="flask_example",
    is_production=False,
)

@inngest_client.create_function(
    fn_id="find_person",
    trigger=inngest.TriggerEvent(event="app/person.find"),
)
def fetch_person(ctx: inngest.ContextSync) -> dict:
    person_id = ctx.event.data["person_id"]
    res = requests.get(f"https://swapi.dev/api/people/{person_id}", verify=False)
    return res.json()

app = flask.Flask(__name__)

# Register functions with the Inngest server
inngest.flask.serve(
    app,
    inngest_client,
    [fetch_person],
)

app.run(port=8000)
```

--------------------------------

### Type Check Code

Source: https://github.com/inngest/inngest-py/blob/main/CONTRIBUTING.md

Use this command to perform type checking across all packages in the monorepo.

```bash
make type-check
```

--------------------------------

### Inngest Interrupts for Control Flow

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest/docs/EXECUTION_MODEL.md

Details the custom interrupts used by the Inngest SDK for internal control flow. These exceptions extend BaseException to avoid accidental catching by user code.

```text
Interrupt             | Extends         | Purpose                                                                                  
----------------------|-----------------|----------------------------------------------------------------------------------------
`ResponseInterrupt`   | `BaseException` | A step was planned or executed; carries the step response(s) back to `ExecutionV0.run()` 
`SkipInterrupt`       | `BaseException` | Step targeting is active and this step isn't the target                                  
`NestedStepInterrupt` | `BaseException` | A step was called inside another step's handler (not supported)                          
```

--------------------------------

### Invoke Function with Encrypted Data

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest_encryption/README.md

Invoke a function with data where the 'encrypted' field will be automatically encrypted by the middleware. Other fields are not encrypted.

```python
await step.invoke(
    "invoke",
    function=child_fn_async,
    data={
        # Everything in this field will be encrypted.
        "encrypted": {
            "phone": "867-5309",
        },

        # Not encrypted.
        "user_id": "abc123",
    },
)
```

--------------------------------

### Send Encrypted Event Data

Source: https://github.com/inngest/inngest-py/blob/main/pkg/inngest_encryption/README.md

Send an event where specific data fields within the 'encrypted' key are automatically encrypted by the middleware. Non-'encrypted' fields remain unencrypted.

```python
await step.send_event(
    "my-step",
    inngest.Event(
        data={
            # Everything in this field will be encrypted.
            "encrypted": {
                "phone": "867-5309",
            },

            # Not encrypted.
            "user_id": "abc123",
        },
        name="my-event",
    ),
)
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.