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.

Every Ntropii runbook subclasses NtroWorkflow and decorates each phase with @ui_step. The decorator both attaches metadata for the Tenant UI breadcrumb and wraps the method so step lifecycle (_current_step_id, _steps_completed) tracks automatically.

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

​
The shape of a runbook

from datetime import timedelta
from typing import Any
from temporalio import workflow

from ntro.workflow import NtroWorkflow, ui_step

from .activities import open_period, check_starting_tb, emit_period_summary
from .models import (
    NavMonthlyContext,
    PeriodSummary,
    PeriodSummaryInput,
    StartingTBVerified,
)


@workflow.defn
class NavMonthlyWorkflow(NtroWorkflow):
    def __init__(self) -> None:
        super().__init__()
        self._tb_signal_received: bool = False

    @workflow.signal
    def tb_submitted(self, payload: dict[str, Any] | None = None) -> None:
        """Worker emits this signal once the period-end TB is in the data plane."""
        self._tb_signal_received = True

    # ── Steps (declaration order = breadcrumb order) ───────────────

    @ui_step(name="period_open", title="Open period", icon="Calendar")
    async def _step_period_open(self, ctx: NavMonthlyContext) -> Any:
        return await workflow.execute_activity(
            open_period,
            ctx,
            start_to_close_timeout=timedelta(minutes=5),
        )

    @ui_step(name="check_starting_tb", title="Verify Starting TB", icon="ClipboardCheck")
    async def _step_check_tb(self, ctx: NavMonthlyContext) -> StartingTBVerified:
        await workflow.wait_condition(lambda: self._tb_signal_received)
        return await workflow.execute_activity(
            check_starting_tb,
            ctx,
            start_to_close_timeout=timedelta(minutes=2),
        )

    @ui_step(name="summary", title="Period summary", icon="CheckCircle2")
    async def _step_summary(self, input: PeriodSummaryInput) -> PeriodSummary:
        return await workflow.execute_activity(
            emit_period_summary,
            input,
            start_to_close_timeout=timedelta(minutes=5),
        )

    @workflow.run
    async def run(self, ctx: NavMonthlyContext) -> PeriodSummary:
        await self._step_period_open(ctx)
        verified = await self._step_check_tb(ctx)
        return await self._step_summary(
            PeriodSummaryInput(period=ctx.period, entity=ctx.entity_slug, ...)
        )
Two things to notice:
  • @workflow.defn comes from temporalio — the workflow runs on Temporal under the hood.
  • @ui_step comes from ntro.workflow — it threads UI metadata into the breadcrumb and wires the step lifecycle, so the Tenant UI can render the right component for this step.
Class-definition order is the breadcrumb order in the UI sidebar (Python preserves __dict__ insertion order). Icons are Lucide names rendered by the Tenant UI.

​
What NtroWorkflow gives you

Beyond the bare Temporal workflow surface, NtroWorkflow bakes in the patterns runbooks need:
Block on a human approve / reject / correct signal:
response = await self.wait_for_action(
    payload=extracted,
    display_hint={
        "type": "review_extraction",
        "schema_slug": ctx.schema_slug,
    },
    reason="Awaiting accountant approval of extraction",
)
if response.action == "approved":
    ...
elif response.action == "rejected":
    ...
The display_hint tells the Tenant UI which review component to render (extraction review, journal review, NAV signoff, etc.). The signal is dispatched from the UI when the human clicks Approve/Reject.
Block until an external signal satisfies a predicate, advertising what’s needed via the current_pending_action query:
submitted = 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 that doesn’t have a fixed schedule.
Dispatch another runbook by slug:
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",
)
Children appear as their own progress trees in the Tenant UI under the parent’s step. Cardinality many lets you fan out and join.
The base class registers current_pending_action, current_steps, and current_ui_state queries plus the user_action signal handler at construction. Your subclass doesn’t need to do anything — the Tenant UI starts polling them as soon as the workflow exists.

​
Observable results — ntro.events

For values that update over time (extraction results being corrected, journals being adjusted), wrap them in ObservableResult so the Tenant UI can render live updates without re-fetching: 
from ntro.events import ObservableResult

class ExtractedPayload(BaseModel):
    fields: dict[str, str]
    confidence_scores: dict[str, float]

    # Mark as observable so corrections stream to the UI
    extracted: ObservableResult[ExtractedPayload]
ObservableResult lives in ntro.events and is used by every runbook template’s models.py. It’s the bridge between activity outputs and the UI’s live-update channel.

​
Reference

Class / decoratorPurpose
NtroWorkflowBase class for all runbooks
@ui_step(name=, title=, icon=)Mark a method as a step; threads UI metadata
wait_for_action(payload, display_hint, reason)Block on a HITL signal
await_signal_with_action(predicate, action, display_hint)Block on external signals with UI advertising
run_child_workflow(slug, input, step_id)Dispatch another runbook
current_pending_action (query)What is the workflow waiting for?
current_steps (query)List of @ui_step-tagged steps and their statuses
current_ui_state (query)Combined snapshot for the UI
user_action (signal)Receives approve/reject/correct from the UI

Skill definitions

The runbook.md frontmatter that drives @ui_step placement.

Testing

Run your NtroWorkflow subclass locally with WorkflowHarness.