The Ntropii client

ntro.workspace.Client is how you talk to Ntropii from Python. It wraps the same surface the CLI and MCP server use, organised into resource accessors.

from ntro.workspace import Client

client = Client.from_config()

Construction

from_config
Explicit

# Default connection from ~/.ntro/config.toml
client = Client.from_config()

# Named connection
client = Client.from_config(connection="production")

Resolution order: connection= param > env vars (NTRO_HOST, NTRO_API_KEY) > the connection named here > default_connection_name in the config file.

client = Client(
    host="https://api.ntropii.com/v1",
    api_key="ntro_prod_xxx",
    timeout=30.0,            # Optional; default 30s
)

Explicit args bypass the config file entirely. Use in tests, notebooks where you want zero environmental dependency, or when the credential lives in a secrets manager.

Resource accessors

Each property on Client returns a resource client. All methods come in async + _sync flavours.

Accessor	What it manages
`client.identity`	Current user identity (`whoami`)
`client.integrations`	Data platform configurations
`client.tenants`	Tenants (clients)
`client.entities`	Entities (SPVs / funds) within tenants
`client.runbooks`	Runbooks — list / get / templates / feedback / deploy bundles registered with the worker
`client.workflows`	Workflows — bindings of a runbook to a tenant entity (with optional schedule)
`client.agents`	External agents — registered references to Claude Managed Agents, Copilot agents, etc.
`client.tasks`	Workflow runs (create + poll)

Identity

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

Integrations

# Register a Snowflake data platform (BYO).
# Skip this entirely for tenants on managed-postgres — Ntropii provisions
# the database itself and no integration is needed.
integration = client.integrations.create_data_platform_sync(
    name="Acme Snowflake UK",
    provider="SNOWFLAKE",
    region="UK-South",
    config={
        "account": "acme-fund.eu-west-2",
        "warehouse": "fund_ops",
        "user": "ntropii",
        "password": "...",
    },
)

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

# List + discover
platforms = client.integrations.list_data_platforms_sync()
schemas = client.integrations.discover_schemas_sync(integration.id)

Tenants & entities

# Managed Postgres — no external config needed
tenant = client.tenants.create_sync(
    name="Acme Fund",
    slug="acme-fund",
    dataPlatform="managed-postgres",
)

# BYO — bind to a registered config; provider must match dataPlatform
tenant = client.tenants.create_sync(
    name="Acme Fund",
    slug="acme-fund",
    dataPlatform="snowflake",
    dataPlatformConfigId=integration.id,
)

