Skip to content

python.md

Latest commit

 

History

History
289 lines (207 loc) · 6.49 KB

python.md

File metadata and controls

289 lines (207 loc) · 6.49 KB

Use the Python SDK

Manage Ascend workspaces, deployments, flows, and flow runs from Python.

Install

uv add ascend-tools

Upgrade to the latest version:

uv add --upgrade ascend-tools

See Installation for all install methods.

Authenticate

From instance config (recommended)

from ascend_tools import Client

client = Client()                       # uses default instance from ~/.ascend-tools/config.toml
client = Client(instance="staging")     # uses a named instance

Set up instances with the CLI: ascend-tools instance add (see CLI guide).

From environment variables

client = Client()  # reads ASCEND_SERVICE_ACCOUNT_ID, etc. from env

See Quickstart for the full service account creation walkthrough.

With explicit credentials

client = Client(
    service_account_id="<YOUR_SERVICE_ACCOUNT_ID>",
    service_account_key="<YOUR_SERVICE_ACCOUNT_KEY>",
    instance_api_url="<YOUR_INSTANCE_API_URL>",
)

All parameters are keyword-only. Resolution order: explicit args > instance config > env vars.

Environments and projects

List environments

environments = client.list_environments()

Get an environment by title

env = client.get_environment(title="Production")

Returns dict with the matching environment. Raises an error if not found or ambiguous.

List projects

projects = client.list_projects()

Get a project by title

project = client.get_project(title="My Project")

Returns dict with the matching project. Raises an error if not found or ambiguous.

List profiles

profiles = client.list_profiles(workspace="My Workspace")
profiles = client.list_profiles(deployment="My Deployment")
profiles = client.list_profiles(project="My Project", branch="main")

Returns list[str] of profile names. Provide exactly one of workspace/deployment/uuid, or project+branch.

Manage workspaces and deployments

List and get workspaces

client.list_workspaces()
client.list_workspaces(environment="Production", project="My Project")
client.get_workspace(title="My Workspace")

Create a workspace

ws = client.create_workspace(
    title="My Workspace",
    environment="Production",
    project="My Project",
    profile="default",
    git_branch="main",
)

Optional parameters: git_branch_base, size, storage_size, auto_snooze_timeout_minutes.

Update a workspace

ws = client.update_workspace(
    title="My Workspace",
    git_branch="feature/abc",
)

Only provided fields are changed. Optional parameters: new_title, git_branch, git_branch_base, profile, size, storage_size, auto_snooze_timeout_minutes.

Pause, resume, delete workspaces

client.pause_workspace(title="My Workspace")
client.resume_workspace(title="My Workspace")
client.delete_workspace(title="My Workspace")

List and get deployments

client.list_deployments()
client.list_deployments(environment="Production", project="My Project")
client.get_deployment(title="My Deployment")

Create a deployment

dep = client.create_deployment(
    title="My Deployment",
    environment="Production",
    project="My Project",
    profile="default",
    git_branch="main",
)

Optional parameters: git_branch_base, size, storage_size, enable_automations.

Update a deployment

dep = client.update_deployment(
    title="My Deployment",
    enable_automations=True,
)

Only provided fields are changed. Optional parameters: new_title, git_branch, git_branch_base, profile, size, storage_size, enable_automations.

Pause/resume automations, delete deployments

client.pause_deployment_automations(title="My Deployment")
client.resume_deployment_automations(title="My Deployment")
client.delete_deployment(title="My Deployment")

Manage flows

List flows

flows = client.list_flows(workspace="My Workspace")
flows = client.list_flows(deployment="My Deployment")

Returns list[dict], each with a name field.

Run a flow

result = client.run_flow(flow="sales", workspace="My Workspace")

Resume a paused workspace before running:

result = client.run_flow(
    flow="sales",
    workspace="My Workspace",
    resume=True,
)

Pass a spec dict for advanced options:

result = client.run_flow(
    flow="sales",
    workspace="My Workspace",
    spec={"full_refresh": True},
)
result = client.run_flow(
    flow="sales",
    workspace="My Workspace",
    spec={
        "components": ["transform_orders", "transform_customers"],
        "parameters": {"date": "2025-01-01"},
        "run_tests": False,
    },
    resume=True,
)

Returns dict with event_uuid and event_type.

See CLI guide for the full spec options reference.

Monitor flow runs

List flow runs

result = client.list_flow_runs(workspace="My Workspace")
runs = result["items"]       # list[dict]
truncated = result["truncated"]  # bool

Filter by status, flow name, time range, or paginate:

client.list_flow_runs(workspace="My Workspace", status="running")
client.list_flow_runs(deployment="My Deployment", flow="sales")
client.list_flow_runs(workspace="My Workspace", since="2025-01-01T00:00:00Z")
client.list_flow_runs(workspace="My Workspace", limit=10, offset=20)

Get a flow run

run = client.get_flow_run(name="fr-...", workspace="My Workspace")

Returns dict with fields: name, flow, build_uuid, runtime_uuid, status, created_at, error.

Otto (AI assistant)

# List providers and models
providers = client.list_otto_providers()

# Chat
response = client.otto(prompt="What flows are running?")
response = client.otto(prompt="Describe the sales flow", workspace="My Workspace")

Return types

  • All methods return dict or list[dict]
  • All parameters are keyword-only
  • Type stubs are provided (core.pyi) for IDE autocomplete
  • The package includes a py.typed marker (PEP 561)

Error handling

The SDK raises exceptions for:

  • Missing configuration (environment variables not set)
  • Authentication failures (invalid or expired key)
  • HTTP errors (API returns non-2xx status)
  • State errors (paused, starting, error state)
try:
    client.run_flow(flow="sales", workspace="My Workspace")
except Exception as e:
    print(f"Error: {e}")