ntro.workflow

ntro.workflow is the orchestration surface every runbook subclasses. For the concept-level walkthrough of authoring a runbook, see Build runbooks. This page is the API reference.

Install

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

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`.

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`

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`

@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 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.

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.

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.

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.

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:

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 for the full walkthrough — when LocalRuntime is selected, what works and what doesn’t, and how to force a specific runtime from test fixtures.

Build runbooks (concept)

Worked walkthrough of authoring a runbook from scratch.

ntro.workflow.task

Load context from prior task executions.

ntro.workflow.agents

Invoke a registered external agent from inside a @runbook.step.

Running locally

Run a runbook end-to-end in-process under LocalRuntime — no Temporal cluster, sub-second feedback.

UI and Temporal signals

How Tenant UI actions reach workflows end-to-end.

Get Started

Tooling

Building workflows

Ntro SDK

API Reference

Deploy & Run

Install

Surface

Runtime backends

`NtroWorkflow`

`@runbook.step`

`wait_for_action`

`await_signal_with_action`

`run_child_workflow`

`WorkflowError`

Running locally

Build runbooks (concept)

ntro.workflow.task

ntro.workflow.agents

Running locally

UI and Temporal signals

​Install

​Surface

​Runtime backends

​NtroWorkflow

​@runbook.step

​wait_for_action

​await_signal_with_action

​run_child_workflow

​WorkflowError

​Running locally

​Related

Build runbooks (concept)

ntro.workflow.task

ntro.workflow.agents

Running locally

UI and Temporal signals

Install

Surface

Runtime backends

`NtroWorkflow`

`@runbook.step`

`wait_for_action`

`await_signal_with_action`

`run_child_workflow`

`WorkflowError`

Running locally

Related