dubsof 0.4.0

Python SDK for the Atrium platform by Dubsof

dubsof

Python SDK for Atrium — 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 four independently importable services, each mapping to an Atrium product:

Import	Product	What it does
from dubsof import run	Atrium Run	Register tools and connect to the workflow engine
from dubsof import compose	Atrium Compose	Generate and manage application architectures
from dubsof import runtime	Atrium Runtime	Deploy and manage containerised services
from dubsof import ingest	Atrium Ingest	Run data-to-graph extraction pipelines

---

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://admin.atrium.dubsof.com",
    client_id="my-service",   # identifies this process on the server
)

Parameter	Default	Description
cred	required	Credentials object
url	"https://admin.atrium.dubsof.com"	Atrium server URL
client_id	"default"	Stable identifier for this SDK process
verify_ssl	True	Set to False for self-signed certs in development

---

Atrium 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://admin.atrium.dubsof.com")

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

asyncio.run(client.connect())

---

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

---

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

---

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

---

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

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

---

Error handling

All HTTP methods raise httpx.HTTPStatusError on non-2xx responses. 429 responses are retried automatically (up to 3×) after the server-specified Retry-After delay.

---

Running tests

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

Integration tests require a running Atrium server:

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