# Quickstart

> Define a workflow, submit a run, execute it in-process.

Source: https://docs.pyrula.io/workflows/quickstart/

Install `"pyrula[workflows]"`, then scaffold a runnable workflow:

```bash
pip install "pyrula[workflows]"
pyrula init --workflow
pyrula dev app.py --open
```

Until the first public release the packages need a source build; see [Install](https://docs.pyrula.io/getting-started/install.md).

## Generated workflow

`pyrula init --workflow` writes `app.py`: reserve inventory, charge a card, sleep, send a receipt. The [getting-started quickstart](https://docs.pyrula.io/getting-started/quickstart.md) shows the file and its output.

## Send this to Cloud

Export the key from Cloud's **Setup** and restart `dev`. It prints `Cloud: connected to ...` and a `Cloud: run <run-id> -> <link>` line.

```bash
export PYRULA_CLOUD_API_KEY="<key>"
pyrula dev app.py
```

See [Connect to Pyrula Cloud](https://docs.pyrula.io/getting-started/cloud.md) for the full output.

## Minimal workflow

```python
import asyncio
from pyrula.workflows import step, workflow, InlineWorkflowRunner, MemoryStore


@step
def build(name: str) -> str:
    return f"hello, {name}"


@workflow(name="greet")
async def greet(ctx, name: str) -> str:
    return await build(name)


async def main() -> None:
    runner = InlineWorkflowRunner(store=MemoryStore())
    run = await runner.client.submit("greet", {"name": "ada"})
    print(await run.result())        # hello, ada
    for event in await run.events():
        print(event.kind, event.payload)


asyncio.run(main())
```

`submit` queues, `result()` runs it in this process, `events()` returns the journal. `MemoryStore` is for tests; `SQLiteStore("pyrula.db")` and Valkey survive restarts ([Choosing a store](https://docs.pyrula.io/getting-started/choosing-a-store.md)).

## Parameters

Params are validated against the body's annotations, and a dataclass, Pydantic model or `list[Order]` arrives as that type. Bad or missing input ends the run with `run:error` code `invalid_request` (the error never echoes the input). An annotation that can't become a schema raises `TypeError` at definition. Concurrency and rate keys read the raw submitted JSON, before coercion.

## Options

```python
@workflow(
    name="report",
    timeout=600.0,                  # per attempt, default 300
    max_event_count_per_run=10_000,     # default 50,000
    max_event_payload_bytes=1_000_000,  # the default; None lifts the limit
    max_attempt_wall_seconds=3600.0,    # min'd with timeout, still per attempt
)
async def report(ctx):
    ...
```

A run with more than `max_event_count_per_run` events ends the same way, its `reason` pointing at `ctx.continue_as_new`; the default is 50,000 and `None` declares none, and a store that trims a run's stream (`ValkeyStore`, below its `maxlen`) lowers it to what the store keeps either way. `ctx.continue_suggested` turns true at 80% of it. A payload over `max_event_payload_bytes` (a step result, submit `params`) ends the run with `stream:overflow` then `run:error` (`stream_too_large`) naming the step. An oversized submit is refused with `PayloadTooLarge` before anything is written. Don't catch it in the body: once a cap trips, later cap checks for that attempt are off.

Next: [Runs as Event Streams](https://docs.pyrula.io/workflows/concepts/runs-as-streams.md).