dubsof

Python SDK for Dubsof Console — the platform for workflow automation, AI composition, service orchestration, and data ingestion.

pip install dubsof

Requires: Python ≥ 3.10, websockets, httpx, betterproto.

Overview

The SDK provides five independently importable services, each mapping to a Dubsof Console product:

Setup

Call initialize_app() once at startup, then import any service directly:

import dubsof
from dubsof import credentials

dubsof.initialize_app(
    credentials.APIKey("atk_..."),
    url="https://console.dubsof.com",
    client_id="my-service",   # identifies this process on the server
)

Connections and shutdown

All services share one connection pool, so a process opens one set of connections rather than one per product and one per call. Release it when you're done:

async with dubsof.initialize_app(credentials.APIKey("atk_..."), url="...") as app:
    await directory.orgs.list()
# pool closed

# or explicitly
app = dubsof.initialize_app(cred, url="...")
try:
    ...
finally:
    await app.aclose()

Long-running services can skip closing entirely; it matters for scripts, tests, and anything that creates apps repeatedly.

An injected http_client is yours to close — aclose() leaves it alone:

import httpx
app = dubsof.initialize_app(cred, url="...", http_client=httpx.AsyncClient(proxy="..."))

Dubsof Run

Register Python functions as tools the workflow engine can call. The connection is a persistent WebSocket; your code runs in your own process with full access to your databases, APIs, and secrets.

import asyncio
from dubsof import run

@run.tool("search_crm", description="Search CRM contacts by name or email")
def search_crm(query: str, tenant_id: str = "default") -> dict:
    return {"results": db.search(query)}

@run.tool("send_email", description="Send a transactional email")
def send_email(to: str, subject: str, body: str) -> dict:
    mailer.send(to, subject, body)
    return {"sent": True}

asyncio.run(run.connect())   # blocks — reconnects automatically

run.connect() registers all decorated tools, then serves tool calls indefinitely. It reconnects with exponential backoff (1 s → 60 s cap) on disconnect.

REST management

All workflow CRUD and event operations are available directly on run — no .http accessor:

# Workflows
wf  = await run.create_workflow("Daily sync", description="Runs every morning")
wf  = await run.generate_workflow("Monitor Slack and post a weather summary daily")
wfs = await run.list_workflows()
wf  = await run.get_workflow(workflow_id)
      await run.update_workflow(workflow_id, name="New name")
      await run.delete_workflow(workflow_id)

# Flows
flow = await run.create_flow(workflow_id, "Fetch and notify", trigger_id=tid, tool_ids=[...])
flows = await run.list_flows(workflow_id)

# Flow dependencies — flow_b waits for flow_a
dep  = await run.add_flow_dep(workflow_id, flow_id=flow_b, depends_on_flow_id=flow_a)
      await run.remove_flow_dep(workflow_id, flow_id=flow_b, depends_on_flow_id=flow_a)
deps = await run.list_flow_deps(workflow_id)

# Tasks
task = await run.create_task(workflow_id, flow_id, "Get weather")

# Triggers and events
trigger  = await run.create_trigger(workflow_id, description="Manual fire")
triggers = await run.list_triggers(workflow_id)
await run.fire_event(trigger.id, payload={"city": "London"})

# Executions
page = await run.list_executions(workflow_id=wf.id, status="running", limit=20)
ex   = await run.get_execution(execution_id)

# Audit log
page  = await run.list_audit(product="run", limit=50)
event = await run.get_audit_event(event_id)

# Tools and rate limits
tools  = await run.list_tools(active_only=True)
limits = await run.get_rate_limit()

Tenant isolation

Declare tenant_id: str = "default" to receive the caller's tenant ID. It is injected by the server and stripped from LLM-generated arguments — the LLM never controls it.

@run.tool("list_invoices", description="List invoices for the current tenant")
def list_invoices(limit: int = 20, tenant_id: str = "default") -> dict:
    return {"invoices": db.invoices.list(tenant=tenant_id, limit=limit)}

