ntro

Official Python SDK for the ntro platform — fund operations automation for private markets firms (real estate, infrastructure, private credit).

pip install ntro

---

Overview

The ntro SDK is the Python interface to the Workspace API — the control plane that manages tenants, entities, workflows, and task execution for the ntro platform.

┌─────────────────────────────┐
│     ntro-cli / ntro-mcp     │   ← Thin interface layers built on this SDK
└──────────────┬──────────────┘
               │ imports
               ▼
┌─────────────────────────────┐
│      ntro (this package)    │   ← httpx + Pydantic, async-first
│      ntro.workspace.Client  │
└──────────────┬──────────────┘
               │ HTTP/REST
               ▼
┌─────────────────────────────┐
│      Workspace API          │   ← https://api.ntropii.com/v1
└─────────────────────────────┘

---

Quick start

from ntro.workspace import Client

# From config file (~/.ntro/config.toml)
client = Client.from_config()                        # default connection
client = Client.from_config(connection="staging")    # named connection

# Explicit (scripting / testing)
client = Client(
    host="http://localhost:3000/v1",
    api_key="ntro_dev_key",
)

---

Configuration

The SDK reads ~/.ntro/config.toml (or $NTRO_HOME/config.toml):

default_connection_name = "local"

[connections.local]
host = "http://localhost:3000/v1"
api_key = "ntro_dev_key"
default_tenant = "acme-fund-admin"

[connections.production]
host = "https://api.ntropii.com/v1"
api_key = "your-api-key-here"

Resolution priority:

1. Explicit constructor args (host=, api_key=)
2. NTRO_HOST / NTRO_API_KEY environment variables
3. Named connection in config (--connection staging)
4. default_connection_name in config

---

Usage

All resources have async methods and _sync wrappers for CLI / scripting use.

Identity

# Async
profile = await client.identity.whoami()

# Sync
profile = client.identity.whoami_sync()
print(profile.email, profile.orgId)

Data platform integrations

# Register a Databricks integration
integration = client.integrations.create_data_platform_sync(
    name="Acme Databricks UK",
    provider="DATABRICKS",
    region="UK-South",
    config={
        "workspaceUrl": "https://adb-1234567890.12.azuredatabricks.net",
        "catalog": "fund_ops",
        "authType": "service-principal",
        "clientId": "...",
        "clientSecret": "...",
    },
)

# Test connectivity
result = client.integrations.test_connection_sync(integration.id)
print(result.success, result.latencyMs)

# List all
platforms = client.integrations.list_data_platforms_sync()

# Discover schemas
schemas = client.integrations.discover_schemas_sync(integration.id)

Tenants

tenant = client.tenants.create_sync(
    name="Acme Fund Administration",
    slug="acme-fund-admin",
    dataPlatformConfigId=integration.id,
)

tenants = client.tenants.list_sync()
tenant = client.tenants.get_sync("acme-fund-admin")

Entities

entity = client.entities.create_sync(
    tenant_id="acme-fund-admin",
    name="Acme Commercial SPV 1",
    slug="acme-commercial-spv1",
    type="real-estate-spv",
    jurisdiction="Jersey",
    currency="GBP",
    schema_="nav_pipeline_spv1",
)

entities = client.entities.list_sync()
entities = client.entities.list_sync(tenant_id="acme-fund-admin")

Workflows

workflow = client.workflows.create_sync(
    name="nav-monthly",
    description="Monthly NAV pipeline",
    tenantId="acme-fund-admin",
    schedule="0 8 5 * *",
    timezone="Europe/London",
)

workflows = client.workflows.list_sync()
workflow = client.workflows.get_sync("nav-monthly")

Deployments

deployment = client.deployments.create_sync(
    workflowId=workflow.id,
    workflowVersionId=version.id,
    tenantId="acme-fund-admin",
)

status = client.deployments.get_sync(deployment.id)

Tasks (workflow runs)

# Trigger a run
task = client.tasks.create_sync(
    tenantId="acme-fund-admin",
    entityId="acme-commercial-spv1",
    workflowId="nav-monthly",
    context={"period": "2026-03", "priority": "HIGH"},
)

# Poll status
import time
while True:
    task = client.tasks.get_sync(task.id)
    print(task.status, [s.name for s in task.steps if s.status == "IN_PROGRESS"])
    if task.status in ("COMPLETED", "FAILED", "CANCELLED"):
        break
    time.sleep(5)

# History
history = client.tasks.history_sync("acme-fund-admin", "acme-commercial-spv1")

---

Async usage

All methods are async by default, with _sync wrappers that call asyncio.run():

import asyncio
from ntro.workspace import Client

async def main():
    client = Client.from_config()

    # Run multiple requests concurrently
    tenants, workflows = await asyncio.gather(
        client.tenants.list(),
        client.workflows.list(),
    )

    await client.close()

asyncio.run(main())

---

Error handling

from ntro.workspace.exceptions import (
    NotFoundError,
    AuthenticationError,
    ValidationError,
    NtroConnectionError,
)

try:
    tenant = client.tenants.get_sync("unknown-slug")
except NotFoundError:
    print("Tenant not found")