entity = client.entities.create_sync(
    tenant_id="acme-fund",
    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(tenant_id="acme-fund")

Runbooks

A runbook is the deterministic Python templates that drive a workflow. Runbooks live in runbook-templates/runbooks/<slug>/ and get deployed to the worker. The SDK surfaces them for browsing, fetching the LLM-facing skill definition, and pushing bundles.

# List runbooks currently available on the worker
runbooks = client.runbooks.list_sync()

# Fetch one with its full templates manifest
detail = client.runbooks.get_sync("nav-monthly")

# Pull the templates payload (workflow.py, activities.py, models.py, skill.md)
templates = client.runbooks.templates_sync("nav-monthly")

# Submit user feedback against a runbook (optionally tied to a task)
client.runbooks.feedback_sync("nav-monthly", "Period close took longer than expected", task_id="...")

# Deploy a runbook bundle to the worker (less common — usually via the CLI)
result = client.runbooks.deploy_sync(
    "nav-monthly",
    version="0.1.0",
    artifact_url="...",  # signed storage URL for the bundle archive
)

Workflows

A workflow binds a runbook to an entity, optionally on a schedule. Per N-80, workflows are anchored to a runbookSlug (the deployed code identifier on the worker); a workflow with a friendly name like "Acme NAV monthly" dispatches to the registered nav-monthly worker code.

# Bind a runbook to an entity
workflow = client.workflows.create_sync(
    runbookSlug="nav-monthly",
    entityId="acme-commercial-spv1",
    name="Acme NAV monthly — SPV 1",
    schedule="0 8 5 * *",
    timezone="Europe/London",
)

# Inspect
workflows = client.workflows.list_sync()
detail = client.workflows.get_sync(workflow.id)

Most of the time, you won’t call these directly. The CLI’s ntro workflow create --path ./runbooks/X/ --tenant T --entity E --schedule … wraps runbook deploy + workflow binding in a single command. Use the resource accessors here when scripting or building your own flow.

Agents (external)

External agents are references to Managed Agents on host platforms (Anthropic, GitHub/Microsoft, …) that runbook steps can invoke via ntro.workflow.agents.invoke. The SDK surfaces registration and lifecycle.

# Register a reference to an Anthropic-hosted Managed Agent.
# `external_ref` encodes both the kind and the host-platform id.
agent = client.agents.create_sync(
    tenantId="byng",
    kind="claude_managed",
    external_ref="anthropic://agents/agent_01SxpziMunFrUbnAYkGB5Hu3",
    name="audit-handover",
)
print(agent.id, agent.api_key)  # api_key is plaintext — shown ONCE

# List + get
agents = client.agents.list_sync(tenant_id="byng")
detail = client.agents.get_sync(agent.id)

# Versioning (optional — supports rolling out new agent prompts)
version = client.agents.create_version_sync(agent.id, changelog="prompt v2")

# Delete
client.agents.delete_sync(agent.id)

Agents are tenant-scoped. The same Anthropic agent id can be registered under multiple tenants — each registration becomes a separate Ntropii agent row with its own API key.

See Register agents for the end-to-end lifecycle, ntro.workflow.agents for the invoke API.

Tasks (workflow runs)

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

# Poll
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 for an entity
history = client.tasks.history_sync("acme-fund", "acme-commercial-spv1")

Async usage

Every method is async-first. Use it when you need concurrency or when you’re inside an async context (Temporal activities, FastAPI handlers):

import asyncio
from ntro.workspace import Client

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

    tenants, workflows = await asyncio.gather(
        client.tenants.list(),
        client.workflows.list(),
    )

    await client.close()

asyncio.run(main())

The _sync wrappers are a convenience for scripts and notebooks — they call asyncio.run() internally and will fail if used inside an async function.

Exception types

from ntro.workspace.exceptions import (
    AuthenticationError,
    AuthorizationError,
    NotFoundError,
    ConflictError,
    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 Ntropii: {e}")

Exception	HTTP status	When
`AuthenticationError`	401	Invalid or missing API key
`AuthorizationError`	403	API key lacks permissions for this tenant
`NotFoundError`	404	Resource doesn’t exist
`ConflictError`	409	Resource already exists (e.g. duplicate slug)
`ValidationError`	422	Request payload is invalid
`NtroConnectionError`	—	Network or timeout error

All inherit from a common NtroError base if you want a catch-all.

What’s next

Build runbooks

Concept walkthrough for authoring a runbook from scratch.

ntro.workflow

API reference for NtroWorkflow, @runbook.step, and the orchestration primitives.

Register agents

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

MCP server

Same surface, exposed to coding agents.

Get Started

Tooling

Building workflows

Ntro SDK

API Reference

Deploy & Run

The Ntropii client

Construction

Resource accessors

Identity

Integrations

Tenants & entities

Runbooks

Workflows

Agents (external)

Tasks (workflow runs)

Async usage

Exception types

What’s next

Build runbooks

ntro.workflow

Register agents

MCP server

​Construction

​Resource accessors

​Identity

​Integrations

​Tenants & entities

​Runbooks

​Workflows

​Agents (external)

​Tasks (workflow runs)

​Async usage

​Exception types

​What’s next

Build runbooks

ntro.workflow

Register agents

MCP server

Construction

Resource accessors

Identity

Integrations

Tenants & entities

Runbooks

Workflows

Agents (external)

Tasks (workflow runs)

Async usage

Exception types

What’s next