> ## Documentation Index > Fetch the complete documentation index at: https://docs.ntropii.com/llms.txt > Use this file to discover all available pages before exploring further. # ntro.workflow > API reference for NtroWorkflow, @runbook.step, and the runbook orchestration primitives. `ntro.workflow` is the orchestration surface every runbook subclasses. For the concept-level walkthrough of authoring a runbook, see [Build runbooks](/workflows/runbooks/overview). This page is the API reference. ## Install ```bash theme={null} pip install 'ntro[workflow]' ``` The `[workflow]` extra brings in Temporal, all `ntro.capabilities.*` modules, `ntro.events`, `ntro.accounting`, and `ntro.data` — everything a runbook needs at runtime. ## Surface ```python theme={null} from ntro.workflow import ( NtroWorkflow, runbook, # facade — .defn .run .signal .query .step # .execute_activity .wait_condition # .info .now .logger activity, # facade — .defn .info .logger WorkflowError, # raised inside activities; the worker marks it # non-retryable ) ``` | Symbol | Kind | Purpose | | -------------------------- | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `NtroWorkflow` | class | Base class for all runbooks. Subclass + decorate with `@runbook.defn`. | | `runbook` | facade object | Mirrors `temporalio.workflow` — decorators (`.defn`, `.run`, `.signal`, `.query`, `.step`) and runtime helpers (`.execute_activity`, `.wait_condition`, `.info`, `.now`, `.logger`). | | `activity` | facade object | Mirrors `temporalio.activity` — `.defn` decorator, `.info()`, `.logger`. | | `WorkflowError` | exception | Raised inside activities to fail the workflow. `non_retryable=True` flags deterministic-input failures so retries don't loop. | | `await_signal_with_action` | method on `NtroWorkflow` | Block until a predicate is satisfied; advertise the pending action via `current_pending_action`. | | `wait_for_action` | method on `NtroWorkflow` | Block on a HITL approve / reject / correct signal from the Tenant UI. | | `run_child_workflow` | method on `NtroWorkflow` | Dispatch another runbook by slug. Children appear as nested progress trees in the UI. | | `previous` | function | Load committed-document context from a prior task. See [`ntro.workflow.task`](/sdk/task). | ## Runtime backends `runbook` and `activity` dispatch to whichever backend is auto-selected at first attribute access: | Backend | Selected when | Behaviour | | ------------------- | ---------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **TemporalRuntime** | `temporalio` is importable AND `TEMPORAL_HOST` is set — i.e. inside an `ntro-worker` pod | Every facade call passes straight through to `temporalio.workflow.X` / `temporalio.activity.X`. Production behaviour, unchanged from pre-N-102. | | **LocalRuntime** | Anywhere else — local dev, CI smoke jobs, Claude MA sandbox | Activities run as direct `await fn(...)` coroutines; child workflows run on the same event loop; signals are plain method calls; `wait_condition` polls at 50ms granularity. No durability, no retries, no replay — exceptions bubble up immediately. | One INFO log line on first selection so it's obvious which is active: ``` ntro.workflow: TemporalRuntime active (TEMPORAL_HOST=…) ntro.workflow: LocalRuntime active (no Temporal detected; running in-process) ``` To force a specific runtime in tests, call `ntro.workflow.set_runtime(...)` **before** importing any runbook module. Otherwise the auto-selection runs once and is cached. ## `NtroWorkflow` ```python theme={null} from ntro.workflow import NtroWorkflow, runbook @runbook.defn class MyWorkflow(NtroWorkflow): @runbook.run async def run(self, ctx): ... ``` Subclasses get four auto-wired queries / signals at construction: | Name | Kind | Returns / accepts | Purpose | | ------------------------ | ------ | --------------------------------- | ------------------------------------------------------ | | `current_pending_action` | query | `{action, display_hint}` | What is the workflow waiting for? Polled by Tenant UI. | | `current_steps` | query | `list[{id, title, status, icon}]` | The `@runbook.step` ladder + per-step status. | | `current_ui_state` | query | combined snapshot | One-shot read for UI rendering. | | `user_action` | signal | `{action, payload}` | UI dispatches approve / reject / correct here. | Your subclass doesn't need to register these — the base class does it. Add domain-specific signals (`tb_submitted`, `document_submitted`, …) as needed. ## `@runbook.step` ```python theme={null} @runbook.step(name="period_open", title="Open period", icon="Calendar") async def _step_period_open(self, ctx): ... ``` | Arg | Type | Purpose | | ------- | ----- | ------------------------------------------------------------------------ | | `name` | `str` | Stable step id (used in `current_steps`, snapshots, child-workflow ids). | | `title` | `str` | Display name in the Tenant UI breadcrumb. | | `icon` | `str` | [Lucide](https://lucide.dev) icon name. | The decorator wraps the method so step lifecycle (`_current_step_id`, `_steps_completed`) tracks automatically. Class-definition order is breadcrumb order in the UI sidebar. ## `wait_for_action` Block on a HITL signal. The `display_hint` tells the Tenant UI which review component to render. ```python theme={null} response = await self.wait_for_action( payload=extracted, display_hint={ "type": "review_extraction", "schema_slug": ctx.schema_slug, }, reason="Awaiting accountant approval of extraction", ) match response.action: case "approved": ... case "rejected": ... case "corrected": ... ``` | Arg | Type | Purpose | | -------------- | ----------------------------- | ---------------------------------------------------------- | | `payload` | `Any` (pydantic-serialisable) | What the user is reviewing — rendered by the UI component. | | `display_hint` | `dict` | UI component selector + config. | | `reason` | `str` | One-line description shown in the pending-action chip. | Returns `UserAction(action, payload, corrections)`. `corrections` is workflow-specific (e.g. row-level edits to an extracted document). ## `await_signal_with_action` Block until a predicate is true. Advertises the pending action via `current_pending_action` so the UI knows what to surface. ```python theme={null} await self.await_signal_with_action( predicate=lambda: self._submission_received, action="awaiting_document_submission", display_hint={"source": ctx.source, "filename_pattern": ctx.expected_pattern}, ) ``` Useful when the workflow waits for an upload, an email arrival, or any human-driven event without a fixed schedule. Pair with a `@runbook.signal` handler that mutates the predicate's state. ## `run_child_workflow` Dispatch another runbook. Children get their own progress tree under the parent's step. ```python theme={null} result = await self.run_child_workflow( slug="document-ingest", input=DocumentIngestContext( period=ctx.period, entity_slug=ctx.entity_slug, source=expected.source, schema=expected.schema, ), step_id="ingest_documents", ) ``` | Arg | Type | Purpose | | --------- | -------------- | ---------------------------------------------------------------------- | | `slug` | `str` | The child runbook's slug (registered with the worker). | | `input` | pydantic model | Child workflow input. Must match the child's declared `Context` model. | | `step_id` | `str` | Parent step id this child belongs to — drives UI nesting. | Cardinality `many` (fan-out): call `run_child_workflow` in a loop; awaits join. ## `WorkflowError` Raised inside an activity to fail the workflow. Replaces the legacy `from temporalio.exceptions import ApplicationError` import. ```python theme={null} from ntro.workflow import WorkflowError @activity.defn(name="commit_starting_tb") async def commit_starting_tb(input): if input.period not in input.extracted.fields: raise WorkflowError( f"Period {input.period} missing from extraction", type="EXTRACTION_INPUT_INVALID", non_retryable=True, ) ... ``` | Arg | Type | Purpose | | --------------- | ------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `message` | `str` | Human-readable failure message. | | `type` | `str \| None` | Error type tag — the worker interceptor uses this to discriminate deterministic-input failures (e.g. `"VALIDATION"`) from infra failures. | | `non_retryable` | `bool` | When `True`, the worker marks the failure as non-retryable so the workflow fails fast rather than re-attempting the activity. Use for deterministic-input failures where retrying can't change the outcome. | Under `TemporalRuntime`, `WorkflowError` subclasses `temporalio.exceptions.ApplicationError` so the ntro-worker's existing non-retryable interceptor catches it unchanged. Under `LocalRuntime` it subclasses `Exception` directly — `non_retryable` has no effect (there are no retries to suppress) and the exception bubbles up to whoever awaited the activity. For Temporal's `ActivityError` wrapping pattern (used in the journal-proposer's HITL-vs-propagate discrimination), also re-exported: ```python theme={null} from ntro.workflow.exceptions import ActivityError, ApplicationError ``` Under `LocalRuntime` both alias to `WorkflowError` so the `except (ActivityError, ApplicationError)` clause parses — though no exception is actually wrapped in `ActivityError` under `LocalRuntime`, the `ApplicationError` branch picks up `WorkflowError` directly. ## Running locally `ntro.workflow.run_local(cls, arg, *, workflow_id=..., task_queue=...)` is the entry helper for `python -m runbooks.X`-style scripts. Instantiates the runbook, sets up the `WorkflowInfo` contextvar, and awaits `cls().run(arg)`. See [Running locally](/workflows/runbooks/running-locally) for the full walkthrough — when `LocalRuntime` is selected, what works and what doesn't, and how to force a specific runtime from test fixtures. ## Related Worked walkthrough of authoring a runbook from scratch. Load context from prior task executions. Invoke a registered external agent from inside a `@runbook.step`. Run a runbook end-to-end in-process under `LocalRuntime` — no Temporal cluster, sub-second feedback. How Tenant UI actions reach workflows end-to-end.