Skip to main content

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

SymbolKindPurpose
NtroWorkflowclassBase class for all runbooks. Subclass + decorate with @workflow.defn.
ui_stepdecoratorMark a method as a step; threads UI metadata (name, title, icon) into the breadcrumb.
await_signal_with_actionmethod on NtroWorkflowBlock until a predicate is satisfied; advertise the pending action via current_pending_action.
wait_for_actionmethod on NtroWorkflowBlock on a HITL approve / reject / correct signal from the Tenant UI.
run_child_workflowmethod on NtroWorkflowDispatch another runbook by slug. Children appear as nested progress trees in the UI.
previousfunctionLoad committed-document context from a prior task. See ntro.workflow.task.

​
NtroWorkflow

from temporalio import workflow
from ntro.workflow import NtroWorkflow, ui_step

@workflow.defn
class MyWorkflow(NtroWorkflow):
    @workflow.run
    async def run(self, ctx): ...
Subclasses get four auto-wired queries / signals at construction:
NameKindReturns / acceptsPurpose
current_pending_actionquery{action, display_hint}What is the workflow waiting for? Polled by Tenant UI.
current_stepsquerylist[{id, title, status, icon}]The @ui_step ladder + per-step status.
current_ui_statequerycombined snapshotOne-shot read for UI rendering.
user_actionsignal{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.

​
@ui_step

@ui_step(name="period_open", title="Open period", icon="Calendar")
async def _step_period_open(self, ctx): ...
ArgTypePurpose
namestrStable step id (used in current_steps, snapshots, child-workflow ids).
titlestrDisplay name in the Tenant UI breadcrumb.
iconstrLucide 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": ...
ArgTypePurpose
payloadAny (pydantic-serialisable)What the user is reviewing — rendered by the UI component.
display_hintdictUI component selector + config.
reasonstrOne-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 @workflow.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",
)
ArgTypePurpose
slugstrThe child runbook’s slug (registered with the worker).
inputpydantic modelChild workflow input. Must match the child’s declared Context model.
step_idstrParent step id this child belongs to — drives UI nesting.
Cardinality many (fan-out): call run_child_workflow in a loop; awaits join.

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

UI and Temporal signals

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