### Run Example Programs

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/README.md

Examples are located in the `examples/` directory. Start a Resonate server with `resonate dev` on `localhost:8001` before running these commands.

```shell
uv run python examples/hello-world
```

```shell
uv run python examples/fibonacci --mode rpc --n 10
```

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

### Start Resonate Development Server

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/README.md

Start the Resonate development server using the CLI.

```shell
resonate dev
```

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

### Running the Recovery Example

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/recovery/README.md

To run this example, ensure a Resonate server is running on localhost:8001. Then, execute the provided command in your terminal.

```sh
uv run python examples/recovery
```

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

### Install Resonate Server & CLI

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/README.md

Use Homebrew to install the Resonate server and command-line interface.

```shell
brew install resonatehq/tap/resonate
```

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

### Install Resonate Python SDK

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/README.md

Install the Resonate Python SDK using pip.

```shell
pip install resonate-sdk
```

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

### Run Saga Example with Different Failure Scenarios

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/saga/README.md

Execute the saga example with options to simulate failures at different stages. This helps test the compensation logic and error handling.

```sh
uv run python examples/saga/main.py

# Force charge_card to fail. The orchestrator runs release_hotel and
# release_flight in reverse order before returning the error.
uv run python examples/saga/main.py --fail charge

# Force reserve_hotel to fail. Only release_flight runs (no hotel ref yet).
uv run python examples/saga/main.py --fail hotel
```

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

### Run Human-in-the-loop Example

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/human-in-the-loop/README.md

Execute the human-in-the-loop example with different decision outcomes. Ensure a Resonate server is running on localhost:8001.

```sh
uv run python examples/human

# Rejected path: simulated reviewer rejects; cancel_order runs instead.
uv run examples/human --decision reject
```

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

### Run Pipeline Command

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/pipeline/README.md

Executes the Python pipeline example using `uv run`. Ensure a Resonate server is running on `localhost:8001`.

```sh
uv run python examples/pipeline/main.py

```

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

### Write a Durable Countdown Function

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/README.md

Example of a durable countdown function using Resonate SDK. This function can run for extended periods and survive worker restarts.

```python
import asyncio
from datetime import timedelta

from resonate.context import Context
from resonate.resonate import Resonate


async def countdown(ctx: Context, count: int, delay: int) -> None:
    for i in range(count, 0, -1):
        # Run a function durably, awaiting its persisted result
        await ctx.run(ntfy, i)
        # Sleep durably -- the worker holds no state while suspended
        await ctx.sleep(timedelta(seconds=delay))
    print("Done!")


async def ntfy(ctx: Context, i: int) -> None:
    print(f"Countdown: {i}")


async def main() -> None:
    # Connect to the Resonate server and register the functions
    resonate = Resonate(url="http://localhost:8001")
    resonate.register(countdown)
    resonate.register(ntfy)

    try:
        # Invoke with execution id "countdown.1"; run() returns a handle
        # immediately, result() awaits the durable outcome
        handle = resonate.run("countdown.1", countdown, 5, 1)
        await handle.result()
    finally:
        await resonate.stop()


if __name__ == "__main__":
    asyncio.run(main())

```

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

### Expected Output for `--fail charge` Scenario

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/saga/README.md

Illustrates the expected log output when the `charge_card` step fails, showing the sequence of operations including compensations.

```text
[book_trip] starting workflow id=saga-... fail_at='charge'
  [reserve_flight] reserved FL-alice-SFO-JFK
  [reserve_hotel] reserved HT-alice-JFK
  [charge_card] FAILED for alice ($850)
  [release_hotel] released HT-alice-JFK
  [release_flight] released FL-alice-SFO-JFK
[book_trip] FAILED: ... card declined for $850
```

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

### Download Stage Implementation

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/pipeline/README.md

Fetches data from a given URL. This stage is a leaf node in the DAG and is designed to be idempotent.

```python
import httpx

from examples.pipeline.types import RpcClient


def download(rpc: RpcClient, url: str) -> bytes:
    """Downloads content from a URL.

    This function is a leaf stage and will only run once.
    """
    response = httpx.get(url)
    response.raise_for_status()
    print(f"  [download] {{url}} -> {{len(response.content)}} bytes")
    return response.content

```

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

### Expected Pipeline Output

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/pipeline/README.md

