Metadata-Version: 2.4
Name: dubsof
Version: 0.7.1
Summary: Python SDK for the Dubsof Console
Author-email: Dubsof <hello@dubsof.com>
License: MIT
Keywords: dubsof,console,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 [Dubsof Console](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 five independently importable services, each mapping to a Dubsof Console product:

| Import | Product | What it does |
|--------|---------|-------------|
| `from dubsof import run` | Dubsof Run | Register tools and connect to the workflow engine |
| `from dubsof import compose` | Dubsof Compose | Generate and manage application architectures |
| `from dubsof import runtime` | Dubsof Runtime | Deploy and manage containerised services |
| `from dubsof import ingest` | Dubsof Ingest | Run data-to-graph extraction pipelines |
| `from dubsof import directory` | Dubsof Directory | Manage tenant orgs, OUs, users, and integrations |

---

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

| Parameter | Default | Description |
|-----------|---------|-------------|
| `cred` | required | Credentials object |
| `url` | `"https://console.dubsof.com"` | Dubsof Console server URL |
| `client_id` | `"default"` | Stable identifier for this SDK process |
| `verify_ssl` | `True` | Set to `False` for self-signed certs in development |
| `timeout` | `30.0` | Per-request timeout in seconds |
| `max_retries` | `3` | 429 retry budget before `RateLimitError` |
| `http_client` | `None` | Your own `httpx.AsyncClient` (proxies, connection limits, ASGI transport) |

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

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

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

```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://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).

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

---

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

---

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

---

## Dubsof Directory

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

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

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

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

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

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

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

| Exception | Status | Typical cause |
|-----------|--------|---------------|
| `BadRequestError` | 400 | Refused: deleting a default OU, a `parent_ou_id` from another org |
| `AuthenticationError` | 401 | Key missing, malformed, expired, or revoked |
| `PermissionDeniedError` | 403 | Key valid, lacks the scope for this call |
| `NotFoundError` | 404 | No such resource in this environment |
| `ConflictError` | 409 | Already exists, or conflicts with a concurrent change |
| `ValidationError` | 422 | Body or parameters rejected |
| `RateLimitError` | 429 | Retry budget exhausted; carries `retry_after` |
| `ServerError` | 5xx | Server failed to handle a valid request |

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

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

Integration tests require a running Dubsof Console server:

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