The Ntropii client

from ntro.workspace import Client

client = Client.from_config()

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

# 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)

# 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")

# 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
)

# 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)

# 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)

# 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")

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())

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}")