except AuthenticationError:
    print("Invalid API key")
except NtroConnectionError as e:
    print(f"Could not reach API: {e}")

Exception	HTTP status	When
AuthenticationError	401	Invalid or missing API key
AuthorizationError	403	Insufficient permissions
NotFoundError	404	Resource does not exist
ConflictError	409	Resource already exists
ValidationError	422	Request payload invalid
NtroConnectionError	—	Network / timeout error

---

Domain model

Concept	Description
Tenant	A client organisation (fund admin or asset manager). Contains entities.
Entity	An SPV or fund within a tenant. Gets its own schema in the customer's data platform.
Workflow	A repeatable process: NAV calculation, document ingestion, period close.
Task	A running instance of a workflow, scoped to a tenant/entity.
Integration	A data platform connection (Databricks, Snowflake, Microsoft Fabric).

---

Authoring workflows (ntro.workflow)

Runbook authors subclass NtroWorkflow and decorate per-phase methods with @ui_step. The decorator both attaches metadata for the UI breadcrumb AND wraps the method so step lifecycle (_current_step_id, _steps_completed) tracks automatically.

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

from .activities import open_period, emit_period_summary
from .models import NavMonthlyContext, PeriodSummary

@workflow.defn
class NavMonthlyWorkflow(NtroWorkflow):

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

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

    @workflow.run
    async def run(self, ctx: NavMonthlyContext) -> PeriodSummary:
        await self._step_period_open(ctx)
        ...
        return await self._step_summary(ctx, ...)

Class-definition order = breadcrumb order (Python preserves __dict__ insertion order). Icons are Lucide names rendered by ui-tenant.

NtroWorkflow also provides:

• await self.wait_for_action(payload, display_hint, reason) — block on a HITL approve/reject signal; surfaces the rich display_hint to ui-tenant.
• await self.await_signal_with_action(predicate, action, display_hint) — block until an external signal satisfies predicate, advertising what's needed via the current_pending_action query.
• await self.run_child_workflow(slug=..., input=..., step_id=...) — dispatch a child runbook by slug.
• Standard queries current_pending_action, current_steps, current_ui_state and the user_action signal handler — wired automatically.

Install with the workflow extra: pip install 'ntro[workflow]'.

---

Testing workflows (ntro.testing)

Inner-loop test harness for runbook authors. Boots an in-memory Temporal (temporalio.testing.WorkflowEnvironment), registers your workflow + child workflows with auto-mocked activities, and drives the agent loop per scenario. Sub-second startup, no docker, no deploy cycle.

import asyncio
from ntro.testing import WorkflowHarness, HAPPY, REJECT_ALL, load_runbook, report

async def main():
    nav, _ = load_runbook("./runbooks/nav-monthly")
    di,  _ = load_runbook("./runbooks/document-ingest")
    nj,  _ = load_runbook("./runbooks/nav-monthly-journals")

    results = []
    for scenario in [HAPPY, REJECT_ALL]:
        async with WorkflowHarness(nav, child_workflows=[di, nj]) as h:
            results.append(await h.run(input=ctx, scenario=scenario))
    print(report.human(results))

asyncio.run(main())

Output:

✓  happy        (0.86s)
    [ 0.13s] submit_file  hly-7a820232  signal=tb_submitted, source=xero-trial-balance
    [ 0.24s] drill_down   hly-7a820232  children=[...:document-ingest]
    [ 0.36s] review       ument-ingest  response=approved
    ...
✓  reject_all   (0.45s)
summary: 2 passed, 0 failed (of 2)

What's auto-mocked

• Activity returns — derived from each @activity.defn's return type via Pydantic introspection. Required fields get type-conformant fakes (str → "auto-mock", Decimal → 0, nested BaseModel → recursive). Defaulted fields are left alone — the runbook author's defaults are usually the most realistic value.
• HITL responses — controlled by the Scenario (HAPPY approves everything; REJECT_ALL rejects everything). Per-step overrides via Scenario(review_overrides={"step_id": "rejected"}).
• submit_file signals — the harness sends a fake document_ref + tenant_slug/entity_slug derived from the workflow's advertised args.

Built-in scenarios

Name	Behaviour
HAPPY	Auto-approve every review; fakes for everything.
REJECT_ALL	Auto-reject every review (verifies clean termination).

Build custom scenarios by instantiating Scenario(name="...", approve_reviews=False, review_overrides={...}, corrections={...}).

CLI

# Single workflow (no children)
ntro workflow test ./runbooks/document-ingest

# Parent + children
ntro workflow test ./runbooks/nav-monthly \
    --child ./runbooks/document-ingest \
    --child ./runbooks/nav-monthly-journals

# Specific scenarios + JSON output for CI
ntro workflow test ./runbooks/nav-monthly --scenario happy --json

Install with the testing extra: pip install 'ntro[testing]'.

---

Related packages

Package	PyPI	Description
ntro	pypi.org/project/ntro	This SDK
ntro-cli	pypi.org/project/ntro-cli	CLI — ntro tenant list, ntro workflow run, ntro workflow test
ntro-mcp	pypi.org/project/ntro-mcp	MCP server for Claude Desktop / claude.ai