Build runbooks

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 runbook-templates/runbooks/<slug>/:

runbooks/nav-monthly/
├── runbook.md              # frontmatter (slug, version, config schema) + the
│                           #   LLM-facing skill body (markdown after the ---)
└── templates/
    ├── workflow.py         # @runbook.defn class, @runbook.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 @runbook.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 ntro.workflow import NtroWorkflow, runbook

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


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

    @runbook.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) ───────────────

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

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

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

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

@runbook.defn and @runbook.step both come from the runbook facade in ntro.workflow (the N-102 runtime-agnostic surface) — not from temporalio directly. defn binds the class to whatever runtime is active (Temporal in the worker, LocalRuntime in tests); going through the facade is exactly what lets the same code run in both. (@runbook.step is the facade alias of @ui_step — either spelling works.)
@runbook.step 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:

HITL approvals — wait_for_action

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.

External signals — await_signal_with_action

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.

Child workflows — run_child_workflow

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.

External agents — ntro.workflow.agents.invoke

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.

UI queries and signals — wired automatically

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.

Determinism rules

Workflow code (the @runbook.defn class — not activities) must be deterministic: Temporal replays it from history to recover state, so the same inputs must always produce the same sequence of commands. Anything that varies per call — the wall clock, randomness, network I/O — breaks replay. This bites silently under LocalRuntime: non-deterministic code runs fine in tests but corrupts Temporal history in production. The N-102 determinism lint flags the common offenders in CI. The rules, and the deterministic alternative for each:

Don’t (in workflow code)	Do instead
`datetime.now()` / `time.time()`	`runbook.now()` — the workflow’s replay-safe clock
`random.*` / `uuid.uuid4()`	Generate it in an activity and pass the value back in
Direct DB / HTTP / file I/O	Wrap the side-effect in an `@activity.defn`; call it via `runbook.execute_activity`
`os.environ` / reading config inline	Thread config through the workflow’s `Context` input

The throughline: side-effects and entropy live in activities; the workflow only orchestrates. Activities run once and their results are recorded in history, so anything non-deterministic is safe there.

Skill definitions

The runbook.md frontmatter + skill body 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, @runbook.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.

Get Started

Tooling

Building workflows

Ntro SDK

API Reference

Deploy & Run

Anatomy of a runbook bundle

The shape of a runbook

What `NtroWorkflow` gives you

Observable results — `ntro.events`

Determinism rules

Skill definitions

ntro.workflow API reference

UI and Temporal signals

Ingest contracts

Register agents

Testing

​Anatomy of a runbook bundle

​The shape of a runbook

​What NtroWorkflow gives you

​Observable results — ntro.events

​Determinism rules

​Related

Skill definitions

ntro.workflow API reference

UI and Temporal signals

Ingest contracts

Register agents

Testing

Anatomy of a runbook bundle

The shape of a runbook

What `NtroWorkflow` gives you

Observable results — `ntro.events`

Determinism rules

Related