Direct client for advanced setups

from dubsof.run import Client

client = Client(client_id="acme", api_key="atk_...", url="https://console.dubsof.com")

@client.tool("search_crm")
def search_crm(query: str) -> dict: ...

asyncio.run(client.connect())

Dubsof Compose

Generate and manage application architectures using TSL (Tinsel System Language).

from dubsof import compose

# Projects
projects = await compose.list_projects()
project  = await compose.create_project("My API", prompt="A REST API for managing orders")
project  = await compose.get_project(project_id)
          await compose.update_project(project_id, name="Order API")
          await compose.delete_project(project_id)

print(project.id, project.name, project.has_output)

# AI generation — streamed in real time
async for event in compose.generate_project(project_id, "A REST API for managing orders"):
    if event.event == "update":
        print(event.data["phase"], event.data["status"])
    elif event.event == "build_done":
        print("Services:", [s["name"] for s in event.data.get("services", [])])
    elif event.event == "error":
        raise RuntimeError(event.data["error"])
    elif event.event == "done":
        break

# Fetch completed project
project = await compose.get_project(project_id)

# Validate and compile
graph  = await compose.parse_project(project_id)    # {"ok", "graph", "types"}
check  = await compose.check_project(project_id)    # CheckResult
result = await compose.compile_project(project_id)  # CompileResult

print("OK" if check.ok else "errors", len(check.diagnostics), "diagnostic(s)")
print("Services:", [s.name for s in result.services])

# Download artefacts
zip_bytes  = await compose.export_project(project_id)    # compiled source ZIP
spec_bytes = await compose.download_openapi(project_id)  # OpenAPI JSON or ZIP

# Models and usage
models = await compose.list_models()
usage  = await compose.get_usage()

Dubsof Runtime

Deploy and manage containerised services produced by Compose.

from dubsof import runtime

# Deploy a Compose project
dep = await runtime.deploy_project(project_id)
print(dep.id, dep.status)   # dep-a1b2c3d4  creating

# Stream until live
async for dep in runtime.stream_deployment(dep.id):
    print(dep.status, dep.urls)
# live  {'hello_world': 'https://hello-world.dubsof.app'}

# List all deployments
deployments = await runtime.list_deployments()
for d in deployments:
    print(d.project.name, d.status, d.urls)
    for svc in d.services:
        print(" ", svc.name, svc.type)

# Get a specific deployment
dep = await runtime.get_deployment(deployment_id)

# Redeploy
new_dep = await runtime.redeploy(dep.id)

Dubsof Ingest

Run data-to-graph extraction pipelines on structured datasets.

from dubsof import ingest

# List available built-in and uploaded datasets
datasets = await ingest.list_datasets()
print(datasets.builtin)    # ["northwind", "chinook"]
print(datasets.uploaded)   # ["my_data"]

# Preview a dataset
preview = await ingest.preview_dataset("northwind")

# Upload your own data
from pathlib import Path
await ingest.upload_dataset("my_data", [Path("customers.csv"), Path("orders.csv")])

# Start a pipeline job
job = await ingest.start_job("northwind", entity_names=["Customer", "Order"])

# Stream results in real time
async for status in ingest.stream_job(job.id):
    if status.partial_result:
        print(status.partial_result.total_entities, "entities so far")

# Get the final result
final = await ingest.get_job(job.id)
print(final.result.total_entities, "entities")
print(final.result.total_edges, "edges")

# Download JSONL output
zip_bytes = await ingest.download_results(job.id)
with open("results.zip", "wb") as f:
    f.write(zip_bytes)

Dubsof Directory

Model your customers' own organizations and users. Every call takes an env_id first — Directory scopes every lookup to one environment.

from dubsof import directory

# TenantOrgs — the top of the hierarchy
org = await directory.create_tenant_org(env_id, external_id="acme", name="Acme Inc")

# Organizational Units — each org gets a default (root) OU named after it
ous  = await directory.list_ous(env_id, org.id)
root = next(o for o in ous if o.is_default)