Illustrates the expected log output when the pipeline runs successfully. Note that the order of 'transform_a' and 'transform_b' logs may vary due to concurrent execution.

```text
[run_pipeline] starting workflow id=pipeline-...
  [download] example.com/doc -> 46 bytes
  [parse] 7 words
  [transform_a] counting words
  [transform_b] uppercased 46 chars
  [merge] combining transforms
  [emit] words=7 upper='THE QUICK BROWN FOX JUMPS OVER EXAMPLE.COM/DOC'
[run_pipeline] OK: sent='ok'

```

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

### Parse Stage Implementation

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/pipeline/README.md

Parses the downloaded content. This stage depends on the 'download' stage.

```python
from examples.pipeline.types import RpcClient


def parse(rpc: RpcClient, data: bytes) -> int:
    """Parses the downloaded data and returns the word count.

    This function is a leaf stage and will only run once.
    """
    words = len(data.split())
    print(f"  [parse] {{words}} words")
    return words

```

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

### Inspect Resonate Execution Tree

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/README.md

Visualize the call graph of a Resonate function execution using the `resonate tree` command.

```shell
resonate tree countdown.1
```

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

### Applying Retry Policy per ctx.run Call

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/retries/README.md

Demonstrates overriding the retry policy for a specific `ctx.run` call using `with_opts`. This allows for fine-grained control over retries for individual operations within a workflow.

```python
ctx.with_opts(retry_policy=Constant(max_retries=5, delay=0)).run(charge, 100)
```

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

### Running a Flaky Leaf Function with Retries

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/retries/README.md

Demonstrates invoking a flaky leaf function through various entrypoints. Resonate retries the leaf function based on its retry policy until it succeeds.

```python
import asyncio

import resonate
from resonate.retry import Constant


async def charge(amount: int) -> str:
    print(f"  [{asyncio.current_task().get_name()}] attempt 1...")
    await asyncio.sleep(0.1)
    print(f"  [{asyncio.current_task().get_name()}] attempt 2...")
    await asyncio.sleep(0.1)
    print(f"  [{asyncio.current_task().get_name()}] attempt 3...")
    await asyncio.sleep(0.1)
    return f"charged ${amount}"


async def checkout() -> str:
    return f"{await ctx.run(charge, 300)} | {await ctx.rpc('charge', 400)}"
```

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

### Resolve Promise Externally with Python SDK

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/human-in-the-loop/README.md

Demonstrates how to resolve a durable promise from an external process using the Resonate Python SDK. This is useful for integrating with external systems or manual interventions.

```python
import asyncio
from resonate.resonate import Resonate
from resonate.types import Value

async def main():
    r = Resonate(url="http://localhost:8001")
    await r.promises.resolve(
        "fulfill-1234567890.1",
        Value.from_serializable({"approve": True, "note": "lgtm"}),
    )
    await r.stop()

asyncio.run(main())
```

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

### Define DAG Workflow with Resonate SDK

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/pipeline/README.md

Orchestrates a multi-stage pipeline with sequential and parallel dependencies. Stages are dispatched via `ctx.rpc` and backed by durable promises for idempotent recovery.

```python
from typing import Any

from examples.pipeline.stages import download, parse, transform_a, transform_b, merge, emit
from examples.pipeline.types import PipelineContext


def run_pipeline(ctx: PipelineContext) -> Any:
    """Runs the pipeline.

    download → parse → ┬─ transform_a ─┐
                       └─ transform_b ─┴─ merge → emit
    """
    # Sequential dependency
    data = await download(ctx.rpc, "example.com/doc")
    parsed_data = await parse(ctx.rpc, data)

    # Parallelism inside a stage
    transform_a_future = transform_a(ctx.rpc, parsed_data)
    transform_b_future = transform_b(ctx.rpc, parsed_data)

    # Await both parallel branches
    transformed_a, transformed_b = await asyncio.gather(
        transform_a_future, transform_b_future
    )

    # Merge and emit
    merged_data = await merge(ctx.rpc, transformed_a, transformed_b)
    result = await emit(ctx.rpc, merged_data)

    return result

```

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

### Run the Resonate Worker

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/README.md

Execute the Python script to run the Resonate worker.

```shell
python countdown.py
```

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

### Checkout Orchestrator and Leaf Functions

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/recovery/README.md

