Metadata-Version: 2.4
Name: dubsof
Version: 0.4.0
Summary: Python SDK for the Atrium platform by Dubsof
Author-email: Dubsof <hello@dubsof.com>
License: MIT
Keywords: atrium,dubsof,sdk,automation,workflow,api
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Information Technology
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: POSIX
Classifier: Operating System :: MacOS :: MacOS X
Classifier: Operating System :: Microsoft :: Windows
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: Topic :: System :: Networking
Classifier: Framework :: AsyncIO
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: betterproto>=0.9.0
Requires-Dist: websockets>=12.0
Requires-Dist: httpx>=0.27
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: respx>=0.21; extra == "dev"
Dynamic: license-file

# dubsof

Python SDK for [Atrium](https://dubsof.com/) — 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:

```python
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.

```python
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:

```python
# 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.

```python
@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

```python
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).

```python
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.

```python
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.

```python
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:

```python
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

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

Integration tests require a running Atrium server:

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