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.

A workflow is a binding of a runbook (the deterministic Python templates that drive the process) to an entity, optionally on a schedule. A runbook may invoke external agents (Claude Managed, Copilot, …) from inside its steps when it needs LLM-driven artefact generation that doesn’t fit the deterministic capabilities surface. This page is the concept-level walkthrough for authoring a runbook. For API reference, see ntro.workflow. For the LLM-facing layer of a runbook, see skill definitions. To delegate a step to an external agent, see Register agents.

​
Anatomy of a runbook bundle

A runbook lives in ntro-runbook-templates/runbooks/<slug>/: 
runbooks/nav-monthly/
├── runbook.md              # frontmatter (slug, version, schedule, config schema)
├── skill.md                # LLM-facing skill definition
└── templates/
    ├── workflow.py         # @workflow.defn class, @ui_step methods
    ├── activities.py       # @activity.defn — the deterministic side-effects
    └── models.py           # Pydantic models for activity payloads + returns
The worker scans /workflows/<slug>/<version>/ for bundles at boot and registers each runbook’s workflow + activities with Temporal. Deploys land under that path via ntro workflow create --path ….

​
The shape of a runbook

Every 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. 
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.
Delegate a step to a registered external agent (Claude Managed, Copilot, …):
from ntro.workflow.agents import invoke

result = await invoke(
    ctx.audit_agent_id,
    input={"period_summary": period_summary},
    tenant_slug=ctx.tenant_slug,
    entity_slug=ctx.entity_slug,
    task_id=ctx.task_id,
)
The adapter starts a session on the agent’s host platform, polls until terminal, pulls any output files, and persists them to ingest.submitted_documents with source='agent_output:<agent_id>'. See Register agents for the agent-side lifecycle.
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.

Skill definitions

The runbook.md / skill.md frontmatter that the platform’s runbook-creation assistant reads when guiding a user through configuring a workflow.

ntro.workflow API reference

Signatures and behaviour of NtroWorkflow, @ui_step, wait_for_action, await_signal_with_action, run_child_workflow.

UI and Temporal signals

How Tenant UI actions reach workflows (approve, row_action) end-to-end.

Ingest contracts

ntro.ingest contracts and set_user_feedback for upload steps.

Register agents

Bring an external agent (Claude Managed, Copilot) into a runbook step.

Testing

Run your NtroWorkflow subclass locally with WorkflowHarness.