This Python code defines the main orchestrator 'checkout' and two leaf functions, 'summarize' and 'pay'. 'summarize' takes a list of items and returns a typed Cart object. 'pay' takes a Cart object and returns a Receipt object. These functions demonstrate typed data exchange across the durability boundary.

```python
from typing import List

import msgspec

from examples.recovery.receipt import Receipt
from examples.recovery.cart import Cart
from examples.recovery.durable import DurableFunction


class Checkout(DurableFunction):
    async def summarize(self, items: List[str]) -> Cart:
        return Cart(items=items, total=len(items) * 10)

    async def pay(self, cart: Cart) -> Receipt:
        print(f"charging {cart.total} for {len(cart.items)} items")
        await self.sleep(1)
        return Receipt(cart=cart, paid=True)

    async def checkout(self, items: List[str]) -> Receipt:
        cart = await self.summarize(items)
        return await self.pay(cart)

```

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

### Resonate SDK Default Retry Policy

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/retries/README.md

Illustrates setting a default retry policy for the entire Resonate SDK instance. This policy is used if no other policy is specified.

```python
resonate.Resonate(retry_policy=Exponential())
```

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

### Transform A Stage Implementation

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/pipeline/README.md

Performs a transformation on the parsed data, specifically counting words. This stage runs in parallel with 'transform_b'.

```python
from examples.pipeline.types import RpcClient


def transform_a(rpc: RpcClient, parsed_data: int) -> str:
    """Counts the words (no-op here, just logs).

    This function is a leaf stage and will only run once.
    """
    print(f"  [transform_a] counting words")
    # In a real scenario, this would do something with parsed_data
    return f"counted {parsed_data} words"

```

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

### Transform B Stage Implementation

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/pipeline/README.md

Performs a transformation on the parsed data, specifically uppercasing it. This stage runs in parallel with 'transform_a'.

```python
from examples.pipeline.types import RpcClient


def transform_b(rpc: RpcClient, data: bytes) -> str:
    """Uppercases the original downloaded content (simulated).

    This function is a leaf stage and will only run once.
    """
    # In a real scenario, this would use the original data, not parsed_data
    # For simplicity, we'll just use a placeholder string here.
    print(f"  [transform_b] uppercased 46 chars")
    return "THE QUICK BROWN FOX JUMPS OVER EXAMPLE.COM/DOC"

```

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

### Registering a Function with a Specific Retry Policy

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/retries/README.md

Shows how to register a function with Resonate and assign a specific retry policy to it. This policy will be used when the function is invoked.

```python
resonate.register(charge, retry_policy=Constant(max_retries=5, delay=0))
```

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

### Emit Stage Implementation

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/pipeline/README.md

The final stage of the pipeline, emitting the merged results. This stage depends on the 'merge' stage completing.

```python
from examples.pipeline.types import RpcClient


def emit(rpc: RpcClient, merged_data: dict) -> dict:
    """Emits the final merged data.

    This function is a leaf stage and will only run once.
    """
    print(f"  [emit] {{merged_data['words']}} {{merged_data['upper']}}")
    return merged_data

```

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

### Receipt Data Structure

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/recovery/README.md

Defines the Receipt data structure using msgspec.Struct. It contains the Cart object and a boolean indicating payment status.

```python
import msgspec

from examples.recovery.cart import Cart


class Receipt(msgspec.Struct):
    cart: Cart
    paid: bool

```

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

### Merge Stage Implementation

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/pipeline/README.md

Merges the results from the parallel 'transform_a' and 'transform_b' stages. This stage depends on both parallel branches completing.

```python
from examples.pipeline.types import RpcClient


def merge(rpc: RpcClient, transformed_a: str, transformed_b: str) -> dict:
    """Merges the results from the two transform stages.

    This function is a leaf stage and will only run once.
    """
    print(f"  [merge] combining transforms")
    return {
        "words": transformed_a,
        "upper": transformed_b,
    }

```

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

### Cart Data Structure

Source: https://github.com/resonatehq/resonate-sdk-py/blob/main/examples/recovery/README.md

Defines the Cart data structure using msgspec.Struct for typed data exchange. It includes a list of items and a total amount.

```python
import msgspec


class Cart(msgspec.Struct):
    items: List[str]
    total: int

```

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