eng      = await directory.create_ou(env_id, org.id, "Engineering")          # under the root OU
platform = await directory.create_ou(env_id, org.id, "Platform", eng.id)     # under Engineering

await directory.update_ou(env_id, platform.id, org.id, name="Infrastructure")
await directory.update_ou(env_id, platform.id, org.id, parent_ou_id=root.id)  # reparent

# TenantUsers — always placed in exactly one OU
alice = await directory.create_tenant_user(
    env_id, org.id, email="alice@acme.com", name="Alice", ou_id=eng.id,
)
await directory.update_tenant_user(env_id, alice.id, ou_id=platform.id)   # move between OUs

in_eng = await directory.list_tenant_users(env_id, tenant_org_id=org.id, ou_id=eng.id)

Omit ou_id on create_tenant_user and the server places the user in the org's default OU. Omit parent_ou_id on create_ou and the new OU is nested under the default OU — the default OU holds the top of the tree, so nothing else can be created at that level. A user is always in some OU, so ou_id cannot be cleared; move them to the default OU instead.

Custom fields, Integrations, and Composio connections are also on this module — see help(directory).

Pagination

orgs, users, and ous are cursor-paginated. Omitting limit returns the whole collection, exactly as before pagination existed:

users = await directory.list_tenant_users(env_id, tenant_org_id=org.id)
len(users)          # a Page is a list, so this is unchanged

Pass limit to take one page at a time, and resume with next_cursor:

page = await directory.list_tenant_users(env_id, tenant_org_id=org.id, limit=100)
page.next_cursor    # opaque; round-trip it, never parse it
page.has_more

nxt = await directory.list_tenant_users(
    env_id, tenant_org_id=org.id, limit=100, cursor=page.next_cursor,
)

For a collection whose size you don't control, iterate. Memory stays proportional to the page size rather than to the user count, and cursors are followed for you:

async for user in directory.iter_tenant_users(env_id, tenant_org_id=org.id):
    ...

async for ou in directory.iter_ous(env_id, org.id):
    ...

Cursors are opaque strings encoding a position, not offsets, so rows inserted while you page never shift a page boundary and never cause a row to be skipped or repeated.

Typed models

All methods return typed dataclass objects. Import them from the relevant service module:

from dubsof.run     import Workflow, Flow, Task, FlowDep, Trigger, Execution, ExecutionPage, Tool, AuditEvent, AuditPage, FireResult
from dubsof.compose import Project, Model, CheckResult, CompileResult, GenerateEvent, Diagnostic, DeploymentRef
from dubsof.runtime import Deployment, Service, ProjectRef
from dubsof.ingest  import Job, JobResult, JobSummary, Dataset, DatasetList, DatasetPreview
from dubsof.directory import TenantOrg, TenantUser, OrganizationalUnit, FieldDef, Integration, Connection

All models are @dataclass(slots=True) — lightweight, no external dependencies, IDE-friendly.

Error handling

All HTTP methods raise a typed exception on non-2xx responses, so you can branch on the kind of failure rather than inspecting status codes:

from dubsof import errors

try:
    ou = await directory.ous.get(ou_id)
except errors.NotFoundError:
    ...
except errors.RateLimitError as e:
    await asyncio.sleep(e.retry_after or 1)
except errors.AtriumError as e:            # catch-all
    log.error("atrium %s: %s (request %s)", e.status_code, e.detail, e.request_id)

Every exception carries status_code, detail (the server's message), and request_id when the server supplied one. All of them subclass AtriumError, which itself subclasses httpx.HTTPStatusError, so code written against earlier SDK versions keeps working.

429 responses are retried automatically (up to 3×) after the server-specified Retry-After delay, and RateLimitError is raised only once that budget is spent.

Running tests

uv pip install -e ".[dev]"
pytest

Integration tests require a running Dubsof Console server:

ATRIUM_URL=https://console.dubsof.com ATRIUM_API_KEY=atk_... pytest tests/test_integration.py